partition_key

The partition_key step keys one record on the publish builder; a body that names it bounds its slot on KinesisPublishOptions.

The step stands among the framework's own: .message(..), .to(..), .partition_key(..), .publish(). It sets the key of this record only, and it wins over every other place a key can come from. The step comes from KinesisPublishSteps in the prelude. It is reachable where the publisher's options are Kinesis's, so the body bounds its slot as Out<impl Publisher<Options = KinesisPublishOptions>, Slot>. That bound is the one place a handler body says which broker it runs on.

The step sets an option of the publish, and it wraps nothing. So the record still leaves through the entry the mount site named: that entry's codec encodes it, its transforms run, and a test attributes the publish to its slot. A test reads the options of a slot's publishes with tb.out::<Slot>().with_options(..).

Example

Every journal entry of one tenant carries the tenant as its key, so the journal keeps each tenant's order.

Macros
Manual

examples/kinesis_publishing_step.rs
/// The slot the journal entries leave through.
#[derive(OutSlot)]
#[publishes(Journalled)]
struct Journal;

/// Every entry of one tenant lands on one shard, so the journal keeps each tenant's order. The
/// bound on the options type puts the step within reach, and says that this body runs on
/// Kinesis.
#[subscriber(KinesisStream::new("orders"))]
async fn record(
    order: &Order,
    Out(journal): Out<impl Publisher<Options = KinesisPublishOptions>, Journal>,
) -> HandlerOutcome {
    let sent = journal
        .message(&Journalled { order: order.id })
        .partition_key(order.tenant.as_str())
        .publish()
        .await;
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/kinesis_publishing_step.rs
/// The slot the journal entries leave through.
#[derive(OutSlot)]
#[publishes(Journalled)]
struct Journal;

/// Every entry of one tenant lands on one shard, so the journal keeps each tenant's order. The
/// bound on the options type puts the step within reach, and says that this body runs on
/// Kinesis.
struct Record;

impl<Entry> Handle<Order, (), Outs<(Entry,)>> for Record
where
    Entry: OutEntry<Journal, Wire: Publisher<Options = KinesisPublishOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        outs.get(Journal)
            .message(&Journalled { order: order.id })
            .partition_key(order.tenant.as_str())
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

Example​

Example