Batches

A slice handler reads one XREADGROUP COUNT on a stream; a list or a channel builds its batch on the client up to the same size.

A handler over &[T] is a batch handler, and batch(n) at the mount site is its largest batch. The step is required on a slice handler: mounting one without it does not compile. A single-message handler does not offer the step at all; its concurrency is workers(n). On a stream the size is the COUNT of the read that fetches the batch, XREADGROUP or XAUTOCLAIM. The server sends at most n entries, and the handler gets exactly what one read returned, never an empty batch. A list and a channel deliver one value at a time. Their subscriptions gather a batch on the client from the values already waiting, up to the same n, and hand it over without waiting for more. So a batch is often smaller than n, and a quiet key yields batches of one. One outcome settles every entry of the batch. In the manual form the body is an impl Handle<[T]>, where Ok(()) settles the batch with an ack and Err(outcomes) settles entry i with outcome i. A stream batch reads a StreamBatchContext, which carries the group and its seek handle and no field of one entry. The block step chains after the size.

Example

Three values waiting on each form arrive in one call. The stream handler and the app:

Macros
Manual

examples/redis_subscriptions_batches.rs
/// Bills the orders one read returned, at most 16.
#[subscriber(RedisStream::new("orders").group("billing"))]
async fn bill(orders: &[Order]) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "orders billed");
    HandlerOutcome::ack()
}

fn app() -> impl App<State = ()> {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        b.include(bill.batch(nonzero!(16)));
        b.include(pack.batch(nonzero!(16)));
        b.include(count.batch(nonzero!(16)));
    })
}

examples/manual/redis_subscriptions_batches.rs
/// Bills the orders one read returned.
struct Bill;

impl Handle<[Order]> for Bill {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = orders.len(), "orders billed");
        ready(Ok(()))
    }
}

fn app() -> impl App<State = ()> {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        let orders = RedisStream::new("orders").group("billing");
        b.include(subscriber(orders, Bill).batch(nonzero!(16)).build());
        b.include(
            subscriber(RedisList::new("jobs"), Pack)
                .batch(nonzero!(16))
                .build(),
        );
        b.include(
            subscriber(RedisPubSub::new("clicks"), Count)
                .batch(nonzero!(16))
                .build(),
        );
    })
}

The list and the channel handlers:

Macros
Manual

examples/redis_subscriptions_batches.rs
/// Packs the jobs waiting on the list `jobs`, at most 16 at once.
#[subscriber(RedisList::new("jobs"))]
async fn pack(jobs: &[Order]) -> HandlerOutcome {
    tracing::info!(count = jobs.len(), "jobs packed");
    HandlerOutcome::ack()
}

/// Counts the clicks heard on the channel `clicks`, at most 16 at once.
#[subscriber(RedisPubSub::new("clicks"))]
async fn count(clicks: &[Order]) -> HandlerOutcome {
    tracing::info!(count = clicks.len(), "clicks counted");
    HandlerOutcome::ack()
}

examples/manual/redis_subscriptions_batches.rs
/// Packs the jobs waiting on the list `jobs`.
struct Pack;

impl Handle<[Order]> for Pack {
    fn handle(
        &self,
        jobs: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = jobs.len(), "jobs packed");
        ready(Ok(()))
    }
}

/// Counts the clicks heard on the channel `clicks`.
struct Count;

impl Handle<[Order]> for Count {
    fn handle(
        &self,
        clicks: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = clicks.len(), "clicks counted");
        ready(Ok(()))
    }
}

Example​

Example