start_at

start_at opens a replayable subscription at a position: the start of the log, a sequence number, or the tip.

A broker over a replayable log owns its position type; for the in-memory broker it is MemoryPosition. start_at(position) seeks the subscription before its first delivery, on every startup. It is one of the mount-site settings, and the attribute takes it as a clause too. Without it, the subscription opens at the broker's default; the in-memory broker's default is the tip of the log, so earlier entries are not delivered. MemoryPosition::start() is the oldest entry the broker still keeps, sequence(n) the n-th entry published to the name, counted from zero, and end() the tip. The in-memory broker keeps a log only when it is built with MemoryBroker::retaining. start_at does not compile against MemoryBroker::new(), which keeps nothing. A broker's subscription accepts a position when it implements Seekable.

Example

The subscription opens at the start of the log and replays what was published before the service started.

Macros
Manual

examples/seek.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "audit")]
struct Entry {
    id: u64,
}

/// The audit trail opens at the start of the log, so it also reads what came before it.
#[subscriber("audit", start_at(MemoryPosition::start()))]
async fn record(entry: &Entry) -> HandlerOutcome {
    tracing::info!(entry.id, "recorded");
    HandlerOutcome::ack()
}

/// Replaying reads what the broker kept, so this one keeps the last 64 entries of every name.
/// `MemoryBroker::new()` keeps nothing, and a mount that seeks does not compile on it.
fn broker() -> MemoryBroker<Retaining> {
    MemoryBroker::retaining(Retention::Messages(nonzero!(64)))
}

examples/manual/seek.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Entry {
    id: u64,
}

// `#[outgoing(name = "audit")]` by hand: the type fixes its destination.
impl OutgoingDestination for Entry {
    type Form = FixedName;

    const DESTINATION: &'static str = "audit";
}

impl MessageHeaders for Entry {
    type Contract = NoHeaders;
}

struct Record;

impl Handle<Entry> for Record {
    fn handle(
        &self,
        entry: &Entry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(entry.id, "recorded");
        ready(Ok(()))
    }
}

/// Replaying reads what the broker kept, so this one keeps the last 64 entries of every name.
/// `MemoryBroker::new()` keeps nothing, and a mount that seeks does not compile on it.
fn broker() -> MemoryBroker<Retaining> {
    MemoryBroker::retaining(Retention::Messages(nonzero!(64)))
}

fn app(broker: MemoryBroker<Retaining>) -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        // The audit trail opens at the start of the log, so it also reads what came before it.
        b.include(
            subscriber("audit", Record)
                .start_at(MemoryPosition::start())
                .build(),
        );
    })
}

Example​

Example