Skip to main content

Batches

A handler over &[T] runs once per batch the broker delivers; the mount site names the batch size, and the body settles the batch or each element.

The slice parameter is the whole declaration: a &[T] handler is a batch handler, and a &T handler is the single-message form of #[subscriber]. batch(n), one of the mount-site settings, caps each batch at n messages, and a batch handler does not mount without it. The broker builds the batches, natively or through the framework's client-side buffer, and the body sees each batch as delivered. A batch may hold fewer than n messages. The batch's return value settles every element with one of the outcomes: HandlerOutcome, (), Result<(), E> or Result<HandlerOutcome, E>. A Vec<HandlerOutcome> settles each element on its own, retry_after included; a vector of the wrong length is logged, and the elements it leaves unsettled are retried. An element that fails to decode is settled by the decode failure policy and never reaches the body. The body gets one context per batch, with the broker's subscription-scoped fields only. Per-message headers come with the elements when the parameter is &[Message<H, T>]. App-wide and router middleware wrap single-message handlers and do not apply to batch handlers. In the manual form the body is an impl Handle<[T]>; Ok(()) acks the batch and Err(outcomes) settles element by element. A test reads the batches with batches and checks their sizes with assert_batch_sizes.

Example​

One outcome settles the whole batch.

examples/subscriber_batch.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
}

/// One call per batch; the one outcome settles every order in it.
#[subscriber("orders")]
async fn store(orders: &[Order]) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "stored in one round-trip");
    HandlerOutcome::ack()
}

fn app(broker: MemoryBroker) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        // At most 64 orders per call.
        b.include(store.batch(nonzero!(64)));
    })
}

One outcome per element​

A Vec<HandlerOutcome> settles element i with outcome i.

examples/subscriber_batch.rs
const RECHECK: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "payments")]
struct Payment {
    id: u64,
    cleared: bool,
}

/// One outcome per element, in slice order: only the uncleared payments come back.
#[subscriber("payments")]
async fn reconcile(payments: &[Payment]) -> Vec<HandlerOutcome> {
    payments
        .iter()
        .map(|payment| {
            if payment.cleared {
                HandlerOutcome::ack()
            } else {
                HandlerOutcome::retry_after(RECHECK)
            }
        })
        .collect()
}