JetStreamContext

JetStreamContext gives a handler the delivery's stream, consumer, sequences, delivery count and pending count, each read through Ctx<K>.

The subscription builds it once per delivery, beside the payload and the headers. The server writes this metadata on each JetStream delivery; a Core NATS delivery has none. A handler binds one field with Ctx<K>, where K is a key of context::keys:

Key	Value	What it is
`Stream`	`Option<String>`	the stream the message is stored in
`Consumer`	`Option<String>`	the consumer that delivered it, the durable name or the server's name for an ephemeral one
`StreamSequence`	`Option<u64>`	the message's sequence in its stream
`ConsumerSequence`	`Option<u64>`	the delivery's sequence in its consumer
`Delivered`	`Option<i64>`	how many times the server has delivered the message, 1 on the first delivery
`Pending`	`Option<u64>`	how many messages the consumer still holds behind this one

Every key reads None on a Core NATS delivery, so one handler works on both kinds of subscription. The keys are imported from context::keys. A body that reads several fields may take ctx: &mut Context<'_, JetStreamContext> instead and call ctx.context(key) per field, as the manual form does. There each key is also a constant, such as keys::STREAM_SEQUENCE, and a name reads as Option<&str>. The stream sequence is what a publish states as its expected last sequence, and Delivered tells a redelivery from a first delivery.

Example

Each order is written to a ledger with where it sits in the stream ORDERS.

Macros
Manual

examples/nats_context.rs
/// Each `Ctx<K>` binds one field of the delivery's `JetStream` metadata.
#[subscriber(JetStreamSubject::new("orders.created", "ORDERS").durable("ledger"), reply)]
async fn record(
    order: &Order,
    Ctx(stream): Ctx<Stream>,
    Ctx(consumer): Ctx<Consumer>,
    Ctx(stream_sequence): Ctx<StreamSequence>,
    Ctx(consumer_sequence): Ctx<ConsumerSequence>,
    Ctx(delivered): Ctx<Delivered>,
    Ctx(pending): Ctx<Pending>,
) -> LedgerEntry {
    tracing::info!(order.id, ?stream_sequence, ?delivered, "order recorded");
    LedgerEntry {
        order_id: order.id,
        stream,
        consumer,
        stream_sequence,
        consumer_sequence,
        delivered,
        pending,
    }
}

examples/manual/nats_context.rs
/// The body reads each field of the delivery's `JetStream` metadata off its context by its key.
struct Record;

impl Handle<Order, LedgerEntry, (), JetStreamContext> for Record {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, JetStreamContext>,
    ) -> impl Future<Output = Result<LedgerEntry, HandlerOutcome>> {
        let stream_sequence = ctx.context(keys::STREAM_SEQUENCE);
        let delivered = ctx.context(keys::DELIVERED);
        tracing::info!(order.id, ?stream_sequence, ?delivered, "order recorded");
        ready(Ok(LedgerEntry {
            order_id: order.id,
            stream: ctx.context(keys::STREAM).map(str::to_owned),
            consumer: ctx.context(keys::CONSUMER).map(str::to_owned),
            stream_sequence,
            consumer_sequence: ctx.context(keys::CONSUMER_SEQUENCE),
            delivered,
            pending: ctx.context(keys::PENDING),
        }))
    }
}

fn ledger() -> JetStreamSubject {
    JetStreamSubject::new("orders.created", "ORDERS").durable("ledger")
}

Example​

Example