RedisPubSubPublish

RedisPubSubPublish, Publish in the Pub/Sub prelude, sends with PUBLISH or SPUBLISH in the framing its codec sets.

The message's name is the channel. mode picks the command: PubSubMode::Classic, the default, sends PUBLISH, and PubSubMode::Sharded sends SPUBLISH to the channel's slot. The mode matches the subscribers': a sharded message reaches only SSUBSCRIBE subscriptions. Headers travel beside the payload in a binary frame by default. codec frames them with a codec instead, which makes the value readable when the data is text; the subscribers set the same codec. A payload that is not text is published byte for byte either way.

The policy pairs into a RedisPubSubPublisher at startup. A message reaches the subscriptions connected when it is sent; Redis keeps no copy for a subscriber that connects later. A publish to a channel nobody hears succeeds.

Example

Each event is answered with an audit entry, sent with PUBLISH to the audit channel its type declares.

Macros
Manual

examples/redis_publishing_pubsub.rs
/// The channel an audit entry goes to is a property of the type.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "audit")]
struct AuditEntry {
    kind: String,
}

#[subscriber(RedisPubSub::new("events"), reply)]
async fn on_event(event: &Event) -> AuditEntry {
    AuditEntry {
        kind: event.kind.clone(),
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            // A `PUBLISH` to the `audit` channel, in place of the default publisher's `XADD`.
            b.include(on_event).out_reply(Publish::new());
        },
    )
}

examples/manual/redis_publishing_pubsub.rs
/// The channel an audit entry goes to is a property of the type.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "audit")]
struct AuditEntry {
    kind: String,
}

struct OnEvent;

impl Handle<Event, AuditEntry> for OnEvent {
    fn handle(
        &self,
        event: &Event,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<AuditEntry, HandlerOutcome>> {
        ready(Ok(AuditEntry {
            kind: event.kind.clone(),
        }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("events", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            let on_event = subscriber(RedisPubSub::new("events"), OnEvent);
            // A `PUBLISH` to the `audit` channel, in place of the default publisher's `XADD`.
            b.include(on_event.reply().build())
                .out_reply(Publish::new());
        },
    )
}

Example​

Example