Struct BufferedStream 

pub struct BufferedStream<PB: CorrelatedBatch, E> { /* private fields */ }

Buffering over a correlation generator, providing both demand-driven streaming and proactive prefetching interfaces. The buffer is refilled in the background by a dispatcher/generator task pair; capacity bounds aggregate outstanding demand and refill_threshold drives proactive top-ups.

Implementations

impl<PB: CorrelatedBatch, E: From<CorrelatedStreamError> + Clone + Send + Debug + 'static> BufferedStream<PB, E>

pub fn new<G: CorrelationGenerator<PB> + Send + 'static>( generator: G, net: G::Net, config: BufferConfig, ) -> Self

where E: From<G::Error>,

Creates a new stream.

pub fn new_on_demand<G: CorrelationGenerator<PB> + Send + 'static>( generator: G, net: G::Net, ) -> Self

where E: From<G::Error>,

Creates a purely on-demand stream: no proactive refill and effectively unbounded admission, so every next_n triggers generation of exactly the shortfall (any surplus is buffered). Suited to streams whose generator is itself the buffer/limiter (e.g. a dealer client).

pub fn new_with_shared_config<G: CorrelationGenerator<PB> + Send + 'static>( generator: G, net: G::Net, config: Arc<RwLock<BufferConfig>>, ) -> Self

where E: From<G::Error>,

Creates a new stream.

pub async fn stop(self)

Shuts down the buffer gracefully and waits for all background tasks to exit.

Closing the command channel causes the dispatcher to exit, which in turn drops the work channel, causing the generator to exit. Both tasks are awaited before returning.

Trait Implementations

impl<PB: CorrelatedBatch, E> Buffer for BufferedStream<PB, E>

fn config(&self) -> &Arc<RwLock<BufferConfig>>

The shared configuration handle.

fn capacity(&self) -> Result<usize, CorrelatedStreamError>

fn max_request_size(&self) -> Result<usize, CorrelatedStreamError>

fn refill_threshold(&self) -> Result<usize, CorrelatedStreamError>

fn set_capacity(&self, capacity: usize) -> Result<(), CorrelatedStreamError>

fn set_max_request_size(&self, n: usize) -> Result<(), CorrelatedStreamError>

fn set_refill_threshold(&self, n: usize) -> Result<(), CorrelatedStreamError>

impl<PB: CorrelatedBatch, E: From<CorrelatedStreamError> + Clone + Send + Debug + 'static> CorrelatedStream<<PB as IntoIterator>::Item> for BufferedStream<PB, E>

type Error = E

The error type returned by the generator.

fn next_n( &self, n_elements: usize, ) -> Result<NextVec<PB::Item, E>, CorrelatedStreamError>

Returns a single future that resolves to all n_elements items at once. Can fail if request is too large/small or if the request cannot be fulfilled by the stream for some reason.

fn prefetch_n(&self, n_elements: usize) -> PrefetchHandle<E>

Proactively generates n_elements items into the buffer so later next_n calls need not wait on generation. Fire-and-forget (proceeds whether or not the handle is awaited); does not advance position. The default is a no-op.

fn position(&self) -> u64

The cumulative number of items delivered through next_n since creation (prefetching doesn’t count). Stays in sync across parties, so it can be exchanged to agree on a common resync target.

fn resync(&self, target: u64) -> ResyncHandle<E>

Advances to the absolute logical target by discarding intervening elements (draining the buffer, then skipping at the generator if supported) — the recovery primitive for non-fatal desyncs, where every party resyncs to the agreed max(position).