Batches

A slice handler takes a batch: one JetStream pull fetch of the mount site's size, or a Core NATS batch the client assembles.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. On a JetStreamSubject one batch is one pull request: the consumer asks the server for up to n messages. The fetch ends once it holds n, or once pull_expires runs out with fewer, 5 seconds by default. pull_expires takes a NonZeroDuration, because a fetch that expires at once would spin. Core NATS has no batch on the wire, so the client collects up to n deliveries and closes a short batch 10 ms after its first one. One outcome settles every message of the batch. A Vec<HandlerOutcome> settles message i with outcome i. In the manual form the body is an impl Handle<[T]>, where Ok(()) acknowledges the batch and Err(outcomes) settles message by message.

Example

Three orders waiting in the stream arrive in one call.

Macros
Manual

examples/nats_subscriptions_batches.rs
/// Reconciles up to 50 orders per call; a fetch that finds fewer closes after 300 ms.
#[subscriber(
    JetStreamSubject::new("orders.*", "ORDERS")
        .durable("reconciler")
        .pull_expires(NonZeroDuration::from_millis(nonzero!(300)))
)]
async fn reconcile(orders: &[Order]) -> HandlerOutcome {
    let total: u64 = orders.iter().map(|order| order.cents).sum();
    tracing::info!(count = orders.len(), total, "orders reconciled");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("reconciliation", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            b.include(reconcile.batch(nonzero!(50)));
        },
    )
}

examples/manual/nats_subscriptions_batches.rs
/// Reconciles up to 50 orders per call; a fetch that finds fewer closes after 300 ms.
struct Reconcile;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("reconciliation", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            let orders = JetStreamSubject::new("orders.*", "ORDERS")
                .durable("reconciler")
                .pull_expires(NonZeroDuration::from_millis(nonzero!(300)));
            b.include(subscriber(orders, Reconcile).batch(nonzero!(50)).build());
        },
    )
}

Example​

Example