Batches

A slice handler reads batches the crate assembles on the client: a batch closes at the size the mount site names or 20 ms after its first delivery.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. An MQTT packet carries one message, so the crate collects deliveries into batches on the client. A batch closes when it holds n deliveries, or 20 milliseconds after its first delivery, whichever comes first. The size is the mount site's, and the 20 ms deadline is the crate's. A subscription with nothing to read waits for its first delivery as long as it takes. Nothing at the mount site says which side of the wire filled a batch, so a batch handler written for another broker mounts here unchanged.

One outcome settles every delivery of the batch, and a Vec<HandlerOutcome> settles delivery i with outcome i. Each delivery is acknowledged on its own. A batch abandoned while it fills leaves its deliveries unacknowledged. In the manual form the body is an impl Handle<[T]>, where Ok(()) acks the batch and Err(outcomes) settles delivery by delivery.

Example

The readings of a share group, stored up to 64 at a time.

Macros
Manual

examples/mqtt_batches.rs
/// Stores the readings of a whole batch in one write; one outcome settles every reading of it.
#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce).shared("workers"))]
async fn ingest(readings: &[Reading]) -> HandlerOutcome {
    tracing::info!(count = readings.len(), "readings stored in one write");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("telemetry-batches", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-batches"),
        |b| {
            // At most 64 readings per call; a batch that has fewer closes 20 ms after its first.
            b.include(ingest.batch(nonzero!(64)))
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
        },
    )
}

examples/manual/mqtt_batches.rs
/// Stores the readings of a whole batch in one write; `Ok` settles every reading of it.
struct Ingest;

impl Handle<[Reading]> for Ingest {
    fn handle(
        &self,
        readings: &[Reading],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = readings.len(), "readings stored in one write");
        ready(Ok(()))
    }
}

fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("telemetry-batches", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-batches"),
        |b| {
            let fleet = MqttFilter::new("devices/+/telemetry")
                .qos(Qos::AtLeastOnce)
                .shared("workers");
            // At most 64 readings per call; a batch that has fewer closes 20 ms after its first.
            b.include(
                subscriber("devices/+/telemetry", Ingest)
                    .batch(nonzero!(64))
                    .build()
                    .map_source(|_| fleet),
            )
            .out_retry(Publish::default())
            .to("devices/retry/telemetry");
        },
    )
}

Example​

Example