Builder

One builder publishes every message: message(&value), the positions its type leaves open, then publish().

A slot, a transaction and any bare Publisher start a publish with message(&value) and return a PublishBuilder. The positions the message type leaves open come next. .to(name) names the destination of a type that declares none, and a template opens .to() with its setters, as on the Destinations page. A publish whose destination is still open does not compile. .with_headers(..) takes the type's declared header contract by reference, or a HeaderMap built by hand. The headers on the wire are the publisher's base headers with the call site's over them, key by key. .with_codec(codec) names a codec for this one publish, over the slot's and the app's. A Serialized value carries its own bytes and has no codec position. A broker adds steps of its own for its per-message settings.

publish() encodes the value and sends it, and returns PublishError on failure. Encode and Serialize mean the value did not encode, and a retry of the same value fails the same way. Publish carries the broker's own error. On a slot the message passes the slot's transforms before the send, and the app's publish layers where they apply.

Example

Macros

examples/publish_builder.rs

#[subscriber("shipments")]
async fn notify(shipment: &Shipment, Out(out): Out<impl Publisher>) -> HandlerOutcome {
    let mut headers = HeaderMap::new();
    headers.insert("carrier", shipment.carrier.clone());
    let sent = out
        .message(&Notice { id: shipment.id })
        .to(format!("notices.{}", shipment.region))
        .with_headers(headers)
        .publish()
        .await;
    match sent {
        Ok(()) => HandlerOutcome::ack(),
        // A value that does not encode fails the same way on every attempt.
        Err(PublishError::Encode(_) | PublishError::Serialize(_)) => HandlerOutcome::drop(),
        Err(_) => HandlerOutcome::retry(),
    }
}

Manual

examples/manual/publish_builder.rs

struct Notify;

impl<Egress> Handle<Shipment, (), Outs<(Egress,)>> for Notify
where
    Egress: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        shipment: &Shipment,
        outs: &Outs<(Egress,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let mut headers = HeaderMap::new();
        headers.insert("carrier", shipment.carrier.clone());
        let sent = outs
            .get(DefaultSlot)
            .message(&Notice { id: shipment.id })
            .to(format!("notices.{}", shipment.region))
            .with_headers(headers)
            .publish()
            .await;
        match sent {
            Ok(()) => Ok(()),
            // A value that does not encode fails the same way on every attempt.
            Err(PublishError::Encode(_) | PublishError::Serialize(_)) => {
                Err(HandlerOutcome::drop())
            }
            Err(_) => Err(HandlerOutcome::retry()),
        }
    }
}