Batches

A slice handler takes a batch assembled on the client, up to the size batch(n) names; a partial batch leaves 10 ms after its first delivery.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. Both clients read one message at a time, so the crate assembles each batch on the client, on a stream file and on a pipe alike. A batch takes what the transport already holds, up to n, and does not wait to fill up. A batch that is not full goes out 10 ms after its first delivery, so at a quiet tail a batch is often one message. A Vec<HandlerOutcome> settles message i with outcome i, and a single HandlerOutcome settles the whole batch. A batch body on a stream file reaches the seek handle through FileBatchContext. In the manual form the body is an impl Handle<[T]>, where Ok(()) settles the whole batch and Err(outcomes) settles message by message.

Example

The handler sums a batch of readings and settles each one.

Macros

examples/file_batches.rs

/// Sums a batch of readings and settles each one; a batch is at most the size the mount site
/// names, and smaller when the file had no more.
#[subscriber(FileStream::new("readings"), start_at(FilePosition::beginning()))]
async fn aggregate(batch: &[Reading]) -> Vec<HandlerOutcome> {
    let total: u32 = batch.iter().map(|reading| reading.millivolts).sum();
    tracing::info!(readings = batch.len(), total, "aggregated");
    batch.iter().map(|_| HandlerOutcome::ack()).collect()
}

Manual

examples/manual/file_batches.rs

/// Sums a batch of readings; `Ok` settles the whole batch.
struct Aggregate;

impl Handle<[Reading]> for Aggregate {
    fn handle(
        &self,
        batch: &[Reading],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        let total: u32 = batch.iter().map(|reading| reading.millivolts).sum();
        tracing::info!(readings = batch.len(), total, "aggregated");
        ready(Ok(()))
    }
}

The batch size

Ten readings recorded in the file arrive as batches of four, four and two.

Macros

examples/file_batches.rs

/// The service on the sensors' stream file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("readings", "0.1.0")).with_broker(file, |b| {
        b.include(aggregate.batch(nonzero!(4)));
    })
}

Manual

examples/manual/file_batches.rs

/// The service on the sensors' stream file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("readings", "0.1.0")).with_broker(file, |b| {
        b.include(
            subscriber(FileStream::new("readings"), Aggregate)
                .batch(nonzero!(4))
                .start_at(FilePosition::beginning())
                .build(),
        );
    })
}