KinesisContext

Context<'_, KinesisContext> gives a handler the record's position and the seek handle, through the Position and SeekHandle keys.

The subscription builds a KinesisContext once per delivery, beside the payload and the headers. Building it costs three reference-count increments: the shard id and the sequence number are shared with the delivery, and the seeker is the subscription's own. A handler binds one field with Ctx<K>, where K is one of the two keys:

Key	Value	What it is
`Position`	`KinesisPosition`	the record's own place: its shard and its sequence number, as `KinesisPosition::Sequence`
`SeekHandle`	`KinesisSeeker`	the handle that moves the subscription

Both keys are in the prelude. A body that reads both takes ctx: &mut Context<'_, KinesisContext> instead and calls ctx.context(key) per field, as the manual form does. A key the context does not carry is a compile error, not a lookup that finds nothing. A batch handler has a context of its own, with the seek handle only.

Example

The handler writes where each order sits in the stream to an audit trail.

Macros
Manual

examples/kinesis_context.rs
/// Writes where each order sits in the stream to the audit trail.
#[subscriber(KinesisStream::new("orders"))]
async fn audit(
    order: &Order,
    Ctx(at): Ctx<Position>,
    Out(trail): Out<impl Publisher, Trail>,
) -> HandlerOutcome {
    // A delivered record's position is always one record of one shard.
    let KinesisPosition::Sequence { shard, sequence } = at else {
        return HandlerOutcome::drop();
    };
    tracing::info!(order.id, shard, sequence, "order located");
    let located = Located {
        order: order.id,
        shard,
        sequence,
    };
    if trail.message(&located).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/kinesis_context.rs
/// Writes where each order sits in the stream to the audit trail.
struct Audit;

impl<Entry> Handle<Order, (), Outs<(Entry,)>, KinesisContext> for Audit
where
    Entry: OutEntry<Trail, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        ctx: &mut Context<'_, KinesisContext>,
    ) -> Result<(), HandlerOutcome> {
        // A delivered record's position is always one record of one shard.
        let KinesisPosition::Sequence { shard, sequence } = ctx.context(Position) else {
            return Err(HandlerOutcome::drop());
        };
        tracing::info!(order.id, shard, sequence, "order located");
        let located = Located {
            order: order.id,
            shard,
            sequence,
        };
        outs.get(Trail)
            .message(&located)
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

Example​

Example