RedisPublish

RedisPublish, Publish in the stream prelude, appends each message to a stream with XADD.

A mount site names it wherever a registration publishes: .out_reply(Publish) for a reply, .out(Slot, Publish) for a slot, .out_retry(Publish) for retry copies. The policy holds no connection and no setting. The runtime pairs it with the connected broker at startup into a RedisPublisher, so a handler never holds a publisher that is not connected. The message's name is the stream key, and the entry id is *, the one Redis assigns. The payload becomes the entry's _payload field, and each header a field of its own named h:<header>. A publish returns once Redis has answered the XADD.

The stream prelude also brings TransactionalPublish, the policy of a mount that publishes in transactions; it pairs into the same publisher, and a cluster refuses it at startup. A list or a channel has its own policy: RedisListPublish and RedisPubSubPublish. A registration that names no policy for its reply publishes through RedisDefaultPublish. Per message, a publish carries a partition key. Outside an app, ConnectedRedisBroker::publisher hands out the same publisher.

A reply goes to the stream its type declares with #[outgoing(name = "..")], or to the one the mount site names in reply(".."). Redis carries no reply address and no correlation of its own: a reply is an ordinary publish to a name both sides agree on.

Example

Macros
Manual

examples/redis_publishing.rs
/// The stream a receipt goes to is a property of the type.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

#[subscriber(RedisStream::new("orders").group("workers"), reply)]
async fn confirm(order: &Order) -> Receipt {
    tracing::info!(order.id, "order confirmed");
    Receipt { order_id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            // Each receipt is an `XADD` to the `receipts` stream.
            b.include(confirm).out_reply(Publish);
        },
    )
}

examples/manual/redis_publishing.rs
/// The stream a receipt goes to is a property of the type.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
#[outgoing(name = "receipts")]
struct Receipt {
    order_id: u64,
}

struct Confirm;

impl Handle<Order, Receipt> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        tracing::info!(order.id, "order confirmed");
        ready(Ok(Receipt { order_id: order.id }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            let confirm = subscriber(RedisStream::new("orders").group("workers"), Confirm);
            // Each receipt is an `XADD` to the `receipts` stream.
            b.include(confirm.reply().build()).out_reply(Publish);
        },
    )
}

Example​

Example