partition_key

partition_key keys one message, so workers(n, by_key) on the consuming side keeps the order of messages sharing that key.

workers(n, by_key) runs a subscription on n lanes, and deliveries with one partition key always share a lane. The key travels in the PARTITION_KEY_HEADER header, which is where the consuming side reads it. Writing that header at the call site works the same, and carries across brokers.

The step goes anywhere after .message(..) on the publish builder, so it composes with a header contract the message type declares. It writes through RedisPublishOptions, the per-message options of every publisher in this crate. The slot is bounded as Out<impl Publisher<Options = RedisPublishOptions>, Slot>, which keeps the step off another broker's publisher. Naming that type is the one case where a handler body imports this crate's prelude instead of the framework's alone. A publish without the step carries no options and leaves the headers as they are.

Example

The intake forwards each order keyed by its customer; the ledger books on four lanes, and one customer's orders keep their order.

Macros
Manual

examples/redis_publishing_partition_key.rs
/// Forwards each order keyed by its customer.
#[subscriber(RedisStream::new("orders.in").group("intake"))]
async fn forward(
    order: &Order,
    Out(ledger): Out<impl Publisher<Options = RedisPublishOptions>, Ledger>,
) -> HandlerOutcome {
    let keyed = KeyedOrder {
        id: order.id,
        customer: order.customer.clone(),
    };
    let published = ledger
        .message(&keyed)
        .partition_key(order.customer.as_str())
        .publish()
        .await;
    if published.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// Four lanes; the orders of one customer share a lane and keep their order.
#[subscriber(RedisStream::new("orders.keyed").group("ledger"), workers(4, by_key))]
async fn book(order: &KeyedOrder) -> HandlerOutcome {
    tracing::info!(order.id, order.customer, "order booked");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(forward).out(Ledger, Publish).build();
            b.include(book);
        },
    )
}

examples/manual/redis_publishing_partition_key.rs
/// Forwards each order keyed by its customer.
struct Forward;

impl<L> Handle<Order, (), Outs<(L,)>> for Forward
where
    L: OutEntry<Ledger, Wire: Publisher<Options = RedisPublishOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(L,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let keyed = KeyedOrder {
            id: order.id,
            customer: order.customer.clone(),
        };
        outs.get(Ledger)
            .message(&keyed)
            .partition_key(order.customer.as_str())
            .publish()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

/// Books an order; the lanes keep one customer's orders in order.
struct Book;

impl Handle<KeyedOrder> for Book {
    fn handle(
        &self,
        order: &KeyedOrder,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, order.customer, "order booked");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            let forward = subscriber(RedisStream::new("orders.in").group("intake"), Forward);
            b.include(forward.build()).out(Ledger, Publish).build();
            // Four lanes; the orders of one customer share a lane and keep their order.
            let book = subscriber(RedisStream::new("orders.keyed").group("ledger"), Book);
            b.include(book.workers_by_key(nonzero!(4)).build());
        },
    )
}

Example​

Example