ConsumerGroup

Ctx<keys::ConsumerGroup> binds the consumer group a stream subscription reads through, on a single delivery or a batch.

The group is fixed when the subscription opens: the descriptor's .group(..), or the broker's default_group for a subscription that names none. A handler that reads it learns the group at run time, so one body serves whichever group it is mounted under. As a Ctx parameter the key yields an owned String. Read off a context with ctx.context(keys::ConsumerGroup), it yields a &str borrowed from the context. A batch body reads it that way off StreamBatchContext, since the group belongs to the whole subscription.

Example

The subscription names only the stream; the group comes from the broker's default, and each invoice records it.

Macros
Manual

examples/redis_context_consumer_group.rs
#[subscriber("orders", reply)]
async fn bill(order: &Order, Ctx(group): Ctx<keys::ConsumerGroup>) -> Invoice {
    tracing::info!(order.id, %group, "order billed");
    Invoice {
        order_id: order.id,
        billed_by: group,
    }
}

#[ruststream::app]
fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379").default_group("billing");
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        b.include(bill).out_reply(Publish);
    })
}

examples/manual/redis_context_consumer_group.rs
struct Bill;

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

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379").default_group("billing");
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("orders", Bill).reply().build())
            .out_reply(Publish);
    })
}

Example​

Example