Struct Connection 

pub struct Connection {
    transport: Transport,
    endpoint: Endpoint,
    draft: DraftVersion,
    control_send: Option<FramedSendStream>,
    control_recv: Option<FramedRecvStream>,
    observer: Option<Box<dyn ConnectionObserver>>,
    pending_events: Vec<ClientEvent>,
    deferred_uni: Mutex<VecDeque<FramedRecvStream>>,
    pending_inbound: Mutex<VecDeque<(FramedSendStream, FramedRecvStream)>>,
}

A live MoQT connection over QUIC or WebTransport, combining the endpoint state machine with actual network I/O.

Fields

transport: Transport

endpoint: Endpoint

draft: DraftVersion

control_send: Option<FramedSendStream>

control_recv: Option<FramedRecvStream>

observer: Option<Box<dyn ConnectionObserver>>

pending_events: Vec<ClientEvent>

Setup events buffered during connect() and replayed when an observer attaches via set_observer — without this, an observer attached after connect returns would never see the handshake.

deferred_uni: Mutex<VecDeque<FramedRecvStream>>

Unidirectional streams accepted while connect was looking for the peer’s control stream, in arrival order.

Data streams are allowed to arrive before the control streams on this draft, so the search cannot assume the first unidirectional stream is the control one — and dropping the ones that are not would silently lose objects the peer already sent. accept_subgroup_stream empties this before it accepts anything new.

Behind a mutex because that method takes &self. The lock is only ever held for a pop_front, never across an await.

pending_inbound: Mutex<VecDeque<(FramedSendStream, FramedRecvStream)>>

Bidirectional streams the peer opened that accept_request_stream took off the transport but did not finish reading a first message from, because its future was dropped. In arrival order.

Without this a caller could not put accept_request_stream in a select! at all: losing the race would lose a stream the peer had already opened and, with it, whatever of the request had arrived. accept_request_stream empties this before it accepts anything new.

Behind a mutex for the same reason deferred_uni is: the lock is only ever held for a push or a pop, never across an await.

Implementations

impl Connection

pub async fn connect( addr: &str, config: ClientConfig, ) -> Result<Self, ConnectionError>

Connect to a MoQT server as a client.

Establishes a QUIC or WebTransport connection (based on config.transport), brings up the control plane, performs the SETUP handshake, and returns a ready-to-use connection.

The control plane is a pair of unidirectional streams

Draft-20 Section 3.3: “MOQT uses a pair of unidirectional streams for creating the session and exchanging control messages. Each peer opens one control stream beginning with a SETUP message.” So each direction is a separate stream opened by the peer that writes on it. This opens one with open_uni and writes SETUP on it, then finds the peer’s by accepting unidirectional streams until one leads with CONTROL_STREAM_TYPE.

Nothing is written ahead of the SETUP: 0x2F00 is both the SETUP message type and the unidirectional stream type for a control stream, so the message’s own first field is the stream header. See CONTROL_STREAM_TYPE.

A bidirectional stream is not the control stream here — the same section makes it a request stream, one that begins with TRACK_STATUS, SUBSCRIBE, PUBLISH, FETCH, PUBLISH_NAMESPACE, SUBSCRIBE_NAMESPACE or SUBSCRIBE_TRACKS: “Bidirectional streams MUST NOT begin with any other message type unless negotiated. If they do, the peer MUST close the Session with a PROTOCOL_VIOLATION.” A SETUP written on a bidirectional stream is exactly that case, so a peer that enforces the topology answers it by closing the session.

Unidirectional streams that arrive before the peer’s control stream

They are kept, not dropped. Section 3.3 expects them: “Unidirectional streams containing Objects or bidirectional stream(s) beginning with a request message could arrive prior to the control streams, in which case the data SHOULD be buffered until both control streams arrive and setup is complete.” Each such stream is set aside and handed to accept_subgroup_stream in arrival order, ahead of any newly accepted stream. Only the leading type varint is read from them here; the rest stays on the transport, unread and still flow-controlled, so nothing is buffered in this process beyond those few bytes.

One limit worth knowing: the search waits for each stream’s type varint in turn, so a peer that opens a unidirectional stream and then writes nothing on it stalls the handshake behind that stream.

pub async fn adopt( transport: Transport, config: ClientConfig, ) -> Result<Self, ConnectionError>

Run the MoQT setup handshake over a transport somebody else established.

For choosing the draft from what the server selected: dial once through crate::transport::dial_quic offering every ALPN, then bring the connection to the module its answer names. Self::connect cannot do this — it derives its single ALPN from the draft it was given.

config.draft must match this module. The transport is adopted as given; nothing here re-checks the ALPN it was negotiated with.

async fn connect_quic( addr: &str, config: &ClientConfig, ) -> Result<Transport, ConnectionError>

Establish a raw QUIC connection.

Offers this draft’s ALPN alone; crate::transport::dial_quic holds the TLS and endpoint setup.

async fn connect_webtransport( _url: &str, _config: &ClientConfig, ) -> Result<Transport, ConnectionError>

Stub for when the webtransport feature is not enabled.

pub fn set_observer(&mut self, observer: Box<dyn ConnectionObserver>)

Attach an observer. Buffered handshake events from connect() are flushed in arrival order before this returns.

pub fn clear_observer(&mut self)

Remove the observer.

fn emit(&self, event: ClientEvent)

Emit an event to the observer, if one is attached.

pub async fn send_control( &mut self, msg: &ControlMessage, ) -> Result<(), ConnectionError>

Send a control message on the control stream.

Wraps the draft-20 message in AnyControlMessage::Draft20 for framing. Of the messages draft-20 Table 5 does not place on a request stream, exactly two reach the wire: SETUP, written by connect as the control stream’s own type varint, and GOAWAY, which is the one message the table gives the Stream value “Control, Request” and which belongs here when it drains the session.

Four messages that this still accepts and should not

Table 5 gives NAMESPACE (0x8), NAMESPACE_DONE (0xE), PUBLISH_SKIPPED (0xF) and REQUEST_UPDATE (0x2) the Stream value Request, not Control. REQUEST_UPDATE is sent “on the same bidi stream as the request” it modifies (Section 10.9); the other three report on the namespace or the publication of the request stream they arrive on. A conforming peer answers any of the four on the control stream by closing the session, and so does this client’s own Endpoint::receive_message — writing one here produces a frame this implementation would itself refuse.

All four now have somewhere better to go, and none of them should come through here. REQUEST_UPDATE goes on the stream its request opened, with send_on_request_stream. NAMESPACE and NAMESPACE_DONE go on a peer’s SUBSCRIBE_NAMESPACE stream, with namespace_on and namespace_done_on; PUBLISH_SKIPPED goes on a peer’s SUBSCRIBE_TRACKS stream, with publish_skipped_on. Those three arrived with the accept path — before it there was no peer-opened stream to put them on, which is why this method still takes them rather than refusing them outright. Narrowing what it accepts changes an existing outbound route and is not part of accepting requests.

GOAWAY is correct here and is not confined here. Draft-18 gave it an optional Request ID present only on the control stream, and draft-20 removed the field, so its control-stream and request-stream forms are now one wire form; send_on_request_stream will put it on an open request stream.

Requests are refused here

Draft-20 Section 3.3 keeps requests off the control plane: “In addition to the control streams, this specification uses bidirectional streams to carry requests. A request stream begins with one of these seven message types: TRACK_STATUS, SUBSCRIBE, PUBLISH, FETCH, PUBLISH_NAMESPACE, SUBSCRIBE_NAMESPACE, and SUBSCRIBE_TRACKS.” The response comes back on that same bidirectional stream, and resetting it cancels the request.

Handing one of those seven to this method returns ConnectionError::RequestOnControlStream and writes nothing — an enforcing peer sees no bytes at all, not a misplaced request. Use the typed helpers, which open a bidirectional stream each: subscribe, fetch, publish, track_status, publish_namespace, subscribe_namespace and subscribe_tracks.

Response types are still permitted, and no longer need to be. A response written here will be refused by a conforming peer, whose endpoint answers a response on the control stream with an error; the route that is correct is accept_request_stream and the respond_* helpers, which write on the stream the request arrived on and take the Request ID off the handle rather than from the caller. publish_done and publish_done_on do not come through here either: each takes the request stream its publication travels on.

pub async fn recv_control(&mut self) -> Result<ControlMessage, ConnectionError>

Read the next control message from the control stream.

Returns the AnyControlMessage and also extracts the draft-20 ControlMessage for internal endpoint dispatch.

pub async fn recv_and_dispatch( &mut self, ) -> Result<ControlMessage, ConnectionError>

Read and dispatch the next incoming control message through the endpoint state machine. Returns the decoded message for inspection.

Responses never arrive here. Draft-20 responses carry no request id and belong on the request stream that asked for them, so the endpoint refuses a response that turns up on the control stream. Read them with recv_on_request_stream.

A violation the draft answers with a session close is closed on the wire here, before the error is returned: the QUIC connection is closed with the code EndpointError::session_error_code names, so the peer that broke the rule learns of it rather than only the local endpoint.

async fn open_request_bi( &self, ) -> Result<(FramedSendStream, FramedRecvStream), ConnectionError>

Open the bidirectional stream a request will be carried on.

Opened before the endpoint allocates a request id, so a transport that refuses a new stream — the peer’s initial_max_streams_bidi is exhausted, the connection is gone — costs nothing. The endpoint has no way to abandon a request it has already allocated, so every failure that can be moved ahead of the allocation is.

Nothing is written here. A request stream carries no stream-type header: its first field is the leading message’s own type field, which is what begin_request writes.

fn or_abandon<T>( halves: &mut (FramedSendStream, FramedRecvStream), built: Result<T, EndpointError>, ) -> Result<T, ConnectionError>

Reset a request stream that was opened but whose request could not be built, and pass the endpoint’s error through.

Without this, an endpoint refusal — the session is draining, the request-id range is exhausted — would leave a bidirectional stream open that never carries a first message, and dropping it would FIN it, telling the peer an empty stream ended cleanly.

async fn begin_request( &mut self, halves: (FramedSendStream, FramedRecvStream), kind: RequestKind, request_id: VarInt, msg: &ControlMessage, ) -> Result<RequestStream, ConnectionError>

Write msg as the first message on an opened bidirectional stream and hand back the RequestStream that owns both halves.

This is the one place a request reaches the wire. Every request helper funnels through it, so the ordering — open, allocate, write, emit — is stated once.

A failed write resets both halves rather than leaving a half-written request stream behind. What it cannot undo is the endpoint’s allocation: the request id and its state machine already exist, and there is no way to retract them, so a write that fails here leaves one pending request the endpoint will never see answered.

pub async fn recv_on_request_stream( &mut self, stream: &mut RequestStream, ) -> Result<ControlMessage, ConnectionError>

Read the next message off a request stream and dispatch it through the endpoint with that stream’s own request id.

On draft-20 a response carries no request id; the stream is the correlation, so the id comes from the handle and not from the wire.

This blocks until a whole message has arrived. Backpressure is per request: a stream nobody reads stays unread, and the peer stays flow controlled on it alone. A peer that reset the stream surfaces as ConnectionError::Transport carrying TransportError::StreamReset with the peer’s code; a caller that is deliberately not reading should watch RequestStream::peer_cancelled instead.

Errors

ConnectionError::Endpoint if the message is not one of this draft’s response types, or if it does not fit the request’s state. The message has already been emitted to the observer by then — what arrived is reported whether or not the endpoint accepts it.

pub async fn send_on_request_stream( &mut self, stream: &mut RequestStream, msg: &ControlMessage, ) -> Result<(), ConnectionError>

Write a follow-up message on an already-open request stream.

The request itself was written when the stream was opened; this is for what comes after it on the same stream, PUBLISH_DONE among them — see publish_done, which uses this.

It does not refuse any message type. Which messages may follow a request on its own stream is not something this implementation can settle, so the choice is left to the caller rather than guessed at.

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

Cancel a request: record it at the endpoint, then terminate its stream. Section 3.3.3 puts the cancel at the stream — “Implementations cancel a request by abruptly terminating any directions of the stream that are still open” — while the request’s own state lives in the endpoint, so the two have to move together. This is the only place that moves both.

The endpoint goes first and the stream is terminated only if it agrees, which is the order every request path here uses: a caller acts on a stream after the endpoint has accepted the step, never before. A refused cancel therefore leaves the stream exactly as it was, and RequestStream::cancel is still there for a caller that wants the stream reset regardless.

Idempotent from both ends: a request that has already ended accepts the cancel and stays where it is, and a handle that has already been cancelled resets nothing a second time.

Errors

ConnectionError::Endpoint if no request carries this stream’s id or the request has not been written, and ConnectionError::Transport if code is outside the QUIC varint range — see RequestStream::cancel, which is what sends it.

pub async fn peer_cancelled_on_request_stream( &mut self, stream: &mut RequestStream, ) -> Result<Option<u64>, ConnectionError>

Wait for the peer to cancel this request, and record it if it does.

RequestStream::peer_cancelled with the endpoint’s record attached. A caller applying backpressure is deliberately not calling recv_on_request_stream, which is the other place a peer reset surfaces, so without this the request would end on the wire and stay open in the endpoint’s record for as long as the backpressure lasts.

Returns what the handle’s own method returns; see it for the Ok(None) case and for what WebTransport can and cannot observe. Cancel-safe, and it grants no flow-control credit.

pub async fn accept_request_stream( &mut self, ) -> Result<(ControlMessage, RequestStream), ConnectionError>

Accept the next bidirectional stream the peer opened, read the request it begins with, and hand back that request and a handle to answer it on.

This is the mirror of the request helpers. Where subscribe and its siblings open a stream and write a request, this takes one the peer opened and reads one. Draft-20 Section 3.3 puts requests in both directions on bidirectional streams, so a client that only ever calls the helpers can never be published to or subscribed from.

The returned RequestStream carries RequestOrigin::Peer. Answer it with respond_subscribe_ok, respond_fetch_ok, respond_ok or respond_error, and hold it for as long as the request lasts — a subscription’s PUBLISH_DONE is written on it, and dropping it resets the stream.

Two refusals, two codes

Draft-20 Section 3.3, on a stream that begins with the wrong type: “Bidirectional streams MUST NOT begin with any other message type unless negotiated. If they do, the peer MUST close the Session with a PROTOCOL_VIOLATION.” Section 10.1, on the Request ID: “If an endpoint receives a Request ID where the least significant bit is incorrect for the sender, or a duplicate Request ID, it MUST close the session with INVALID_REQUEST_ID.” Both are closes of the session on the wire, with different codes, and both happen before this returns — the error handed back reports a session that is already gone, not one the caller must remember to close.

Cancelling this future loses nothing

A stream taken off the transport but not yet read is put back on an internal queue, and the next call takes it before accepting anything new — including whatever bytes of the request had already arrived, which live in the stream’s own reader. So this is safe to select! against a shutdown signal or a timer. See pending_inbound_count.

What it is not safe to do is run concurrently with another method on the same connection: this takes &mut self because registering the peer’s request moves endpoint state, and no signature avoids that while the connection owns the endpoint. A caller blocked in recv_on_request_stream waiting for its own response is not accepting, and the peer’s request streams queue up in the transport behind it. One loop that never blocks indefinitely on a single read is the shape this supports.

Ordering

The endpoint is told about the request last, after every step that can fail or be cancelled, and building the handle afterwards cannot fail. This is the inverse of the outbound path’s reasoning — it opens the stream before allocating a Request ID for the same reason — and rests on the same fact: the endpoint has no way to abandon a request it has already registered. Registering earlier would let a cancelled accept leave a state machine keyed to a stream nobody holds, and the peer’s next use of that Request ID would then be reported as a duplicate — a session close, over an id the peer used exactly once.

Errors

• ConnectionError::NonRequestOnRequestStream — the session has been closed with PROTOCOL_VIOLATION and the stream reset.
• ConnectionError::Endpoint carrying RequestId or DuplicateRequestId — the session has been closed with INVALID_REQUEST_ID and the stream reset.
• ConnectionError::Endpoint carrying NotActive or Draining — the stream is reset, the session is left alone.
• ConnectionError::Transport or ConnectionError::Codec — the stream is reset, the session is left alone.

fn take_pending_inbound(&self) -> Option<(FramedSendStream, FramedRecvStream)>

Take the oldest stream pair a cancelled accept_request_stream put back, if any.

Synchronous on purpose, like take_deferred_uni: the guard is dropped before the caller awaits, so the lock is never held across a suspension point.

pub fn pending_inbound_count(&self) -> usize

How many peer-opened request streams a cancelled accept_request_stream put back and a later call has not yet taken.

Zero unless an accept future was dropped mid-read.

async fn respond( &mut self, stream: &mut RequestStream, msg: ControlMessage, fin: bool, ) -> Result<(), ConnectionError>

Write msg on the request stream stream carries, driving the endpoint first and the wire second.

The response goes on the request’s own bidirectional stream and never on the control stream: draft-20 responses carry no Request ID, so the stream is the only thing that says what is being answered. Taking the id off the handle rather than from the caller makes that correlation unforgeable.

fin is true only for REQUEST_ERROR. See respond_error.

pub async fn respond_ok( &mut self, stream: &mut RequestStream, response: RequestOk, ) -> Result<(), ConnectionError>

Answer a peer’s PUBLISH, PUBLISH_NAMESPACE, SUBSCRIBE_NAMESPACE, SUBSCRIBE_TRACKS or TRACK_STATUS with REQUEST_OK.

Draft-18 folded PUBLISH_OK into REQUEST_OK and draft-20 keeps the fold, so this is the one success message for five of the seven request kinds — there is no respond_publish_ok here, where draft-17 has one.

The send half is left open. A SUBSCRIBE_NAMESPACE responder still owes the peer the namespaces it accepted, so finishing here would end the request before it had been served; a TRACK_STATUS responder owes nothing further and may call RequestStream::finish straight after.

Errors

ConnectionError::RespondedToOwnRequest if stream is one this endpoint opened, and ConnectionError::Endpoint if no request of a kind REQUEST_OK answers is pending on it — including the Section 10.5 case of Track Properties on anything but a TRACK_STATUS response. Nothing is written in any of them.

pub async fn respond_subscribe_ok( &mut self, stream: &mut RequestStream, response: SubscribeOk, ) -> Result<(), ConnectionError>

Answer a peer’s SUBSCRIBE with SUBSCRIBE_OK.

The send half is left open, and it must be: this endpoint is now the publisher of an established subscription and owes it a PUBLISH_DONE, which travels on this same stream — publish_done_on. Until that has been sent, RequestStream::finish refuses, per draft-20 Section 3.3.2.

pub async fn respond_fetch_ok( &mut self, stream: &mut RequestStream, response: FetchOk, ) -> Result<(), ConnectionError>

Answer a peer’s FETCH with FETCH_OK.

The send half is left open. The fetched objects travel on separate unidirectional streams, so a fetch responder may call RequestStream::finish as soon as this returns; it is not done here because nothing about FETCH_OK says the responder has no more to write.

pub async fn respond_error( &mut self, stream: &mut RequestStream, response: RequestError, ) -> Result<(), ConnectionError>

Reject a peer’s request with REQUEST_ERROR, and finish the send half.

The FIN is part of the act, not a convenience: draft-20 Section 3.3.3 says “When an endpoint rejects a request without performing any application processing, it SHOULD send a REQUEST_ERROR and FIN the stream.” It is also the one response that can be finished immediately, because a rejected request leaves nothing further to send — every success path owes the peer something more, which is what Section 3.3.2 makes a MUST NOT for the rest of them.

A finished handle does nothing further on Drop, so the rejected stream is not then reset.

pub async fn publish_done_on( &mut self, stream: &mut RequestStream, status_code: VarInt, stream_count: VarInt, reason_phrase: Vec<u8>, ) -> Result<(), ConnectionError>

End a subscription this endpoint accepted, on the stream the peer’s SUBSCRIBE opened.

The mirror of publish_done, which ends a publication this endpoint offered with PUBLISH. Both write PUBLISH_DONE on a request stream and take the Request ID off the handle; they differ in which state machine moves, and therefore in which one refuses.

The send half is finished once the message is out, for the reason publish_done gives: Section 3.3.2 names PUBLISH_DONE as the last required message on a publisher’s direction and asks for the FIN promptly after it.

pub async fn namespace_on( &mut self, stream: &mut RequestStream, response: Namespace, ) -> Result<(), ConnectionError>

Announce a namespace on a peer’s SUBSCRIBE_NAMESPACE stream.

Draft-20 Section 10.17 puts NAMESPACE “on the response stream of a SUBSCRIBE_NAMESPACE request”, and Table 5 gives it the Stream value Request rather than Control — which is why this exists at all and why send_control is the wrong route for it. Section 10.18 has the announcements follow the acceptance, so this refuses until respond_ok has run on the same stream.

pub async fn namespace_done_on( &mut self, stream: &mut RequestStream, response: NamespaceDone, ) -> Result<(), ConnectionError>

Withdraw a namespace on a peer’s SUBSCRIBE_NAMESPACE stream.

Section 10.19: “The publisher MUST NOT send NAMESPACE_DONE for a namespace suffix before the corresponding NAMESPACE.” Which suffixes have been announced is the caller’s bookkeeping, not this connection’s, so what is enforced here is only that the request was accepted first.

pub async fn publish_skipped_on( &mut self, stream: &mut RequestStream, response: PublishSkipped, ) -> Result<(), ConnectionError>

Tell the peer a track matching its SUBSCRIBE_TRACKS will not be published, on that request’s own stream.

Section 10.21: “All PUBLISH_SKIPPED messages are in response to a SUBSCRIBE_TRACKS,” which is why this refuses any other kind of stream — the endpoint looks the Request ID up among the SUBSCRIBE_TRACKS requests alone.

pub async fn subscribe( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, parameters: Vec<KeyValuePair>, ) -> Result<RequestStream, ConnectionError>

Send a SUBSCRIBE on a bidirectional stream of its own.

The returned RequestStream is where SUBSCRIBE_OK, REQUEST_ERROR and later PUBLISH_DONE arrive — read them with recv_on_request_stream. Hold it for the subscription’s life: dropping it resets the stream, which cancels the subscription.

pub async fn fetch( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, parameters: Vec<KeyValuePair>, ) -> Result<RequestStream, ConnectionError>

Send a FETCH on a bidirectional stream of its own.

FETCH_OK or REQUEST_ERROR comes back on the returned RequestStream; the fetched objects arrive on separate unidirectional data streams. Dropping the handle cancels the fetch.

The range is a parameter now

Draft-20 Section 10.13 rebuilt FETCH: there is no Fetch Type, no Standalone or Joining structure, and no inline Start Location / End Location pair. The range travels in the LOCATION_FILTER parameter, so this takes the same three arguments subscribe does. fetch_range is the same call with the filter built for you; a FETCH with no filter covers {0,0} through Largest Object, inclusive.

pub async fn fetch_range( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, range: &LocationFilter, parameters: Vec<KeyValuePair>, ) -> Result<RequestStream, ConnectionError>

Send a FETCH for one range, carried as a LOCATION_FILTER parameter.

The range is inclusive at both ends. Draft-19’s End Location was “the last Object, plus 1; or 0 to indicate the entire Group”; draft-20 Sections 5.1.2 and 10.13 both call the filter’s range inclusive and delete both conventions, and nothing here restores them. Port a draft-19 call by removing the + 1.

pub async fn subscribe_namespace( &mut self, namespace_prefix: TrackNamespace, parameters: Vec<KeyValuePair>, ) -> Result<RequestStream, ConnectionError>

Send a SUBSCRIBE_NAMESPACE on a bidirectional stream of its own.

Draft-18 split the draft-17 SUBSCRIBE_NAMESPACE into two messages and draft-20 keeps the split. This call sends the renumbered SUBSCRIBE_NAMESPACE (type 0x50), which subscribes to NAMESPACE / NAMESPACE_DONE announcements only. To receive PUBLISH messages for matching tracks, use Self::subscribe_tracks instead — a separate request on a request stream of its own.

pub async fn subscribe_tracks( &mut self, namespace_prefix: TrackNamespace, parameters: Vec<KeyValuePair>, ) -> Result<RequestStream, ConnectionError>

Send a SUBSCRIBE_TRACKS (type 0x51, new in draft-18) on a bidirectional stream of its own. Causes the relay to PUBLISH matching tracks back to us.

Draft-17 has no such message: this is the seventh request type, and the one that makes draft-20’s set differ from draft-17’s.

pub async fn publish_namespace( &mut self, track_namespace: TrackNamespace, parameters: Vec<KeyValuePair>, ) -> Result<RequestStream, ConnectionError>

Send a PUBLISH_NAMESPACE on a bidirectional stream of its own.

pub async fn track_status( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, parameters: Vec<KeyValuePair>, ) -> Result<RequestStream, ConnectionError>

Send a TRACK_STATUS on a bidirectional stream of its own.

pub async fn publish( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, track_alias: VarInt, parameters: Vec<KeyValuePair>, track_properties: Vec<KeyValuePair>, ) -> Result<RequestStream, ConnectionError>

Send a PUBLISH on a bidirectional stream of its own.

REQUEST_OK — which is what draft-20 answers a PUBLISH with, PUBLISH_OK having been folded into it — or REQUEST_ERROR comes back on the returned RequestStream, and publish_done is written back on it when the publication ends, so the handle must be held for as long as the publication lasts.

pub async fn publish_done( &mut self, stream: &mut RequestStream, status_code: VarInt, stream_count: VarInt, reason_phrase: Vec<u8>, ) -> Result<(), ConnectionError>

Send a PUBLISH_DONE on the request stream the PUBLISH opened.

PUBLISH_DONE is a response and carries no request id on the wire, so the stream is the only thing that says which publication ended. The id the endpoint needs is taken off stream, which makes the correlation unforgeable — there is no way to name one request and write on another’s stream.

The send half is finished once the message is out. Draft-20 Section 3.3.2 names PUBLISH_DONE as the last required message on a publisher’s direction — “the publisher of an Established subscription MUST send PUBLISH_DONE, before sending a FIN” — and asks for the FIN promptly afterwards, since nothing further is owed on that direction. Leaving it open instead would end the publication at Drop, which resets the stream with REQUEST_CANCELLED and tells the peer the request was abandoned rather than completed.

The receive half stays open. A FIN closes one direction, so the subscriber can still be heard from on its own.

pub async fn open_subgroup_stream( &self, header: &AnySubgroupHeader, ) -> Result<FramedSendStream, ConnectionError>

Open a new unidirectional stream for sending subgroup data.

pub async fn open_fetch_stream( &self, header: &AnyFetchHeader, ) -> Result<FramedSendStream, ConnectionError>

Open a new unidirectional stream for sending a FETCH’s objects.

The objects answering a FETCH do not go on the request’s own stream: they go on a unidirectional stream of their own, which opens with a FETCH_HEADER naming the request they belong to. This writes that header and hands back the stream, the same way open_subgroup_stream does for a subgroup.

The caller owns the stream that comes back. Nothing here remembers which request it belongs to, so an endpoint serving several fetches at once keeps its own map from Request ID to stream.

pub async fn open_fetch_stream_on<'s>( &self, stream: &'s mut RequestStream, header: &AnyFetchHeader, ) -> Result<&'s mut FramedSendStream, ConnectionError>

Open the data stream for a FETCH this endpoint is answering, and keep the handle on the request.

The same stream open_fetch_stream returns, parked on the request stream it belongs to. That is what lets a rule about the fetch reach the objects it is serving: a refused REQUEST_UPDATE has to reset this stream, and the connection cannot reset a handle the caller walked away with.

Write objects through RequestStream::fetch_data, or through the borrow this returns.

pub async fn accept_subgroup_stream( &self, ) -> Result<(AnySubgroupHeader, FramedRecvStream), ConnectionError>

Accept an incoming unidirectional data stream and read its subgroup header.

Streams the peer opened before its control stream are returned first, in arrival order, before any new one is accepted from the transport: connect had to look at them to find the control stream and set the rest aside rather than drop them. They are otherwise ordinary — the type varint connect read is still on the front of each one.

pub async fn accept_fetch_stream( &self, ) -> Result<(AnyFetchHeader, FramedRecvStream), ConnectionError>

Accept the next unidirectional stream and read its fetch header.

accept_subgroup_stream’s twin. The two are separate because the header decides how every object after it is framed, so a caller has to know which it is expecting before the first byte is read.

Objects come off the returned stream with FramedRecvStream::read_fetch_object.

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

Accept the next unidirectional stream as a fill fetch stream, new in draft-20 (Section 5.1.3).

A fill fetch stream is framed exactly as a fetch response — a FETCH_HEADER and the fetch object records after it, read with FramedRecvStream::read_fetch_object — and differs only in what its Request ID names: the SUBSCRIBE (or the REQUEST_UPDATE, which travels under the same Request ID) that asked for the fill, rather than a FETCH. So this is accept_fetch_stream with the attribution done: the endpoint is asked whether that request asked for a fill, the stream is counted against the subscription, and the event goes out under StreamKind::Fill.

The object reader is already started

Unlike accept_fetch_stream, the returned stream needs no begin_fetch_objects call: objects can be read straight off it. A fetch stream cannot be started here because the Group Order lives on the FETCH the caller sent and nothing on this side holds it; a fill stream can, because the endpoint kept the order when the subscription asked for the fill — see Endpoint::fill_group_order and fill::group_order for the three-step resolution Sections 5.1.3 and 10.2.8 define.

Doing it here rather than leaving it to the caller is the difference between an error and a wrong answer: a reader started in the wrong direction still parses every frame, and reports Group IDs that walk the wrong way. A caller that has since learned better restarts it by calling begin_fetch_objects itself before reading the first object.

What ends it, and what that does not do

Section 5.1.3.1: a FIN means the whole fill range was delivered, and a reset means the fill failed — “Because there is no REQUEST_ERROR associated with a fill fetch stream, the publisher signals a fill failure by resetting the stream”. Neither touches the subscription: “Resetting or cancelling a fill fetch stream, by either endpoint, does not affect the subscription, which continues to deliver objects using subscribe subgroups and datagrams.” Tell the endpoint which happened with Endpoint::on_fill_stream_ended; a subscriber that wants to stop one early sends STOP_SENDING with FramedRecvStream::stop, which Section 5.1.3.1 gives it outright.

Errors

ConnectionError::Endpoint carrying EndpointError::UnrequestedFillStream when the header names a request that asked for no fill. Not fatal to the session — draft-20 states no rule for the case, and the stream is the only thing wrong.

fn take_deferred_uni(&self) -> Option<FramedRecvStream>

Take the oldest stream connect set aside, if any.

Synchronous on purpose: the guard is dropped before the caller awaits, so the lock is never held across a suspension point. A poisoned lock is recovered rather than propagated — nothing here can leave the queue in a state a later reader could be misled by, since the only mutation is a pop_front.

pub fn deferred_stream_count(&self) -> usize

How many unidirectional streams connect set aside and accept_subgroup_stream has not yet handed back.

Zero for a peer that opened its control stream first, which is the ordinary case.

pub fn send_datagram( &self, header: &AnyDatagramHeader, payload: &[u8], ) -> Result<(), ConnectionError>

Send an object via datagram.

A header whose object status the framing cannot carry is refused rather than sent. A datagram writes a status field only when its type byte sets the STATUS bit, so a header carrying End of Group under a type byte without that bit would go out as an ordinary payload object with the marker missing — a datagram the peer cannot tell from one that never had a status. AnyDatagramHeader::encode answers that with CodecError::InvalidField and writes nothing, on every draft.

pub async fn recv_datagram( &self, ) -> Result<(AnyDatagramHeader, Bytes), ConnectionError>

Receive a datagram and decode its header.

Errors with ConnectionError::PropertiesOnNonNormalStatus on a draft-20 datagram carrying properties on a status other than Normal. Draft-20 Section 11.3.1 builds the datagram’s Properties field out of the structure defined in Section 11.2.1.2, and that section answers the combination with a session close — the same rule the subgroup form obeys, on the same receive side.

The event is emitted before the check, so an observer of the client’s event stream still sees the datagram that caused the error.

pub fn endpoint(&self) -> &Endpoint

Access the underlying endpoint state machine.

pub fn endpoint_mut(&mut self) -> &mut Endpoint

Mutable access to the endpoint state machine.

pub fn draft(&self) -> DraftVersion

Returns the draft version this connection is using.

fn close_for(&self, err: &EndpointError)

Close the session on the wire when the endpoint says a violation is fatal to it, and hand the error back unchanged.

EndpointError::session_error_code answers Some for exactly the errors draft-20 tells the receiver to close the session over, and the endpoint has already moved its own state machine to Closed by the time this runs. Without this step that move was purely internal: the local endpoint refused to start anything new while the peer, which is the one that broke the rule, saw a session that was still open and went on sending. “MUST close the session with a PROTOCOL_VIOLATION” is a statement about the wire, so it takes a CONNECTION_CLOSE to satisfy it.

The reason phrase is the error’s own Display text, which names the message and the rule rather than repeating the numeric code the close already carries.

Errors that answer None are recoverable and nothing is sent.

fn close_if_session_fatal(&self, err: EndpointError) -> ConnectionError

close_for, then the error unchanged, for the common case where the endpoint’s error is also what the caller returns.

fn codec_session_error_code(err: &CodecError) -> Option<SessionErrorCode>

The code to close the session with when a control message could not be decoded because the peer broke a rule draft-20 answers with a close.

Every variant listed here comes from a sentence in the draft that names the consequence: the reason phrase and GOAWAY URI maxima (Sections 1.4.4 and 10.4), the KVP value maximum and the delta-encoded type overflow (Section 1.4.3), the duplicate-parameter rule (Section 10.2), the Track Namespace field, count and length rules (Section 2.4.1), and the Object ID delta wrap (Section 11.4.2). Each of those reads “MUST close the session with a PROTOCOL_VIOLATION”.

The Object ID wrap is the one that arrives here from a data stream rather than a control message, and it is drafts 18, 19 and 20 only: “The Object ID Delta + 1 is added to the previous Object ID in the Subgroup stream if there was one… If the resulting Object ID would be greater than 2^64 - 1, the endpoint MUST close the session with a PROTOCOL_VIOLATION.” Draft-17 describes the same arithmetic and states no consequence for overflowing it, so its connection deliberately leaves the wrap off this list and treats it as a decode failure alone.

One more rule reaches this table without naming a code: “An endpoint that receives an unknown message type MUST close the session”, stated in those words by all fourteen drafts. Protocol Violation is what carries it, as it does on every draft below this one.

None for everything else, including CodecError::InvalidField. That variant is shared by a dozen unrelated malformations, only some of which the draft answers with a close, so treating it as fatal would close sessions the draft does not ask to be closed. Splitting it is the way to bring the rest of those rules under this function; widening the match is not — which is exactly how the Object ID wrap arrived, as its own variant rather than as one more reading of InvalidField.

fn close_for_codec(&self, err: ConnectionError) -> ConnectionError

Close the session on the wire when a decode failure is one draft-20 answers with a close, and hand the error back unchanged.

The codec’s counterpart to close_if_session_fatal. Without it every bound the decoder enforces would stop at this endpoint refused the frame while the peer, which is the one that broke the rule, saw a session that was still open and went on sending. “MUST close the session with a PROTOCOL_VIOLATION” is a statement about the wire.

pub fn requests_to_cancel(&self, err: &ConnectionError) -> Vec<VarInt>

Name every request whose stream the caller must reset, for a track a data path has just found malformed.

Section 2.4.2 answers its whole list of conditions at once: “it MUST cancel any corresponding subscription or fetches for that Track from that publisher”. On this draft cancelling a request is a transport operation rather than a message — Section 3.3.3: “Implementations cancel a request by abruptly terminating any directions of the stream that are still open, using RESET_STREAM for a direction they are sending and STOP_SENDING for a direction they are receiving.”

No RFC 2119 keyword on this draft. Drafts 17 and 18 say implementations SHOULD cancel this way; draft-20 states it as what cancelling is, and adds the case of an endpoint that has already sent a FIN and sends STOP_SENDING alone.

Why this returns ids instead of doing it

Because the streams are the caller’s. Every request on this draft lives at the front of a bidirectional stream of its own, and Connection::recv_on_request_stream hands that stream back as a RequestStream. There is no handle here to reset. So the connection does the half it can — note the track, and work out which requests receive it — and the caller passes each id to Connection::cancel_request_stream, which resets the stream and moves the endpoint’s record.

This is the one place in this crate where the two halves of an answer are split across the API boundary, and it is the draft that splits them: drafts 12 through 16 answer with a control message, which the connection owns, so withdraw_for_data_stream there does the whole thing.

Both data paths come here

Unlike the drafts that answer with a message, where a datagram is read through the connection and answers itself. Here neither path can, for the same reason, so there is one entry point rather than two. Pass it whatever error a read returned; anything that is not this condition gives back an empty list.

Empty is not “the track was fine” — it is also what an alias no live binding names gives, and what a track this endpoint only publishes gives.

pub fn close_for_data_stream(&self, err: &ConnectionError) -> bool

Close the session when a failure raised while reading a data stream is one draft-20 answers with a close. Reports whether it closed.

recv_control does this for itself, because it owns both the stream and the connection. A data stream does not: accept_subgroup_stream hands the caller a FramedRecvStream, which holds no connection and so cannot close one, and the reads that raise these failures happen there. The caller is the only party holding both halves, which is what this is for:

if let Err(err) = framed.read_subgroup_object().await {
    conn.close_for_data_stream(&err);
    return Err(err.into());
}

Splitting it this way rather than closing inside the reader keeps a caller that is deliberately permissive — a tool reproducing a capture, say — able to read a violating stream and report it without tearing the session down. The rule is stated at endpoints, and this is where an endpoint decides it is one.

Only ConnectionError::Codec failures are matched, against the same codec_session_error_code table the control path uses, so a rule is answered with one code whichever stream carried it. The Object ID delta wrap of Section 11.4.2 is the entry that can only arrive this way.

pub fn close(&self, code: u32, reason: &[u8])

Close the connection.