Position

Ctx<Position> binds the delivery's place in the file; seeking back to it redelivers exactly that message.

Position is a context key of the stream file form. Ctx(at): Ctx<Position> binds a FilePosition, and for a delivery it is always FilePosition::Sequence(n): the message's number under its key, counted from one. The value is pinned: a seek to it delivers that same message again, then the rest of the log in order. Stored, it is the checkpoint a later run hands to start_at. The same number travels in the delivery's stream-sequence header (SEQUENCE_HEADER), which each element of a batch carries too. In the manual form the body names FileContext as its context, impl Handle<Order, Reply, (), FileContext>, and reads ctx.context(Position).

Example

Each invoice records the sequence of the order it was made from.

Macros

examples/file_position_key.rs

/// Records each invoice with the sequence of the order it came from, so the next run knows
/// where to resume.
#[subscriber(FileStream::new("orders"), reply)]
async fn invoice(order: &Order, Ctx(at): Ctx<Position>) -> Invoiced {
    tracing::info!(order.id, ?at, "invoiced");
    let sequence = match at {
        FilePosition::Sequence(sequence) => sequence,
        // A delivery always sits at a sequence; the other forms name places to start or seek to.
        FilePosition::Beginning | FilePosition::End | FilePosition::Timestamp(_) => 0,
    };
    Invoiced {
        order: order.id,
        sequence,
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("invoicing", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/shop/orders.ss"),
        |b| {
            b.include(invoice);
        },
    )
}

Manual

examples/manual/file_position_key.rs

/// Records each invoice with the sequence of the order it came from, so the next run knows
/// where to resume.
struct Invoice;

impl Handle<Order, Invoiced, (), FileContext> for Invoice {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, FileContext>,
    ) -> impl Future<Output = Result<Invoiced, HandlerOutcome>> {
        let at = ctx.context(Position);
        tracing::info!(order.id, ?at, "invoiced");
        let sequence = match at {
            FilePosition::Sequence(sequence) => sequence,
            // A delivery always sits at a sequence; the other forms name places to start or
            // seek to.
            FilePosition::Beginning | FilePosition::End | FilePosition::Timestamp(_) => 0,
        };
        ready(Ok(Invoiced {
            order: order.id,
            sequence,
        }))
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("invoicing", "0.1.0")).with_broker(
        FileBroker::new("/var/lib/shop/orders.ss"),
        |b| {
            b.include(
                subscriber(FileStream::new("orders"), Invoice)
                    .reply()
                    .build(),
            );
        },
    )
}