Subscriptions

A key in #[subscriber] reads that stream through the broker's default group; a descriptor names the form and its settings.

#[subscriber("orders")] reads the Redis stream orders. A stream is read only through a consumer group, and a bare key names none, so the group comes from RedisBroker::default_group. A bare key on a broker with no default group stops the service at startup, and the error names the setting and the key. Every other subscription is a descriptor, written where the key goes, as the attribute's first argument. One descriptor is one form of delivery and carries all of that form's settings:

Form	Descriptor	What one subscription reads
Stream	`RedisStream`	entries of a stream, through a consumer group
List	`RedisList`	elements popped off a list, each by one consumer
Channel	`RedisPubSub`	messages published to a channel while it listens
Pattern	`RedisPubSubPattern`	messages of every channel a glob matches

The manual form passes the key, or the descriptor itself, to subscriber(source, body). Each subscription reads on a connection of its own, beside the pool that publishes and settles. The settings of the mount chain apply to every form: the batch size, the dispatch lanes, the retry cap, the publishers. block is the one Redis setting among them.

Example

The stream orders, read in the group workers.

Macros
Manual

examples/redis_subscriptions_name.rs
/// Reads the stream `orders`; the consumer group comes from the broker.
#[subscriber("orders")]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order placed");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379").default_group("workers");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(place);
    })
}

examples/manual/redis_subscriptions_name.rs
/// Reads the stream `orders`; the consumer group comes from the broker.
struct Place;

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

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

Example​

Example