StreamBatchContext

A stream batch handler's context, StreamBatchContext, carries only the consumer group and the seek handle, which belong to the whole subscription.

A batch spans many entries, so a field that differs per entry has no single value for it. The body reads the batch's fields with ctx.context(key): keys::ConsumerGroup, keys::SeekHandle and keys::FredPool. A Ctx<K> parameter binds the per-delivery context, so a batch body reads its fields off the context instead. The handler names the context in its signature, ctx: &mut Context<'_, StreamBatchContext>, beside a slice payload. The batch itself is the read's COUNT: .batch(n) at the mount site asks Redis for up to n entries per XREADGROUP.

Example

The batch is billed in one call, and every invoice records the group, read once off the batch's context.

Macros
Manual

examples/redis_context_batch.rs
/// Bills a batch of orders; the group is read once, off the batch's context.
#[subscriber(RedisStream::new("orders").group("billing"), reply)]
async fn bill_all(
    orders: &[Order],
    ctx: &mut Context<'_, StreamBatchContext>,
) -> Result<Vec<Invoice>, HandlerOutcome> {
    let group = ctx.context(keys::ConsumerGroup);
    tracing::info!(orders = orders.len(), group, "batch billed");
    Ok(orders
        .iter()
        .map(|order| Invoice {
            order_id: order.id,
            billed_by: group.to_owned(),
        })
        .collect())
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(bill_all.batch(nonzero!(32))).out_reply(Publish);
        },
    )
}

examples/manual/redis_context_batch.rs
/// Bills a batch of orders; the group is read once, off the batch's context.
struct BillAll;

impl Handle<[Order], Vec<Invoice>, (), StreamBatchContext> for BillAll {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        ctx: &mut Context<'_, StreamBatchContext>,
    ) -> impl Future<Output = Result<Vec<Invoice>, Vec<HandlerOutcome>>> {
        let group = ctx.context(keys::ConsumerGroup);
        tracing::info!(orders = orders.len(), group, "batch billed");
        ready(Ok(orders
            .iter()
            .map(|order| Invoice {
                order_id: order.id,
                billed_by: group.to_owned(),
            })
            .collect()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            let bill_all = subscriber(RedisStream::new("orders").group("billing"), BillAll);
            b.include(bill_all.batch(nonzero!(32)).reply().build())
                .out_reply(Publish);
        },
    )
}

Example​

Example