Trait BatchSubscriber 

pub trait BatchSubscriber: Subscriber {
    type Batch: IntoIterator<Item = <Self as Subscriber>::Message> + Send;

    // Required method
    fn batches(
        &mut self,
        size: NonZeroUsize,
    ) -> impl Stream<Item = Result<Self::Batch, <Self as Subscriber>::Error>> + Send + '_;
}

A subscriber that delivers messages in batches.

Every broker offers this: those that batch on the wire (Kafka polls, JetStream pull consumers, Redis XREADGROUP COUNT) translate the batch size into their client’s own parameter, and those whose transport delivers one message at a time assemble batches on the client through the Buffered adapter. The batch a handler sees is exactly the batch the broker delivered: the runtime never splits or merges one.

Implementing it

The mount site names one number, the batch size, and the runtime passes it to batches unchanged. Everything else about how a batch forms (a block timeout, a consumer group, a prefetch window) is the broker’s own vocabulary on its subscription source, so a service writes the core’s word first and the broker’s after it: b.include(handler.batch(nonzero!(6)).block(Duration::from_secs(5))).

Put the capability on every subscriber a mount can reach. #[subscriber("topic")] opens Subscribe::Subscriber, so a &[T] body on that form asks for BatchSubscriber there too; a crate that wires it onto its descriptor’s subscriber alone leaves the string-literal form failing to compile.

A transport that delivers one message at a time implements it anyway, through Buffered. The size is not the broker’s to choose; the deadline that closes a partial batch is. Expose it on the subscription descriptor (.max_wait(..)) and hand it to the wrapper as the subscription opens: the wrapper’s 10 ms default suits an in-process bus, and the broker crates that sit behind a network round trip settle between 10 and 50 ms.

Declining the capability is legitimate where batching would break a guarantee the transport carries. A ZeroMQ ROUTER answers each peer at its own address, while a whole batch reaches its reply wiring with one context, so a batch’s replies would all go to one peer. A crate that declines says so in its docs: a &[T] body then does not compile on that transport.

conformance::capabilities::batches checks that a batch never comes back larger than the size it was opened with. It is not part of harness::run_suite: a crate calls it for this capability itself.

Required Associated Types

type Batch: IntoIterator<Item = <Self as Subscriber>::Message> + Send

Container yielded by batches. Implementations choose between Vec, custom iterators, or anything else that yields the underlying Subscriber::Message.

Required Methods

fn batches( &mut self, size: NonZeroUsize, ) -> impl Stream<Item = Result<Self::Batch, <Self as Subscriber>::Error>> + Send + '_

Returns a stream of batches of at most size messages.

size is the one parameter the framework passes down: it comes from the registration’s batch(n), which every batch handler names. Everything else about how batches are formed - a block timeout, a consumer group, a prefetch window - is the broker’s own business, configured on its subscription source.

A batch must never carry more than size messages; it may carry fewer (that is what a partial batch or a deadline is for). Implementations translate size into whatever their client already speaks (XREADGROUP COUNT, a pull batch size, a poll limit); one with no native batching wraps its subscriber in Buffered, which honours the size on the client.

Cancel safety

Same guarantees as Subscriber::stream: cancel-safe between polls.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<Log: LogMode> BatchSubscriber for MemorySubscriber<Log>

Greedy batching: a batch is the first awaited delivery plus everything already buffered, up to the size the registration asked for. Partial batches ship immediately, so no deadline timer is needed - the in-memory transport has nothing to wait for. Each batch is allocated at the requested size up front, so a partial batch holds the room of a full one.

type Batch = Vec<MemoryMessage<Log>>

impl<S: Subscriber> BatchSubscriber for BufferedSubscriber<S>

type Batch = Vec<<S as Subscriber>::Message>