SEQUENCE_HEADER

Every delivery carries its sequence number in the stream-sequence header, from one on a stream file and from zero on a pipe.

SEQUENCE_HEADER is the header's name, stream-sequence. The value is the number in decimal, as each client counts. A stream file numbers each stream key from one and keeps counting across restarts, so the number names the message's place in the file. A pipe numbers the lines of each key from zero, unless a line's header names a sequence of its own. The reader adds the header to every delivery, so the handler reads it with ctx.headers() like any other header. On a stream file Ctx<Position> gives the same place as a FilePosition, the form start_at and a seek take.

Example

The service numbers each order with the sequence it sits at in the file.

Macros
Manual

examples/file_sequence_header.rs
/// Numbers each order with the sequence it sits at in the file.
#[subscriber(FileStream::new("orders"), reply)]
async fn number(order: &Order, ctx: &mut Context<'_>) -> Result<Numbered, HandlerOutcome> {
    let Some(sequence) = ctx
        .headers()
        .get_str(SEQUENCE_HEADER)
        .and_then(|sequence| sequence.parse().ok())
    else {
        return Err(HandlerOutcome::drop());
    };
    tracing::info!(order.id, sequence, "numbered");
    Ok(Numbered {
        order: order.id,
        sequence,
    })
}

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

examples/manual/file_sequence_header.rs
/// Numbers each order with the sequence it sits at in the file.
struct Number;

impl Handle<Order, Numbered> for Number {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Numbered, HandlerOutcome>> {
        let Some(sequence) = ctx
            .headers()
            .get_str(SEQUENCE_HEADER)
            .and_then(|sequence| sequence.parse().ok())
        else {
            return ready(Err(HandlerOutcome::drop()));
        };
        tracing::info!(order.id, sequence, "numbered");
        ready(Ok(Numbered {
            order: order.id,
            sequence,
        }))
    }
}

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

Example​

Example