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.

Macros
Manual

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)));
    })
}

examples/manual/subscriber_batch.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Order {
    id: u64,
}

// `#[outgoing(name = "orders")]` by hand: the type fixes its destination.
impl OutgoingDestination for Order {
    type Form = FixedName;

    const DESTINATION: &'static str = "orders";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

/// One call per batch; `Ok` settles every order in it.
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(), "stored in one round-trip");
        ready(Ok(()))
    }
}

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(subscriber("orders", Store).batch(nonzero!(64)).build());
    })
}

One outcome per element

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

Macros
Manual

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()
}

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

#[derive(Debug, PartialEq, Deserialize, Serialize, schemars::JsonSchema)]
struct Payment {
    id: u64,
    cleared: bool,
}

// `#[outgoing(name = "payments")]` by hand: the type fixes its destination.
impl OutgoingDestination for Payment {
    type Form = FixedName;

    const DESTINATION: &'static str = "payments";
}

impl MessageHeaders for Payment {
    type Contract = NoHeaders;
}

/// One outcome per element, in slice order: only the uncleared payments come back.
struct Reconcile;

impl Handle<[Payment]> for Reconcile {
    fn handle(
        &self,
        payments: &[Payment],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        ready(Err(payments
            .iter()
            .map(|payment| {
                if payment.cleared {
                    HandlerOutcome::ack()
                } else {
                    HandlerOutcome::retry_after(RECHECK)
                }
            })
            .collect()))
    }
}

fn app(broker: MemoryBroker) -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber("payments", Reconcile)
                .batch(nonzero!(64))
                .build(),
        );
    })
}

Example​

One outcome per element​

Example

One outcome per element