Lanes

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

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

MQTT has no partitions, so the key is the header PARTITION_KEY_HEADER, partition-key, which travels as an MQTT 5 user property. The protocol gives the property 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 subscription sets the header on each message, such as the tenant or the device the message belongs to. The lanes order work inside one connection. A share group spreads messages over instances and keeps no order by key between them. The manual form writes the lanes as .workers_by_key(n) on the mount chain.

Example

The checkout keys each order by its tenant. Orders of one tenant run in order on one lane, and different tenants run side by side.

Macros
Manual

examples/mqtt_subscriptions_lanes.rs
use ruststream_rumqttc::PARTITION_KEY_HEADER;
use ruststream_rumqttc::prelude::*;

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

/// The checkout keys every order it places by its tenant.
#[subscriber("checkout/completed")]
async fn checkout(purchase: &Purchase, Out(orders): Out<impl Publisher>) -> HandlerOutcome {
    let order = Order {
        id: purchase.id,
        tenant: purchase.tenant.clone(),
    };
    let mut key = HeaderMap::new();
    key.insert(PARTITION_KEY_HEADER, purchase.tenant.clone());
    if orders
        .message(&order)
        .with_headers(key)
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "orders-svc"),
        |b| {
            b.include(place);
            b.include(checkout)
                .out(DefaultSlot, Publish::default())
                .build();
        },
    )
}

examples/manual/mqtt_subscriptions_lanes.rs
use ruststream_rumqttc::PARTITION_KEY_HEADER;
use ruststream_rumqttc::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, tenant = %order.tenant, "order placed");
        ready(Ok(()))
    }
}

/// The checkout keys every order it places by its tenant.
struct Checkout;

impl<Entry> Handle<Purchase, (), Outs<(Entry,)>> for Checkout
where
    Entry: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        purchase: &Purchase,
        outs: &Outs<(Entry,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let order = Order {
            id: purchase.id,
            tenant: purchase.tenant.clone(),
        };
        let mut key = HeaderMap::new();
        key.insert(PARTITION_KEY_HEADER, purchase.tenant.clone());
        let placed = outs.get(DefaultSlot).message(&order).with_headers(key);
        if placed.publish().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "orders-svc"),
        |b| {
            // Eight lanes: one tenant's orders never run at the same time, and different
            // tenants run side by side.
            b.include(
                subscriber("orders/placed", Place)
                    .workers_by_key(nonzero!(8))
                    .build(),
            );
            b.include(subscriber("checkout/completed", Checkout).build())
                .out(DefaultSlot, Publish::default())
                .build();
        },
    )
}

Example​

Example