Struct StreamReader 

pub struct StreamReader<R> { /* private fields */ }

Reads tensors from an async source in the Hurray streaming wire format.

Call next_tensor in a loop until it returns Ok(None) (clean EOF).

Examples

use hurray_io::stream::StreamReader;

let wire: &[u8] = &[]; // replace with actual data
let mut reader = StreamReader::new(wire);
while let Some(tensor) = reader.next_tensor().await? {
    println!(
        "tensor with {} buffer(s)",
        tensor.buffers.len()
    );
}

Implementations

impl<R: AsyncRead + Unpin> StreamReader<R>

pub fn new(inner: R) -> Self

Creates a reader with default options.

pub fn cross_machine(inner: R) -> Self

Creates a reader that enforces ProducerSynced on every buffer.

Use when the stream was produced by a remote machine and GPU/semaphore sync primitives are not meaningful locally.

pub fn with_options(inner: R, options: StreamReaderOptions) -> Self

Creates a reader with custom options.

pub async fn next_tensor(&mut self) -> Result<Option<StreamTensor>>

Reads the next tensor from the stream.

Returns Ok(None) on a clean EOF (no bytes remaining before a descriptor starts).

Errors

• Error::UnexpectedEof — stream ended mid-descriptor or mid-buffer
• Error::InvalidHeader — malformed descriptor prefix
• Error::FrameTooLarge — descriptor or buffer exceeds configured limit
• Error::InvalidCrossMachineSyncMode — cross-machine mode and a non-ProducerSynced buffer
• Error::Core — descriptor decode failed
• Error::Io — underlying read error

pub async fn next_item(&mut self) -> Result<Option<StreamItem>>

Reads the next item, assembling composites (head + members) into a StreamItem::Composite and validating them.

When the next descriptor on the wire is a composite head (layout_tag = 0x0B), this reads its declared member_count members — each of which may itself be a composite (recursion) — validates the group with [CompositeValidator] (member count, plus partition coverage / overlay ordering), and returns them together. Otherwise it behaves like next_tensor, returning a StreamItem::Tensor. Returns Ok(None) on a clean EOF.

Examples

use hurray_io::stream::{StreamItem, StreamReader};

let wire: &[u8] = &[]; // replace with actual data
let mut reader = StreamReader::new(wire);
while let Some(item) = reader.next_item().await? {
    match item {
        StreamItem::Tensor(t) => println!("tensor, {} buffer(s)", t.buffers.len()),
        StreamItem::Composite(c) => println!("composite, {} member(s)", c.members.len()),
    }
}

Errors

In addition to the errors of next_tensor:

• Error::TornComposite — the stream ended before the head’s member_count members were read
• Error::CompositeNestingTooDeep — nesting exceeded max_composite_depth
• Error::Core — composite validation failed (e.g. partition does not cover the index space, overlay ordering, member-count mismatch)

pub fn into_inner(self) -> R

Returns the underlying reader.