FilePosition

FilePosition names a place in a stream key's log: the beginning, the tip, a sequence number, or an instant.

A stream file keeps its log, so a subscription opens at a place in it with start_at, and a handler moves it with a seek. FilePosition::beginning() is the first message the file retains under the key. FilePosition::end() is the tip, where the next append lands. FilePosition::sequence(n) is message number n of the key, and the message itself is delivered again. A stream key is numbered from one, and the numbering carries on when a later run appends to the same file. FilePosition::timestamp(millis) is the first message stamped strictly later than that instant, in milliseconds since the Unix epoch. A delivery reports its own place as a sequence, which Ctx<Position> binds. The sequence the file writes next, one after its last message, is the tip. A sequence further ahead, and an instant later than every message the file holds, are refused. A position read off a delivery, the beginning and the tip are always within the file.

Example

The revenue report is rebuilt from the first order the file retains on every startup, then follows the orders appended after.

Macros
Manual

examples/file_position.rs
/// Rebuilds the revenue report from the first order the file retains, then follows new orders.
#[subscriber(FileStream::new("orders"), start_at(FilePosition::beginning()))]
async fn report(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.total_cents, "counted");
    HandlerOutcome::ack()
}

/// The report on the shop's stream file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("revenue", "0.1.0")).with_broker(file, |b| {
        b.include(report);
    })
}

examples/manual/file_position.rs
/// Rebuilds the revenue report from the first order the file retains, then follows new orders.
struct Report;

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

/// The report on the shop's stream file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("revenue", "0.1.0")).with_broker(file, |b| {
        b.include(
            subscriber(FileStream::new("orders"), Report)
                .start_at(FilePosition::beginning())
                .build(),
        );
    })
}

Example​

Example