Struct FramedRecvStream 

pub struct FramedRecvStream {
    inner: RecvStream,
    buf: BytesMut,
    draft: DraftVersion,
    subgroup_io: Option<SubgroupObjectReader>,
    fetch_io: Option<FetchObjectReader>,
    tracking: Option<(TrackObjects, u64)>,
}

A framed reader for a recv stream. Handles MoQT varint-length decoding.

Fields

inner: RecvStream

buf: BytesMut

draft: DraftVersion

subgroup_io: Option<SubgroupObjectReader>

Stateful subgroup object reader.

fetch_io: Option<FetchObjectReader>

Stateful fetch object reader, holding the prior Object’s Location and the Group Order its deltas count in. Started by FramedRecvStream::begin_fetch_objects rather than by the fetch header, which does not carry the order.

tracking: Option<(TrackObjects, u64)>

The record this stream’s objects are measured against, and the Group ID its header named.

One group for the whole stream: a subgroup header names it once and no object header repeats it. None on a stream that was never given one - a stream for an alias no live binding names, and every stream built outside Connection::accept_subgroup_stream.

Implementations

impl FramedRecvStream

pub fn new(inner: RecvStream, draft: DraftVersion) -> Self

Create a new framed receive stream for the given draft version.

pub fn stream_id(&self) -> u64

Get the transport-level stream ID.

fn measure_objects_against(&mut self, objects: TrackObjects, group: u64)

Measure this stream’s objects against objects, all of them in group.

fn note_subgroup_object( &self, object: u64, status: Option<u64>, ) -> Result<(), ConnectionError>

Judge one object this stream carried against where its track ended.

async fn fill(&mut self) -> Result<bool, ConnectionError>

Read more data from the stream into the internal buffer.

async fn ensure(&mut self, n: usize) -> Result<(), ConnectionError>

Ensure at least n bytes are available in the buffer.

pub async fn peek_stream_type(&mut self) -> Result<u64, ConnectionError>

Read this stream’s leading variable-length integer without consuming it, and return its value.

Every unidirectional MoQT stream on draft-18 opens with a varint naming what it is (Section 3.4): 0x05 for FETCH_HEADER, 0x10-0x1D for SUBGROUP_HEADER, and CONTROL_STREAM_TYPE for SETUP. Telling the peer’s control stream apart from a data stream means reading that varint, and taking it off the transport would destroy it: the control stream’s type varint is the SETUP message’s type field, so a stream whose type had been stripped would no longer decode as a SETUP.

Nothing is stripped. The bytes land in this reader’s own buffer, and every other method here — read_control, read_subgroup_header, read_fetch_header — decodes out of that buffer and advances it only on a successful decode. A stream this was called on is indistinguishable from one it was not, which is what makes it safe to peek a stream and then hand it to whichever reader the type turned out to call for.

It reads, so it can block: a peer that opens a stream and then writes nothing leaves this pending until a byte arrives or the stream ends.

Errors

• ConnectionError::UnexpectedEnd if the stream ends before a whole varint has arrived.
• ConnectionError::Transport if the peer reset the stream.
• ConnectionError::VarInt if the bytes are not a valid varint.

Whatever did arrive stays in the buffer in every case.

pub async fn read_control( &mut self, capture_raw: bool, ) -> Result<(AnyControlMessage, Option<Vec<u8>>), ConnectionError>

Read a control message from the stream.

When capture_raw is true, the returned tuple includes a clone of the framed wire bytes (for observer emission). When false, the second element is None and the payload clone is skipped.

pub async fn read_subgroup_header( &mut self, ) -> Result<AnySubgroupHeader, ConnectionError>

Read a subgroup stream header. Also initializes the internal delta-decoding state.

pub async fn read_fetch_header( &mut self, ) -> Result<AnyFetchHeader, ConnectionError>

Read a fetch response header.

pub async fn read_subgroup_object( &mut self, ) -> Result<SubgroupObject, ConnectionError>

Read the next draft-18 subgroup object from this stream using the stateful reader seeded by FramedRecvStream::read_subgroup_header.

Errors with ConnectionError::PropertiesOnNonNormalStatus on an Object that carries properties on a status other than Normal, which draft-18 Section 11.2.1.2 answers with a session close. The Object is consumed from the stream before the check, so the reader stays in step with the wire and a caller that reports the violation and reads on sees the following Object rather than a re-parse of this one.

pub async fn read_fetch_stream_header( &mut self, ) -> Result<FetchHeader, ConnectionError>

Read the next draft-18 fetch header from this stream.

pub fn stop(&mut self, code: u64) -> Result<(), ConnectionError>

Stop accepting data on this stream with code as the STOP_SENDING application error code, discarding anything unread.

Dropping a receive stream also stops it, but with a hard-coded 0. See RecvStream::stop.

pub async fn received_reset(&mut self) -> Result<Option<u64>, ConnectionError>

Wait for the peer to reset this stream, consuming nothing.

See RecvStream::received_reset for what Ok(None) means and why a caller must not re-poll after it.

pub fn begin_fetch_objects(&mut self, group_order: GroupOrder)

Start reading the objects of a fetch stream whose Groups arrive in group_order.

Section 11.4.4.1 makes a Group ID Delta count downward under Descending and upward under Ascending, so the same bytes are two different Locations and nothing on the data stream says which. The order is the one the request asked for — Section 10.12.3: “The publisher responding to a FETCH is responsible for delivering all available Objects in the requested range in the requested order” — carried by the GROUP_ORDER parameter on the FETCH, or by its absence, which Section 10.2.8 reads as Ascending. Either way it is a control message this stream never sees. Draft-19 encodes both IDs the same way and needs the same call; drafts 15, 16 and 17 resolve their fetch objects without an order, because none of those drafts encodes an ID as a difference.

Call it after FramedRecvStream::read_fetch_header and before the first FramedRecvStream::read_fetch_object; calling it again restarts the running state, which is what a second fetch stream on the same connection would want and what the middle of one would not.

pub async fn read_fetch_object( &mut self, ) -> Result<(FetchObject, Vec<u8>), ConnectionError>

Read the next draft-18 fetch object and its payload.

The mirror of FramedSendStream::write_fetch_object. What comes back is a FetchObject rather than a header: draft-18’s reader resolves the Location as it reads, and the resolved Group ID, Subgroup ID, Object ID and Priority are the fields a subscriber acts on. The header it decoded from is inside it.

The payload comes back with it because payload_length says how many bytes follow, and a reader that takes the wrong number of them desynchronises every later object on the stream.

Errors

ConnectionError::DataStreamState when FramedRecvStream::begin_fetch_objects has not been called, ConnectionError::UnexpectedEnd when the stream ends inside the header or inside the payload it declared, and ConnectionError::Codec on every rule Section 11.4.4.1 states — a first Object that inherits, a delta that runs off either end of the space, or an Object ID above 2^64-1.

async fn read_object_payload( &mut self, length: &VarInt, ) -> Result<Vec<u8>, ConnectionError>

Take the length payload bytes that follow a fetch object’s header.

Separate from the header read because the header is decoded from a probe cursor that may have to be retried after a fill, and the payload is a flat byte count that never is.

pub fn draft(&self) -> DraftVersion

Returns the draft version this stream is framed for.