RedisStream

RedisStream::new reads fresh entries off a stream through a consumer group, and each ack is an XACK.

The read is XREADGROUP with the id >: entries no consumer of the group has been handed yet. group names the group, and a descriptor without one stops the service at startup. The subscription creates its group when it opens, with XGROUP CREATE ... MKSTREAM, so neither the group nor the stream has to exist first. Where a new group starts is its StreamStart. Every group reads every entry of the stream. The consumers of one group share its entries: each entry goes to one of them. consumer names this subscription's consumer; without it each subscription gets a unique generated name. An entry is handed out once and waits in the group's pending entries list until its handler settles it. An ack is XACK, which takes it off that list. An entry whose consumer died before settling stays there, and a reclaim or a claiming subscription takes it back. That makes a stream the recommended form for durable work: the server itself keeps what each consumer holds and how often it was delivered. A reliable list recovers the same work only through a recovery set that the service sweeps. The three constructors are the three read modes, because each returns a different set of entries. A stream keeps its entries until it is trimmed, so a group can be moved back over them or forward past them.

Example

Billing and shipping read every order, each in its own group.

Macros
Manual

examples/redis_streams_new.rs
/// Bills every order, in the group `billing`.
#[subscriber(RedisStream::new("orders").group("billing"))]
async fn bill(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order billed");
    HandlerOutcome::ack()
}

/// Ships every order, in the group `shipping`: a second group reads the same entries again.
#[subscriber(RedisStream::new("orders").group("shipping").consumer("shipping-1"))]
async fn ship(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order shipped");
    HandlerOutcome::ack()
}

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

examples/manual/redis_streams_new.rs
/// Bills every order.
struct Bill;

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

/// Ships every order.
struct Ship;

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

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let billing = RedisStream::new("orders").group("billing");
        b.include(subscriber(billing, Bill).build());
        // A second group reads the same entries again.
        let shipping = RedisStream::new("orders")
            .group("shipping")
            .consumer("shipping-1");
        b.include(subscriber(shipping, Ship).build());
    })
}

Example​

Example