Batches

A slice handler on a queue or fan-out gets a batch assembled on this side, closed at the size the mount 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. A receive on a socket yields one multipart message, so a peer still sends one message at a time, in the same frames. The subscription collects the deliveries itself. A batch closes once it holds n deliveries, or 20 ms after its first one, whichever comes first. The 20 ms is a constant of the crate. So a busy socket fills batches to n, and a quiet one hands over batches of one, 20 ms after each message arrived.

One outcome settles every delivery of the batch, and a Vec<HandlerOutcome> answers them one by one. In the manual form the body is an impl Handle<[T]>, where Ok(()) settles the batch and Err(outcomes) settles it delivery by delivery.

A ZmqRpc responder takes no batches, and .batch(..) on one does not compile. A responder answers each request at the peer that sent it, while a batch carries one publish context for all of its replies. The compile error names ZmqRpcSubscriber, which implements no BatchSubscriber.

Pipelines with PUSH/PULL runs a sink that collects results in batches.

Example

Three jobs pushed at once arrive in one call. A job pushed alone arrives 20 ms after it was pushed.

Macros
Manual

examples/zmq_batches.rs
use ruststream_zeromq::queue::prelude::*;

/// Stores the jobs of one batch in a single round-trip.
#[subscriber("jobs")]
async fn drain(jobs: &[Job]) -> HandlerOutcome {
    tracing::info!(count = jobs.len(), "stored a batch of jobs");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
        |b| {
            // At most 32 jobs per call.
            b.include(drain.batch(nonzero!(32)));
        },
    )
}

examples/manual/zmq_batches.rs
use ruststream_zeromq::queue::prelude::*;

/// Stores the jobs of one batch in a single round-trip; `Ok` settles the whole batch.
struct Drain;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
        |b| {
            // At most 32 jobs per call.
            b.include(subscriber("jobs", Drain).batch(nonzero!(32)).build());
        },
    )
}

Example​

Example