Publisher

Publisher is the capability every slot names; a publish policy at the mount site becomes one when startup pairs it with the connected broker.

A handler bounds its slot on the capability, Out<impl Publisher>, and never on a broker's publisher type. The same handler then mounts on a production broker and on its in-process test transport unchanged. A policy holds no connection, so it is built anywhere: a broker's prelude names its plain one Publish. Startup pairs each policy with the connected broker, so a handler never holds a publisher that is not connected. A reply with no policy named leaves through the broker's default one, DefaultPublish. A startup hook such as after_startup takes a policy and gets its live publisher, for a publish outside any handler. A bare publisher starts a publish with the same builder, under the default codec.

A publisher declares three types. Payload says how its client takes the bytes: Lend when it only reads them, so the runtime lends one reused buffer, and Take when it keeps them. Options are its per-message settings, and Error is what a failed publish returns. Its base_headers go on every message it sends, under the headers of the call site. The narrower capabilities extend it: TransactionalPublisher, OwnedTransactions and RequestReply.

Example

The handler names the capability; the mount site and the startup hook name the policy.

Macros

examples/publisher.rs

/// The bound names a capability, never the broker's publisher type.
#[subscriber("orders")]
async fn archive(order: &Order, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    if out
        .message(order)
        .to("orders.archive")
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Manual

examples/manual/publisher.rs

/// The bound names a capability, never the broker's publisher type.
struct Archive;

impl<Egress> Handle<Order, (), Outs<(Egress,)>> for Archive
where
    Egress: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        if outs
            .get(DefaultSlot)
            .message(order)
            .to("orders.archive")
            .publish()
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Macros

examples/publisher.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("archive", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // A policy holds no connection; startup pairs it with the connected broker.
        b.include(archive).out(DefaultSlot, Publish).build();
        // A startup hook gets the live publisher of the policy it names.
        b.after_startup(Publish, async move |publisher| {
            publisher
                .message(&Started {
                    service: "archive".into(),
                })
                .publish()
                .await
        });
    })
}

Manual

examples/manual/publisher.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("archive", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // A policy holds no connection; startup pairs it with the connected broker.
        b.include(subscriber("orders", Archive).build())
            .out(DefaultSlot, Publish)
            .build();
        // A startup hook gets the live publisher of the policy it names.
        b.after_startup(Publish, async move |publisher| {
            publisher
                .message(&Started {
                    service: "archive".into(),
                })
                .publish()
                .await
        });
    })
}