Channels

A stream channel reports its group, consumer, read mode and idle threshold; a list its acknowledgement; a channel its mode.

Each channel a service reads carries form, the Redis structure behind it: stream, list or pubsub. The rest of the extension is the descriptor's own settings.

Form	Fields
stream	group, consumer, readMode (fresh, reclaim or claiming), and minIdleMs for the two modes that claim
list	reliable, processing (the processing list of a reliable list), envelope
pubsub	mode (classic or sharded), pattern, envelope

A RedisPubSubPattern channel reports pattern: true, so a reader knows the address is a glob and not a channel to publish to. envelope.contentType is the framing a list entry or a channel message carries its headers in. A stream entry holds its headers as fields, so a stream has no envelope. A setting the descriptor leaves to the broker, such as a consumer name it did not set, stays out.

Example

The channel orders reports a claiming read in the group workers, as the consumer worker-1, from thirty seconds of idle time. The channel jobs reports a reliable list, and events a sharded channel.

Macros

examples/redis_asyncapi.rs

#[subscriber(
    RedisStream::claiming("orders", Duration::from_secs(30))
        .group("workers")
        .consumer("worker-1"),
    reply("orders.done")
)]
async fn confirm(order: &Order) -> Receipt {
    Receipt { id: order.id }
}

#[subscriber(RedisList::new("jobs").reliable(), reply("jobs.done"))]
async fn run_job(order: &Order) -> Receipt {
    Receipt { id: order.id }
}

#[subscriber(RedisPubSub::new("events").mode(PubSubMode::Sharded))]
async fn on_event(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "event seen");
    HandlerOutcome::ack()
}

#[subscriber(RedisPubSubPattern::new("audit.*"))]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "audited");
    HandlerOutcome::ack()
}

Manual

examples/manual/redis_asyncapi.rs

/// A claiming read in the group `workers`, as the consumer `worker-1`.
fn orders() -> RedisStream {
    RedisStream::claiming("orders", Duration::from_secs(30))
        .group("workers")
        .consumer("worker-1")
}

fn jobs() -> RedisList {
    RedisList::new("jobs").reliable()
}

fn events() -> RedisPubSub {
    RedisPubSub::new("events").mode(PubSubMode::Sharded)
}

fn audit() -> RedisPubSubPattern {
    RedisPubSubPattern::new("audit.*")
}