RedisPubSubPattern

RedisPubSubPattern reads every channel a glob matches; a retry copy needs a destination named at the mount site.

RedisPubSubPattern::new("quotes.*") subscribes with PSUBSCRIBE. It reads quotes.eu, quotes.us and any other channel the glob matches, with the glob syntax of Redis. A channel needs no setup in Redis, so a channel first used after startup is read as well. The subscription's name is the glob itself. Patterns are classic Pub/Sub: sharded delivery has no pattern subscription, and the descriptor takes no mode. A handler reads the channel a delivery came on through the Channel key of PubSubContext. A glob is not a channel a PUBLISH can name, so the descriptor offers no address for a retry copy. The registration names one with .out_retry(policy).to(channel), and a registration that names none does not start. The rest of the channel semantics are those of RedisPubSub: nothing is kept and nothing is settled.

Example

The handler reads every quotes.* channel.

Macros
Manual

examples/redis_pubsub.rs
/// Every channel whose name matches `quotes.*`: `quotes.eu`, `quotes.us` and any added later.
#[subscriber(RedisPubSubPattern::new("quotes.*"))]
async fn on_quote(tick: &Tick) -> HandlerOutcome {
    tracing::info!(symbol = %tick.symbol, tick.cents, "quote");
    HandlerOutcome::ack()
}

examples/manual/redis_pubsub.rs
/// Every channel whose name matches `quotes.*`: `quotes.eu`, `quotes.us` and any added later.
struct OnQuote;

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

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

Its registration names quotes.retry for retry copies, a channel the glob matches, so a copy comes back to the same handler.

Macros
Manual

examples/redis_pubsub.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("market", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(on_price);
            // A glob is not a channel a `PUBLISH` can name, so the registration names the one
            // a retry copy goes to. The pattern matches it, so the copy comes back here.
            b.include(on_quote)
                .out_retry(Publish::default())
                .to("quotes.retry");
            b.include(on_trade);
        },
    )
}

examples/manual/redis_pubsub.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("market", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(subscriber(prices(), OnPrice).build());
            // A glob is not a channel a `PUBLISH` can name, so the registration names the one
            // a retry copy goes to. The pattern matches it, so the copy comes back here.
            b.include(subscriber(quotes(), OnQuote).build())
                .out_retry(Publish::default())
                .to("quotes.retry");
            b.include(subscriber(trades(), OnTrade).build());
        },
    )
}

Example​

Example