default_group

default_group names the consumer group a bare #[subscriber("key")] reads through; without it such a subscription refuses to start.

A bare name subscribes to the stream under that key and reads its fresh entries. Redis reads a stream only through a consumer group, and a bare name names none. The broker's default group supplies it, for every bare name of the broker. A subscription created at startup creates its group too, so the stream and the group need not exist beforehand.

Without a default group the service does not start. The error is RedisError::InvalidOptions, and its message names the key and the setting. A RedisStream descriptor names its group itself, and the default group does not apply to it.

Example

The handler reads the stream orders by its bare name, in the group workers.

Macros
Manual

examples/redis_broker_default_group.rs
#[subscriber("orders")]
async fn take(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order taken");
    HandlerOutcome::ack()
}

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(take);
    })
}

/// Every bare-name subscription of this broker reads in the group `workers`.
#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(RedisBroker::standalone("redis://localhost:6379").default_group("workers"))
}

examples/manual/redis_broker_default_group.rs
struct Take;

impl Handle<Order> for Take {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order taken");
        ready(Ok(()))
    }
}

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("orders", Take).build());
    })
}

/// Every bare-name subscription of this broker reads in the group `workers`.
fn app() -> impl App<State = ()> {
    service(RedisBroker::standalone("redis://localhost:6379").default_group("workers"))
}

Example​

Example