StreamContext

StreamContext gives a stream handler the entry id, the position that redelivers it, the consumer group and the group's seek handle.

The subscription builds it once per delivery, beside the payload and the headers. A handler names it as its context type, ctx: &mut Context<'_, StreamContext>, and reads a field with ctx.context(key). A handler that needs one field binds it as a parameter with Ctx<K> instead. Each key is a type in context::keys, and the stream prelude brings the module in as keys:

Key	Value	What it is
`EntryId`	`EntryId`	the `<milliseconds>-<sequence>` id the entry was read at; it orders the way the stream does
`Position`	`RedisGroupPosition`	the group cursor that delivers this entry again
`ConsumerGroup`	`&str`	the group the subscription reads through, as ConsumerGroup shows
`SeekHandle`	`RedisGroupSeeker`	the handle that moves the whole group to another position
`FredPool`	`fred::clients::Pool`	the broker's connection pool, as FredPool shows

Each key is a zero-sized type, and reading it is a plain field access the compiler resolves: no hashing, no boxing, no downcast. A key belongs to the context types that carry its field, so a key read off the wrong form's context does not compile. A batch handler reads StreamBatchContext, a Pub/Sub handler PubSubContext. A list delivery carries its payload and headers alone; a list handler reads the pool through FredPool or keeps the default () context. The two claiming read modes report a delivery's count and idle time as the DELIVERY_COUNT_HEADER and IDLE_MS_HEADER headers, which every transport reads the same way. A context costs a handler that does not name one nothing: the default () context builds no value.

Example

The handler writes where each order sat in the stream, and which group read it, to an order trail.

Macros
Manual

examples/redis_context_stream.rs
/// Reads the entry id and the consumer group off the delivery's context.
#[subscriber(RedisStream::new("orders").group("audit"), reply)]
async fn trace(order: &Order, ctx: &mut Context<'_, StreamContext>) -> Trail {
    let entry_id = ctx.context(keys::EntryId);
    let group = ctx.context(keys::ConsumerGroup);
    tracing::info!(order.id, %entry_id, group, "order traced");
    Trail {
        order_id: order.id,
        entry_id: entry_id.to_string(),
        group: group.to_owned(),
    }
}

examples/manual/redis_context_stream.rs
/// Reads the entry id and the consumer group off the delivery's context.
struct Trace;

impl Handle<Order, Trail, (), StreamContext> for Trace {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, StreamContext>,
    ) -> impl Future<Output = Result<Trail, HandlerOutcome>> {
        let entry_id = ctx.context(keys::EntryId);
        let group = ctx.context(keys::ConsumerGroup);
        tracing::info!(order.id, %entry_id, group, "order traced");
        ready(Ok(Trail {
            order_id: order.id,
            entry_id: entry_id.to_string(),
            group: group.to_owned(),
        }))
    }
}

Example​

Example