Batches

A slice handler takes a batch the service assembles: full at the mount site's size, or closed batch_wait after its first delivery.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. The Pulsar client hands over one delivery at a time, so the batch is assembled in the service. A batch closes when it holds n deliveries, or when batch_wait has passed since its first delivery, whichever comes first. batch_wait is a setting of the descriptor, 10 ms by default; a longer wait gives fuller batches on a quiet topic, at the cost of latency. An idle subscription waits for the first delivery as long as it takes, so a batch is never empty. One outcome settles every delivery of the batch. A Vec<HandlerOutcome> settles delivery i with outcome i. workers(n) keeps up to n batches in flight at once. A batch body reads a PulsarBatchContext, which carries the seek handle and no per-delivery field. In the manual form the body is an impl Handle<[T]>, where Ok(()) acks the batch and Err(outcomes) settles delivery by delivery.

Acknowledgement records the stock counts of warehouse scans in batches.

Example

Readings that arrive together are stored in one call.

Macros

examples/pulsar_batches.rs

/// Stores a batch of readings in one round-trip; one outcome settles the whole batch. A batch
/// that is not full closes 250 ms after its first reading.
#[subscriber(
    PulsarSubscription::new("readings", "aggregator")
        .batch_wait(Duration::from_millis(250))
)]
async fn aggregate(readings: &[Reading]) -> HandlerOutcome {
    tracing::info!(count = readings.len(), "stored in one round-trip");
    HandlerOutcome::ack()
}

Manual

examples/manual/pulsar_batches.rs

/// Stores a batch of readings in one round-trip; `Ok` settles the whole batch.
struct Aggregate;

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

The batch size

The mount site names the size, so the same handler can run with another size in another service.

Macros

examples/pulsar_batches.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("readings", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            // At most 64 readings per call.
            b.include(aggregate.batch(nonzero!(64)));
        },
    )
}

Manual

examples/manual/pulsar_batches.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("readings", "0.1.0")).with_broker(
        PulsarBroker::new("pulsar://localhost:6650"),
        |b| {
            // At most 64 readings per call; a batch that is not full closes 250 ms after its
            // first reading.
            b.include(
                subscriber(
                    PulsarSubscription::new("readings", "aggregator")
                        .batch_wait(Duration::from_millis(250)),
                    Aggregate,
                )
                .batch(nonzero!(64))
                .build(),
            );
        },
    )
}