partition_key

The partition_key step keys one publish from a handler body; it wins over a partition-key header and keeps the slot's codec.

The key does what the partition-key header does: it places the message on a partition and orders KeyShared consumers. The step writes the one field of PulsarPublishOptions and returns the builder. The message still leaves through the slot the mount site wired, in that slot's codec and through its transforms. A body reaches the step by bounding its slot Out<impl Publisher<Options = PulsarPublishOptions>, Slot>. That body imports ruststream_pulsar::prelude::*, the one case where a handler body names this broker. The mount site declares nothing for it, and a publish without the step goes unkeyed. Where a publish carries both the step and the header, the step wins. A test reads the key a slot publish carried with tb.out::<Slot>().with_options(..).

Publishing and replies keys each pick list by its warehouse.

Example

Macros
Manual

examples/pulsar_publishing_partition_key.rs
/// Records a receipt per order, keyed by the user who placed it, so one user's receipts stay
/// in order. A body that names a per-message setting imports this broker's prelude and states
/// the settings type on its slot.
#[subscriber(PulsarSubscription::new("orders", "ledger"))]
async fn record(
    order: &Order,
    Out(ledger): Out<impl Publisher<Options = PulsarPublishOptions>, Ledger>,
) -> HandlerOutcome {
    let sent = ledger
        .message(&Receipt { order_id: order.id })
        .partition_key(format!("user-{}", order.user))
        .publish()
        .await;
    match sent {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(order.id, %err, "the receipt was not recorded");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/pulsar_publishing_partition_key.rs
/// Records a receipt per order, keyed by the user who placed it, so one user's receipts stay
/// in order.
struct Record;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Record
where
    Entry: OutEntry<Ledger, Wire: Publisher<Options = PulsarPublishOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        outs.get(Ledger)
            .message(&Receipt { order_id: order.id })
            .partition_key(format!("user-{}", order.user))
            .publish()
            .await
            .map_err(|err| {
                tracing::warn!(order.id, %err, "the receipt was not recorded");
                HandlerOutcome::retry()
            })
    }
}

Example​

Example