Keys

The nats-partition-key header keys the lanes of workers(n, by_key): messages sharing a key run on one lane, in order.

workers(n, by_key) runs n sequential lanes in one subscription. Messages with one key run in order on one lane, and different keys run at once. NATS has no partitions or message keys, so the key travels in the nats-partition-key header. The sender sets it: any NATS client writes the header, and a RustStream message type declares it in its typed header contract. The value is text, because a NATS header is; a value that is not a single line of UTF-8 fails the publish. A message without the header has no key and rotates over the lanes. The lanes work the same on Core NATS and on a stream. The manual form writes the lanes as workers_by_key(n) on the mount chain.

Example

Two tenants' orders, each order a second of work: the tenants run side by side, two seconds in all.

Macros
Manual

examples/nats_subscriptions_keys.rs
/// Eight lanes keyed by the partition-key header: one tenant's orders run in order, different
/// tenants at once.
#[subscriber(CoreSubject::new("orders.placed"), workers(8, by_key))]
async fn place(order: &Order) -> HandlerOutcome {
    process(order).await;
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(place);
        },
    )
}

examples/manual/nats_subscriptions_keys.rs
/// Eight lanes keyed by the partition-key header: one tenant's orders run in order, different
/// tenants at once.
struct Place;

impl Handle<Order> for Place {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        process(order).await;
        Ok(())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let placed = CoreSubject::new("orders.placed");
            b.include(
                subscriber(placed, Place)
                    .workers_by_key(nonzero!(8))
                    .build(),
            );
        },
    )
}

Example​

Example