Skip to main content

Fan-out with Pub/Sub

This guide broadcasts messages to every running instance of a service with Redis Pub/Sub.

The example is a storefront. Every instance refreshes its price cache when a price changes, follows order events through a pattern, and announces shipped orders.

Every subscriber receives the message​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "prices.changed")]
struct PriceChanged {
    sku: String,
    price_cents: u64,
}

#[subscriber(RedisPubSub::new("prices.changed").buffer(nonzero!(1_000)))]
async fn refresh_price(change: &PriceChanged) -> HandlerOutcome {
    tracing::info!(change.sku, change.price_cents, "price cache refreshed");
    HandlerOutcome::ack()
}

#[subscriber(RedisPubSub::new("prices.changed"))]
async fn reindex_price(change: &PriceChanged) -> HandlerOutcome {
    tracing::info!(change.sku, "search index updated");
    HandlerOutcome::ack()
}

RedisPubSub subscribes to one channel. A message published to the channel reaches every subscription connected at that moment: refresh_price and reindex_price both receive it, in every running instance. Nothing is shared out and nothing is acknowledged; the outcome a handler returns settles nothing.

Redis keeps no copy. A message published while no subscription is connected is gone, and so is one published to an instance that is restarting. A subscription opens only once Redis has routed its channel to it, so a message published after the service has started is not lost.

A slow subscriber​

Redis pushes each message as soon as it is published, whether the handler is ready or not. The subscription holds what the handler has not taken yet, up to its buffer: 32 messages unless buffer sets another size. Once the buffer is full, each new message drops the oldest, and the subscription logs a warning with the channel and the number it lost. refresh_price sets a buffer of a thousand messages, so a burst of price changes waits for it instead of being cut. The buffer is reserved when the subscription opens, about 600 bytes per message whatever the payloads weigh.

Patterns​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct OrderEvent {
    order_id: u64,
}

#[subscriber(RedisPubSubPattern::new("orders.events.*"))]
async fn track_order(event: &OrderEvent, Ctx(channel): Ctx<keys::Channel>) -> HandlerOutcome {
    tracing::info!(event.order_id, channel, "order event");
    HandlerOutcome::ack()
}

RedisPubSubPattern subscribes to a glob with PSUBSCRIBE, here every channel under orders.events.. The handler reads the channel a message came on from the Pub/Sub context with Ctx<keys::Channel>. keys::FromPattern tells whether the message matched through a pattern.

Publishing to a channel​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    order_id: u64,
}

#[subscriber(
    RedisStream::new("shipments").group("notifier"),
    reply("orders.events.shipped")
)]
async fn announce_shipment(shipment: &Shipment) -> OrderEvent {
    OrderEvent {
        order_id: shipment.order_id,
    }
}

announce_shipment reads shipments off a stream and replies with an OrderEvent on the channel orders.events.shipped.

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("storefront", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(refresh_price);
            b.include(reindex_price);
            // A glob is no channel a publish can reach, so the registration names one for copies.
            b.include(track_order)
                .out_retry(pubsub::Publish::default())
                .to("orders.events.retry");
            b.include(announce_shipment)
                .out_reply(pubsub::Publish::default());
        },
    )
}

The reply leaves through pubsub::Publish, which sends PUBLISH. The broker's default publisher treats a name that no subscription of this service reads as a stream key and sends XADD to it. The pattern subscription does not read orders.events.shipped by name, so the registration names the Pub/Sub policy. The pattern subscription in the same service receives the announcement.

A pattern names no channel a copy could be published to. So track_order's registration names one with out_retry(..).to("orders.events.retry"), and the service does not start without it. That channel matches the pattern, so a retried event comes back to the handler, in every running instance.

Sharded channels on a cluster​

On a cluster, a classic PUBLISH goes to every node. PubSubMode::Sharded uses SSUBSCRIBE and SPUBLISH instead: a channel lives on the node that owns its hash slot, so the traffic spreads across the cluster. It needs Redis 7 or later and has no patterns. The two modes do not reach each other, so the subscription and the publish policy name the same mode: RedisPubSub::new(..).mode(PubSubMode::Sharded) and pubsub::Publish::new().mode(PubSubMode::Sharded).

When the connection drops​

The broker reconnects after a dropped connection, a restart of Redis or a failover, and every Pub/Sub subscription subscribes again. The messages published while it was away are not delivered. A sharded subscription is not restored after a resharding moves its slot to another node.

When to use a stream instead​

Pub/Sub suits what is useful only now: a cache to refresh, a live dashboard, a notice to every instance. When each reader must receive every message, including the ones published while it was down, use a stream with one consumer group per reader. Each group reads every entry, and an entry stays in the stream until it is trimmed.

The tests in the example file publish a price change and a shipment and check that every subscription received what was published.