Workers

workers(n) handles up to n deliveries of one subscription at once; workers(n, by_key) keeps the deliveries of one key in order.

The subscription starts n long-lived workers, tasks of the runtime the app runs on. A free worker takes the next delivery. While n deliveries are in flight, the subscription does not read the broker, so back-pressure holds. Deliveries finish in any order: the subscription's global order is lost. Workers suit a handler that waits on I/O; a handler that computes belongs on threads. The setting is a clause of the attribute, or .workers(n) among the mount-site settings. The count is a NonZeroUsize, and nonzero! writes one. A router chain takes a Workers value through Router::workers. Handing a delivery to a worker allocates nothing. A batch handler gets a pool of batches, each batch in a task of its own. On shutdown, the deliveries in flight finish within the app's shutdown_timeout.

Example

Four charges that each wait a second on a payment provider take one second together.

Macros

examples/workers.rs

/// Up to four orders are charged at once; the next one waits for a free worker.
#[subscriber("orders", workers(4))]
async fn bill(order: &Order) -> HandlerOutcome {
    charge(order).await;
    HandlerOutcome::ack()
}

Manual

examples/manual/workers.rs

/// Up to four orders are charged at once; the next one waits for a free worker.
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders", Bill).workers(nonzero!(4)).build());
    })
}

By key

workers(n, by_key) makes the same n workers into n sequential lanes. A delivery goes to the lane picked by hashing its partition key, which the broker's message reports through Partitioned. Deliveries with one key never reorder, and deliveries with different keys run at once. A delivery without a key rotates over the lanes. The in-memory broker reads the key from the partition-key header. At the mount site the setting is .workers_by_key(n).

Macros

examples/workers.rs

/// Four lanes by the message's partition key: one customer's orders are charged in order,
/// different customers' at once.
#[subscriber("orders", workers(4, by_key))]
async fn bill(order: &Order) -> HandlerOutcome {
    charge(order).await;
    HandlerOutcome::ack()
}

Manual

examples/manual/workers.rs

/// Four lanes by the message's partition key: one customer's orders are charged in order,
/// different customers' at once.
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(
            subscriber("orders", Bill)
                .workers_by_key(nonzero!(4))
                .build(),
        );
    })
}