Batches

A slice handler gets the deliveries assembled on the client, up to the size the mount site names, closed early by batch_wait.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. AMQP 1.0 transfers one message at a time, and credit is flow control rather than a batch size, so the subscription assembles the batches itself. A batch closes when it holds n deliveries, or when batch_wait has passed since its first delivery. The default, DEFAULT_BATCH_WAIT, is 10 ms. A longer wait fills batches under a trickle of traffic, at the cost of latency for the first delivery of each batch. batch_wait applies to batch handlers only: a single-message subscription hands every delivery over as it arrives. One outcome settles every delivery of the batch, and a Vec<HandlerOutcome> settles delivery i with outcome i. In the manual form the body is an impl Handle<[T]>, where Ok(()) acks the batch and Err(outcomes) settles it delivery by delivery.

Acknowledgement and delivery guarantees restocks a warehouse from batches.

Example

Three orders waiting on the subscription arrive in one call.

Macros
Manual

examples/amqp_batches.rs
/// Stores a batch of orders in one round-trip; one outcome settles every order of the batch.
/// `batch_wait` caps how long a partial batch waits for the deliveries that would fill it.
#[subscriber(AmqpAddress::queue("orders").batch_wait(Duration::from_millis(50)))]
async fn store(orders: &[Order]) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "orders stored in one round-trip");
    HandlerOutcome::ack()
}

examples/manual/amqp_batches.rs
/// Stores a batch of orders in one round-trip; `Ok` settles every order of the batch.
struct Store;

impl Handle<[Order]> for Store {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = orders.len(), "orders stored in one round-trip");
        ready(Ok(()))
    }
}

/// `batch_wait` caps how long a partial batch waits for the deliveries that would fill it.
fn orders() -> AmqpAddress {
    AmqpAddress::queue("orders").batch_wait(Duration::from_millis(50))
}

The mount site names the size:

Macros
Manual

examples/amqp_batches.rs
#[ruststream::app]
fn app() -> impl App {
    let broker = AmqpBroker::new("amqp://artemis:artemis@localhost:5672");
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        // At most 32 orders per call.
        b.include(store.batch(nonzero!(32)));
    })
}

examples/manual/amqp_batches.rs
fn app() -> impl App {
    let broker = AmqpBroker::new("amqp://artemis:artemis@localhost:5672");
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        // At most 32 orders per call.
        b.include(subscriber(orders(), Store).batch(nonzero!(32)).build());
    })
}

Example​

Example