PubSubMode

PubSubMode picks classic SUBSCRIBE and PUBLISH, broadcast to every node, or sharded SSUBSCRIBE and SPUBLISH, local to a slot.

PubSubMode::Classic is the default. On a cluster a classic message travels to every node, so every node carries the traffic of every channel. It is the mode for standalone and sentinel servers, and the only one a pattern has. PubSubMode::Sharded needs Redis 7 or later. A sharded channel lives on the node that owns its hash slot, so its traffic stays there and a cluster scales with the number of channels. The two modes do not reach each other: a sharded subscriber hears only SPUBLISH, and a classic one only PUBLISH. So the subscription sets its mode with RedisPubSub::mode, and a publish policy aimed at it sets the same one with RedisPubSubPublish::mode. One service reads a channel in one mode: a second subscription that reads it in the other mode does not start.

Example

Orders arrive on a sharded channel, and the shipment answering each one leaves with SPUBLISH to a sharded channel a tracker reads.

Macros
Manual

examples/redis_pubsub_sharded.rs
/// Reads `orders.eu` with `SSUBSCRIBE` and answers on `shipments.eu`.
#[subscriber(RedisPubSub::new("orders.eu").mode(PubSubMode::Sharded), reply)]
async fn ship(order: &Order) -> Shipment {
    tracing::info!(order.id, "shipping");
    Shipment { order: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        RedisBroker::cluster(["redis://localhost:7000"]),
        |b| {
            // The reply leaves with `SPUBLISH`: a sharded subscriber hears only a sharded publish.
            b.include(ship)
                .out_reply(Publish::new().mode(PubSubMode::Sharded));
            b.include(track);
        },
    )
}

examples/manual/redis_pubsub_sharded.rs
/// Reads `orders.eu` with `SSUBSCRIBE` and answers on `shipments.eu`.
struct Ship;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        RedisBroker::cluster(["redis://localhost:7000"]),
        |b| {
            let orders = RedisPubSub::new("orders.eu").mode(PubSubMode::Sharded);
            // The reply leaves with `SPUBLISH`: a sharded subscriber hears only a sharded publish.
            b.include(subscriber(orders, Ship).reply().build())
                .out_reply(Publish::new().mode(PubSubMode::Sharded));
            let shipments = RedisPubSub::new("shipments.eu").mode(PubSubMode::Sharded);
            b.include(subscriber(shipments, Track).build());
        },
    )
}

Example​

Example