ordering_key

The ordering_key step names the ordering key of one publish, on the framework's publish builder.

The step sits between message(..) and publish(), so the publish stays the mount site's: its codec, its transforms and its slot hold. It exists only on a builder over a Pub/Sub publisher. The handler bounds its slot Out<impl Publisher<Options = PubSubPublishOptions>> and imports the prelude, which brings the PubSubOrdering trait. The key reaches the client as the message's own ordering key and never as an attribute. A key named by the call wins over the partition-key header and over the policy's key. A test reads the setting back with tb.out::<Slot>().with_options(..) and a PubSubPublishOptions value.

Example

Each confirmation goes out under its order's key.

Macros

examples/pubsub_publishing_ordering_key.rs

/// Each confirmation goes out under its order's key, named on the call.
#[subscriber("orders-workers")]
async fn confirm(
    order: &Order,
    Out(out): Out<impl Publisher<Options = PubSubPublishOptions>>,
) -> HandlerOutcome {
    let sent = out
        .message(&Confirmation { order_id: order.id })
        .to("confirmations")
        .ordering_key(format!("order-{}", order.id))
        .publish()
        .await;
    match sent {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

Manual

examples/manual/pubsub_publishing_ordering_key.rs

/// Each confirmation goes out under its order's key, named on the call.
struct Confirm;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Confirm
where
    Entry: OutEntry<DefaultSlot, Wire: Publisher<Options = PubSubPublishOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let sent = outs
            .get(DefaultSlot)
            .message(&Confirmation { order_id: order.id })
            .to("confirmations")
            .ordering_key(format!("order-{}", order.id))
            .publish()
            .await;
        sent.map_err(|_| HandlerOutcome::retry())
    }
}