Struct StreamHandle 

pub struct StreamHandle { /* private fields */ }

A handle to an inbound stream that can be cloned for multiple consumers.

Each clone maintains its own read position, allowing independent consumers to read the stream at different rates.

The design separates the lock-free buffer from synchronization state:

• buffer: Lock-free fragment storage using OnceLock<Bytes> slots
• sync: Cancelled flag and wakers (requires lock)

Fragment insertion goes directly to the buffer without acquiring any lock, and the buffer’s built-in Notify handles async notification.

Implementations

impl StreamHandle

pub fn new(stream_id: StreamId, total_bytes: u64) -> Self

Creates a new stream handle.

This is public for use by OrphanStreamRegistry which needs to create handles and for transport layer stream registration.

pub fn total_bytes(&self) -> u64

Returns the total expected bytes.

pub fn is_complete(&self) -> bool

Returns true if all fragments have been received.

pub fn received_fragments(&self) -> usize

Returns the number of fragments received so far.

pub fn total_fragments(&self) -> usize

Returns the total expected number of fragments.

pub fn stream(&self) -> StreamingInboundStream

Creates a new streaming view starting from fragment 1.

Each call creates an independent stream with its own read position. Fragments are cloned but remain in the buffer, allowing multiple consumers to read the same data via fork().

Memory Behavior

All fragments remain in the buffer until the stream is dropped or explicitly cleared. This means:

• Pro: Multiple consumers can read the same data independently
• Pro: Safe to use with fork() for parallel processing
• Con: Memory usage equals full stream size until completion

When to Use

Use stream() when:

• Multiple consumers need to read the same data
• You need to fork the stream for parallel processing
• Memory is not a concern for the stream size
• You might need to re-read fragments

For single-consumer scenarios with large streams, prefer stream_with_reclaim() for better memory efficiency.

pub fn stream_with_reclaim(&self) -> StreamingInboundStream

Creates a streaming view with automatic memory reclamation.

Unlike stream(), this version takes ownership of fragments as they are read, freeing memory progressively.

Memory Behavior

Each fragment is removed from the buffer immediately after being read:

• Pro: Memory usage stays constant regardless of stream size
• Pro: Ideal for processing large streams (10MB+) without OOM
• Con: Fragments cannot be read again once consumed
• Con: Incompatible with fork() or multiple consumers

When to Use

Use stream_with_reclaim() when:

• Only one consumer will read the stream
• The stream is large and memory is a concern
• Data is processed once and discarded (e.g., forwarding, hashing)
• You don’t need fork() or parallel consumers

Warning

Do not use with fork() or multiple consumers. Once a fragment is read by this stream, it is permanently removed from the buffer. Other consumers (including forked handles) will wait forever for fragments that have already been consumed.

Example

// Single consumer processing a large file
let mut stream = handle.stream_with_reclaim();
let mut hasher = Sha256::new();
while let Some(chunk) = stream.next().await {
    hasher.update(&chunk?);
    // Memory for this chunk is freed immediately
}

pub fn fork(&self) -> Self

Forks this handle, creating an independent consumer.

The forked handle shares the same underlying buffer but maintains its own read position when used with .stream().

pub fn try_assemble(&self) -> Option<Vec<u8>>

Assembles the complete data if all fragments are present.

This is a convenience method for waiting until the stream is complete and then getting all data at once.

pub async fn assemble(&self) -> Result<Vec<u8>, StreamError>

Waits for the stream to complete and returns the assembled data.

This is useful when you don’t need incremental processing. Uses the buffer’s built-in Notify for efficient async waiting.

Trait Implementations

impl Clone for StreamHandle

fn clone(&self) -> StreamHandle

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for StreamHandle

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

Formats the value using the given formatter.