start_at

start_at opens a subscription at a position on every startup; a stored sequence is how a service resumes on a stream file.

It is a clause of the attribute, start_at(FilePosition::beginning()), or .start_at(position) among the mount-site settings. The mount-site form takes a position the service learns only when it starts, such as a stored checkpoint. A subscription with no start_at opens at the tip. The file keeps no record of how far a service read, because ack and nack report Unsupported on this transport (Acknowledgement). So a service that resumes where it stopped stores the sequence it reached, read with Ctx<Position>, and opens the next run at the first sequence it did not handle. FilePosition::sequence(n) delivers message n itself, so the stored sequence of the last handled message is the one to add one to. The sequence the file writes next, one after its last message, opens the subscription at the tip, so a service that handled every message resumes there. A sequence further ahead stops the service at startup. The position applies on every startup: a subscription that starts at the beginning handles the whole retained file each time the service starts. The guide Resuming and seeking builds a service that stores its sequence and resumes there.

Example

The previous run invoiced the orders up to sequence 3, so this run opens at sequence 4.

Macros
Manual

examples/file_start_at.rs
/// Invoices each order once, across restarts.
#[subscriber(FileStream::new("orders"))]
async fn invoice(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "invoiced");
    HandlerOutcome::ack()
}

/// The service, opened at the first sequence the previous run did not invoice.
fn app(file: FileBroker, resume_at: u64) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("invoicing", "0.1.0")).with_broker(file, |b| {
        b.include(invoice.start_at(FilePosition::sequence(resume_at)));
    })
}

examples/manual/file_start_at.rs
/// Invoices each order once, across restarts.
struct Invoice;

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

/// The service, opened at the first sequence the previous run did not invoice.
fn app(file: FileBroker, resume_at: u64) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("invoicing", "0.1.0")).with_broker(file, |b| {
        b.include(
            subscriber(FileStream::new("orders"), Invoice)
                .start_at(FilePosition::sequence(resume_at))
                .build(),
        );
    })
}

Example​

Example