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;
use ruststream::memory::prelude::*;
use ruststream::runtime::{Declared, SubscriberBuilder};
use serde::Deserialize;

/// A broker crate's mount-site step for a transport that delivers one message at a time: the
/// batches are assembled on the client, and a partial one closes `wait` after it opened.
pub trait Linger {
    type Out;
    fn linger(self, wait: Duration) -> Self::Out;
}

impl<Def: Declared, State, DefCodec> Linger
    for SubscriberBuilder<Def, MemorySource, State, DefCodec>
{
    type Out = SubscriberBuilder<Def, Buffered<MemorySource>, State, DefCodec>;

    fn linger(self, wait: Duration) -> Self::Out {
        self.map_source(|source| Buffered::new(source).max_wait(wait))
    }
}

#[derive(Deserialize)]
struct Order {
    id: u64,
}

#[subscriber(MemorySource)]
async fn settle(orders: &[Order]) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "settled a batch");
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(
            settle
                .name("orders")
                .batch(nonzero!(64))
                .linger(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.