PubSubContext

PubSubContext gives a channel handler the channel a message arrived on and whether a pattern matched it.

The subscription builds it once per message, on a RedisPubSub channel and a RedisPubSubPattern glob alike. On a channel subscription the channel is the one it names. On a pattern subscription it is the concrete channel the glob matched, so it differs from message to message.

Key	Value	What it is
`Channel`	`&str`	the channel the message was published to
`FromPattern`	`bool`	`true` when a `PSUBSCRIBE` glob matched the message
`FredPool`	`fred::clients::Pool`	the broker's connection pool, as FredPool shows

A handler names the context in its signature, ctx: &mut Context<'_, PubSubContext>, and reads each field with ctx.context(key). A pattern subscription names where its retry copies go with out_retry, because a glob is no channel a copy could be published to.

Example

One handler hears every events.* channel and records the channel each event came from.

Macros
Manual

examples/redis_context_pubsub.rs
/// Hears every `events.*` channel and records which one each event came from.
#[subscriber(RedisPubSubPattern::new("events.*"), reply)]
async fn hear(event: &Event, ctx: &mut Context<'_, PubSubContext>) -> Heard {
    let channel = ctx.context(keys::Channel);
    tracing::info!(event.id, channel, "event heard");
    Heard {
        event_id: event.id,
        channel: channel.to_owned(),
        from_pattern: ctx.context(keys::FromPattern),
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            // A pattern names no channel a retry could go back to, so the mount names one.
            b.include(hear)
                .out_reply(stream::Publish)
                .out_retry(pubsub::Publish::new())
                .to("events.retry");
        },
    )
}

examples/manual/redis_context_pubsub.rs
/// Hears every `events.*` channel and records which one each event came from.
struct Hear;

impl Handle<Event, Heard, (), PubSubContext> for Hear {
    fn handle(
        &self,
        event: &Event,
        _outs: &(),
        ctx: &mut Context<'_, PubSubContext>,
    ) -> impl Future<Output = Result<Heard, HandlerOutcome>> {
        let channel = ctx.context(keys::Channel);
        tracing::info!(event.id, channel, "event heard");
        ready(Ok(Heard {
            event_id: event.id,
            channel: channel.to_owned(),
            from_pattern: ctx.context(keys::FromPattern),
        }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            let hear = subscriber(RedisPubSubPattern::new("events.*"), Hear);
            // A pattern names no channel a retry could go back to, so the mount names one.
            b.include(hear.reply().build())
                .out_reply(stream::Publish)
                .out_retry(pubsub::Publish::new())
                .to("events.retry");
        },
    )
}

Example​

Example