Struct Buffered 

pub struct Buffered<S> { /* private fields */ }

A SubscriptionSource adapter that buffers the wrapped source’s subscriber into a BatchSubscriber.

A batch closes when it holds the size the registration asked for, or once max_wait has elapsed after its first delivery, whichever comes first; an idle subscription waits indefinitely for that first delivery. The default deadline is 10 ms.

Under the testing harness, injecting through tb.message(&value).publish() drives the whole reaction to a standstill before it returns, which closes the batch: one call per message yields one batch per message, each holding a single element. A test that wants a longer batch takes a producer handle off the broker before the app is built, publishes the run through it, and then drives the reaction once with tb.settle().

Examples

use std::time::Duration;

use ruststream::{Buffered, Name};

// What a broker crate wraps its own source in, so its subscriber batches on the client.
let source = Buffered::new(Name::new("orders")).max_wait(Duration::from_millis(20));

Implementations

impl<S> Buffered<S>

pub fn new(source: S) -> Self

Wraps source, batching its subscriber with the default deadline.

pub fn max_wait(self, max_wait: Duration) -> Self

Caps how long a partial batch waits for more deliveries after its first one.

Trait Implementations

impl<S: Clone> Clone for Buffered<S>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<S: Debug> Debug for Buffered<S>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<C, S> RedeliveryAddressed<C> for Buffered<S>

where Self: SubscriptionSource<C, Copies = AddressedCopies>, C: ConnectedBroker, S: RedeliveryAddressed<C> + Send + Sync, S::Subscriber: Send,

Batching happens on the client, so a retry copy goes where the wrapped source says it does.

fn redelivery_address( &self, connected: &C, ) -> impl Future<Output = Result<RedeliveryAddress, C::Error>> + Send

Where a publish reaches this subscription again.

impl<C, S> SubscriptionSource<C> for Buffered<S>

where C: ConnectedBroker, S: SubscriptionSource<C> + Send + Sync, S::Subscriber: Send,

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

Batching happens on the client, so the declaration reaches the wrapped source unchanged.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

The same for the half of it the broker takes: batching is this process’s business.

type Subscriber = BufferedSubscriber<<S as SubscriptionSource<C>>::Subscriber>

The subscriber type this source opens.

type Copies = <S as SubscriptionSource<C>>::Copies

Who publishes the copies this subscription’s retries are made of, and who names where they go: AddressedCopies, NamedCopies or BrokerMoves.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

async fn subscribe(self, connected: &C) -> Result<Self::Subscriber, C::Error>

Opens the subscription against the connected broker. Called once at startup.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.

fn operation_bindings(&self) -> Bindings

What this subscription adds to its receive operation in the document.

fn message_bindings(&self) -> Bindings

What this subscription adds to the messages that arrive on it.