KinesisPublishSettings

partition_key at the mount site keys every record of a position, a reply included, which has no publish call to put a step on.

A replying handler returns a value, and the runtime publishes it. So a reply has no publish builder for the partition_key step. KinesisPublishSettings::partition_key names the key on the policy instead, right after the position it applies to: .out_reply(Publish::default()).partition_key("receipts-v1"), or after .out(Slot, ..). KinesisPublish::partition_key is the same setting written on the policy value. The setting is available after this crate's policy and after no other broker's.

The key ranks third among the places a key comes from, so a step or a partition-key header on a record wins over it. A retry copy carries the delivery's headers, its partition-key among them, so on the retry position the copy keeps the record's own key. A reply that needs a key of its own per delivery takes a publish transform at the mount site that writes partition_key into the options. The AsyncAPI document reports the key on the message object.

Example

Every receipt carries one key, so all receipts land on one shard, in order.

Macros
Manual

examples/kinesis_replies.rs
use ruststream_kinesis::prelude::*;

/// A receipt always goes to the receipts stream, so the type is where that stream is named.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order: u64,
}

/// The handler names what it replies with; how that reply is published is the mount's.
#[subscriber(KinesisStream::new("orders"), reply)]
async fn confirm(order: &Order) -> Receipt {
    Receipt { order: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        // One key means one shard: every receipt is ordered against every other, at the cost of
        // a single shard's throughput. Leave it off to spread them instead.
        b.include(confirm)
            .out_reply(Publish::default())
            .partition_key("receipts-v1");
    })
}

examples/manual/kinesis_replies.rs
use ruststream_kinesis::prelude::*;

/// A receipt always goes to the receipts stream, so the type is where that stream is named.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing, JsonSchema)]
#[outgoing(name = "receipts")]
struct Receipt {
    order: u64,
}

/// The body names what it replies with; how that reply is published is the mount's.
struct Confirm;

impl Handle<Order, Receipt> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Receipt, HandlerOutcome>> {
        ready(Ok(Receipt { order: order.id }))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(KinesisBroker::new(), |b| {
        b.include(
            subscriber(KinesisStream::new("orders"), Confirm)
                .reply()
                .build(),
        )
        // One key means one shard: every receipt is ordered against every other, at the cost of
        // a single shard's throughput. Leave it off to spread them instead.
        .out_reply(Publish::default())
        .partition_key("receipts-v1");
    })
}

Example​

Example