Batches

A slice handler gets a batch assembled on the client, up to the mount site's size or until batch_wait (50 ms) passes since the first delivery.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. AMQP pushes one delivery at a time, so the subscription collects deliveries into a batch on the client. A batch closes when it holds n deliveries, or when batch_wait has passed since its first delivery, 50 ms by default. A raised wait gives fuller batches on a sparse queue; a lowered one hands a short batch over sooner. The server sends no more than the prefetch window, so a window below n keeps every batch below n.

One outcome settles every delivery of the batch. In the manual form the body is an impl Handle<[T]>: Ok(()) acknowledges the batch, and Err(outcomes) settles delivery i with outcome i. workers(n) keeps up to n batches in flight at once.

Example

Payments settle up to 32 at a time, and a batch that does not fill closes 200 ms after its first payment.

Macros
Manual

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

/// Settles up to 32 payments in one call; a batch that does not fill closes 200 ms after its
/// first payment arrived.
#[subscriber(RabbitQueue::new("payments")
    .prefetch(nonzero!(64))
    .batch_wait(Duration::from_millis(200)))]
async fn settle(payments: &[Payment]) -> HandlerOutcome {
    let cents: u64 = payments.iter().map(|payment| payment.cents).sum();
    tracing::info!(count = payments.len(), cents, "payments settled");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            b.include(settle.batch(nonzero!(32)));
        },
    )
}

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

/// Settles a batch of payments; `Ok` acknowledges the whole batch.
struct Settle;

impl Handle<[Payment]> for Settle {
    fn handle(
        &self,
        payments: &[Payment],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        let cents: u64 = payments.iter().map(|payment| payment.cents).sum();
        tracing::info!(count = payments.len(), cents, "payments settled");
        ready(Ok(()))
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672"),
        |b| {
            // Up to 32 payments per call; a batch that does not fill closes 200 ms after its
            // first payment arrived.
            let payments = RabbitQueue::new("payments")
                .prefetch(nonzero!(64))
                .batch_wait(Duration::from_millis(200));
            b.include(subscriber(payments, Settle).batch(nonzero!(32)).build());
        },
    )
}

Example​

Example