buffer

buffer sets how many messages a channel subscription holds while its handler is busy; past it the oldest is lost with a warning.

Redis pushes every message to the subscription as soon as it is published, whether the handler is ready or not. The subscription keeps what arrives in a buffer until the handler reads it. Without the setting the buffer holds fred's default of 32 messages. When it is full, the oldest message is dropped, and the subscription logs a warning with the channel and the number of messages lost. A handler that has to absorb a burst sets a buffer as large as the burst. The room is reserved when the subscription opens, about 600 bytes per message whatever the payloads weigh, because fred sizes all eight notification channels of a client by it. RedisPubSubPattern::buffer sets the same for a pattern subscription.

Example

Trades arrive in bursts, so their subscription holds up to 1024 of them.

Macros

examples/redis_pubsub.rs

/// A burst of trades waits in a buffer of 1024 messages while the handler works through it.
#[subscriber(RedisPubSub::new("trades").buffer(nonzero!(1024)))]
async fn on_trade(tick: &Tick) -> HandlerOutcome {
    tracing::info!(symbol = %tick.symbol, tick.cents, "trade");
    HandlerOutcome::ack()
}

Manual

examples/manual/redis_pubsub.rs

/// A burst of trades waits in a buffer of 1024 messages while the handler works through it.
struct OnTrade;

impl Handle<Tick> for OnTrade {
    fn handle(
        &self,
        tick: &Tick,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(symbol = %tick.symbol, tick.cents, "trade");
        ready(Ok(()))
    }
}

fn trades() -> RedisPubSub {
    RedisPubSub::new("trades").buffer(nonzero!(1024))
}