Lanes

workers(n, by_key) keeps deliveries that share the amqp-partition-key header on one lane, in order, while other keys run in parallel.

workers(n, by_key) runs n sequential lanes over one queue. The lane of a delivery is picked by a hash of its partition key, so deliveries with one key run one after another, in queue order. Deliveries with different keys run at once on different lanes. A delivery without a key goes to the next lane in rotation.

On RabbitMQ the key is the header PARTITION_KEY_HEADER, amqp-partition-key. AMQP gives the header no meaning: it is a convention between the publisher, which sets it, and this crate, which reads it back. A service that publishes to a keyed queue sets the header on each message, such as the tenant or the account the message belongs to. The lanes order work inside one service instance. Spreading a stream over several queues by key is the consistent-hash exchange, which the server runs. The manual form writes the lanes as .workers_by_key(n) on the mount chain.

Example

Orders of one tenant run in order on one lane, and different tenants run side by side.

Macros
Manual

examples/lapin_subscriptions_lanes.rs
use ruststream_lapin::prelude::*;

/// Eight lanes: one tenant's orders never run at the same time, and different tenants run side
/// by side.
#[subscriber(RabbitQueue::new("orders"), workers(8, by_key))]
async fn place(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.tenant, "order placed");
    HandlerOutcome::ack()
}

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

examples/manual/lapin_subscriptions_lanes.rs
use ruststream_lapin::prelude::*;

struct Place;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            // Eight lanes: one tenant's orders never run at the same time, and different
            // tenants run side by side.
            let orders = RabbitQueue::new("orders");
            b.include(
                subscriber(orders, Place)
                    .workers_by_key(nonzero!(8))
                    .build(),
            );
        },
    )
}

Example​

Example