Batches

A slice handler takes batches assembled on the client: batch(n) bounds their size, and batch_wait the wait for a partial one.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. The streaming pull hands over one delivery at a time, so the subscription gathers deliveries into a batch itself. A batch goes to the handler when it holds n deliveries, or when batch_wait has passed since its first one, 50 milliseconds by default. So a quiet subscription yields small batches, and a busy one full ones. One outcome settles every delivery of the batch; a Vec<HandlerOutcome> settles delivery i with outcome i. Every delivery of a batch is outstanding while the batch waits, so max_outstanding below n keeps batches short of n. In the manual form the body is an impl Handle<[T]>, where Ok(()) acks the batch and Err(outcomes) settles delivery by delivery.

Example

Three orders published together arrive in one call.

Macros
Manual

examples/pubsub_subscriptions_batches.rs
/// The slice makes this a batch handler; a partial batch waits 200 milliseconds for the rest.
#[subscriber(GooglePubSub::new("orders-workers").batch_wait(Duration::from_millis(200)))]
async fn settle(orders: &[Order]) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "orders settled");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            // At most 50 orders reach the handler per call.
            b.include(settle.batch(nonzero!(50)));
        },
    )
}

examples/manual/pubsub_subscriptions_batches.rs
/// The slice makes this a batch handler; a partial batch waits 200 milliseconds for the rest.
struct Settle;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            // At most 50 orders reach the handler per call.
            b.include(
                subscriber(
                    GooglePubSub::new("orders-workers").batch_wait(Duration::from_millis(200)),
                    Settle,
                )
                .batch(nonzero!(50))
                .build(),
            );
        },
    )
}

Example​

Example