Struct Connection 

pub struct Connection {
    transport: Transport,
    endpoint: Endpoint,
    control_send: Option<Mutex<FramedSendStream>>,
    control_recv: Option<FramedRecvStream>,
    observer: Option<Box<dyn ConnectionObserver>>,
    pending_events: Vec<ClientEvent>,
}

A live draft-13 MoQT connection over QUIC or WebTransport.

Fields

transport: Transport

endpoint: Endpoint

control_send: Option<Mutex<FramedSendStream>>

Behind a lock because a control message is written from two kinds of place. Most of them are the caller’s own request, made through &mut self. The UNSUBSCRIBE that answers a Malformed Track is not: the conditions that make a track malformed are detected on the data plane, where this connection is reached through a shared reference. The lock also makes one message the unit of writing, so two of them cannot interleave on the stream.

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.

Implementations

impl Connection

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

Connect to a draft-13 MoQT server as a client.

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( &self, msg: &ControlMessage, ) -> Result<(), ConnectionError>

Send a control message on the control stream.

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

Read the next control message from the control stream.

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.

pub async fn subscribe( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, subscriber_priority: u8, group_order: GroupOrder, filter_type: FilterType, parameters: Vec<KeyValuePair>, ) -> Result<VarInt, ConnectionError>

Send a SUBSCRIBE and return the allocated request ID.

Draft-12: the subscriber no longer chooses a track alias. The alias is returned by the publisher in SUBSCRIBE_OK and can be retrieved later from the endpoint via endpoint().track_alias_for(request_id).

pub async fn subscribe_range( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, subscriber_priority: u8, group_order: GroupOrder, start_location: Location, end_group: Option<VarInt>, parameters: Vec<KeyValuePair>, ) -> Result<VarInt, ConnectionError>

Send a SUBSCRIBE for a range of the track, starting at a given location.

The Filter Type is derived from the range, so the message cannot name a filter whose fields it does not carry.

pub async fn unsubscribe( &mut self, request_id: VarInt, ) -> Result<(), ConnectionError>

Send an UNSUBSCRIBE for the given request ID.

pub async fn subscribe_ok( &mut self, request_id: VarInt, track_alias: VarInt, expires: VarInt, group_order: GroupOrder, parameters: Vec<KeyValuePair>, ) -> Result<(), ConnectionError>

Accept a subscription the peer opened, sending SUBSCRIBE_OK and giving its track a Track Alias.

The endpoint refuses an alias a live track of its own already holds and refuses a second answer to one SUBSCRIBE, so nothing is written on the wire when it does either.

pub async fn subscribe_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<(), ConnectionError>

Reject a subscription the peer opened, sending SUBSCRIBE_ERROR.

The endpoint refuses a second answer to one SUBSCRIBE, so nothing is written on the wire when it does.

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

End a subscription this endpoint accepted, sending SUBSCRIBE_DONE.

pub async fn subscribe_update( &mut self, request_id: VarInt, start_group: VarInt, start_object: VarInt, end_group: VarInt, subscriber_priority: u8, forward: Forward, parameters: Vec<KeyValuePair>, ) -> Result<(), ConnectionError>

Narrow a subscription this endpoint opened, sending SUBSCRIBE_UPDATE.

No Request ID is spent: on this draft the message’s one identifier field names the subscription it modifies.

pub async fn publish( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, track_alias: VarInt, group_order: GroupOrder, largest_location: Option<Location>, forward: Forward, parameters: Vec<KeyValuePair>, ) -> Result<VarInt, ConnectionError>

Offer the peer a subscription to a track this endpoint publishes, and return the Request ID the offer was allocated.

Nothing is written on the wire when the endpoint refuses to build the offer, which it does when the Track Alias is one another live track of this session already holds.

pub async fn publish_ok( &mut self, request_id: VarInt, forward: Forward, subscriber_priority: u8, group_order: GroupOrder, filter_type: FilterType, start_group: Option<VarInt>, start_object: Option<VarInt>, end_group: Option<VarInt>, ) -> Result<(), ConnectionError>

Accept a PUBLISH the peer sent, which establishes the subscription it opened.

The endpoint refuses a second answer to one PUBLISH, so nothing is written on the wire when it does.

pub async fn publish_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<(), ConnectionError>

Reject a PUBLISH the peer sent, which ends the subscription it opened before it was established.

The endpoint refuses a second answer to one PUBLISH, so nothing is written on the wire when it does.

pub async fn fetch( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, subscriber_priority: u8, group_order: GroupOrder, start_group: VarInt, start_object: VarInt, end_group: VarInt, end_object: VarInt, parameters: Vec<KeyValuePair>, ) -> Result<VarInt, ConnectionError>

Send a FETCH and return the allocated request ID.

pub async fn joining_fetch( &mut self, subscriber_priority: u8, group_order: GroupOrder, joining_request_id: VarInt, joining_start: VarInt, parameters: Vec<KeyValuePair>, ) -> Result<VarInt, ConnectionError>

Send a Relative Joining Fetch and return the allocated request ID.

joining_start counts groups back from the live edge of the subscription it names. Section 8.16.1 computes the range from “the Preceding Group Offset”, which is this field under the name it carried two drafts ago; the field list calls it Joining Start and says that for a Relative Joining Fetch “a value of 0 indicates the Fetch starts at the beginning of the Current Group”. For the form that names the group outright, see absolute_joining_fetch.

pub async fn absolute_joining_fetch( &mut self, subscriber_priority: u8, group_order: GroupOrder, joining_request_id: VarInt, joining_start: VarInt, parameters: Vec<KeyValuePair>, ) -> Result<VarInt, ConnectionError>

Send an Absolute Joining Fetch and return the allocated request ID.

joining_start is the group to begin at. Section 8.16 has an Absolute Joining Fetch “Identical to a Relative Joining Fetch except that the Start Group is determined by an absolute Group value rather than a relative offset to the subscription”, so a subscriber that knows the group it wants can ask for it without first being told a Largest Location to count back from.

Two calls rather than one taking a Fetch Type, because these are the only two the joining form admits. The endpoint is named the same way, so no call between an application and the wire has a Fetch Type in it to be given a wrong one.

pub async fn fetch_cancel( &mut self, request_id: VarInt, ) -> Result<(), ConnectionError>

Send a FETCH_CANCEL for the given request ID.

pub async fn fetch_ok( &mut self, request_id: VarInt, group_order: GroupOrder, end_of_track: u8, end_location: Location, parameters: Vec<KeyValuePair>, ) -> Result<(), ConnectionError>

Accept a fetch the peer opened, sending FETCH_OK.

The endpoint refuses a Joining Fetch naming a subscription this session cannot join and refuses a second answer to one FETCH, so nothing is written on the wire when it does either.

pub async fn fetch_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<(), ConnectionError>

Refuse a fetch the peer opened, sending FETCH_ERROR.

The endpoint refuses a second answer to one FETCH, and refuses a Joining Fetch’s refusal under any code but the one the draft names for it, so nothing is written on the wire when it does either.

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

Send a SUBSCRIBE_NAMESPACE. Returns the allocated request ID.

pub async fn subscribe_namespace_ok( &mut self, request_id: VarInt, ) -> Result<(), ConnectionError>

Accept a namespace subscription the peer made, sending SUBSCRIBE_NAMESPACE_OK.

The endpoint refuses a second answer to one SUBSCRIBE_NAMESPACE, so nothing is written on the wire when it does.

pub async fn subscribe_namespace_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<(), ConnectionError>

Refuse a namespace subscription the peer made, sending SUBSCRIBE_NAMESPACE_ERROR.

The other half of the same sentence: one answer, and this is the other one it can be.

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

Send an ANNOUNCE. Returns the allocated request ID.

pub async fn unannounce( &mut self, track_namespace: TrackNamespace, ) -> Result<(), ConnectionError>

Send an UNANNOUNCE.

pub async fn announce_ok( &mut self, request_id: VarInt, ) -> Result<(), ConnectionError>

Accept an announcement the peer made, sending ANNOUNCE_OK.

The endpoint refuses a second answer to one ANNOUNCE, so nothing is written on the wire when it does.

pub async fn announce_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<(), ConnectionError>

Refuse an announcement the peer made, sending ANNOUNCE_ERROR.

The other half of the same sentence: one answer, and this is the other one it can be.

pub async fn announce_cancel( &mut self, track_namespace: TrackNamespace, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<(), ConnectionError>

Revoke an acceptance, sending ANNOUNCE_CANCEL.

The endpoint refuses one for an announcement it never accepted, so nothing is written on the wire when it does.

pub async fn track_status( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, subscriber_priority: u8, group_order: GroupOrder, forward: Forward, filter_type: FilterType, parameters: Vec<KeyValuePair>, ) -> Result<VarInt, ConnectionError>

Send a TRACK_STATUS (draft-13 subscribe-like request). Returns the allocated request ID.

pub async fn track_status_ok( &mut self, request_id: VarInt, expires: VarInt, group_order: GroupOrder, parameters: Vec<KeyValuePair>, ) -> Result<(), ConnectionError>

Accept a track status the peer asked for, sending TRACK_STATUS_OK.

The endpoint refuses a second answer to one request, so nothing is written on the wire when it does. It also chooses the Track Alias the message carries, because the draft leaves only one value open.

pub async fn track_status_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<(), ConnectionError>

Refuse a track status the peer asked for, sending TRACK_STATUS_ERROR.

The other answer the request can have, and the endpoint holds it to the same count of one.

async fn withdraw_malformed_track( &self, alias: u64, condition: MalformedTrackCondition, )

Answer Section 2.5 for the track alias names: withdraw from it, and send the UNSUBSCRIBE that withdrawal is.

“When a subscriber detects a Malformed Track, it MUST UNSUBSCRIBE from the Track and SHOULD deliver an error to the application.” The caller delivers the error by returning it; this is the other half, and the caller does not have to remember to ask for it.

Nothing here ends the session, and that is the point of it. The conditions in Section 2.5 are answered by giving up one track.

Why a write that fails is not reported

The error the caller is about to receive is the one that says what went wrong with the track. A control stream that will not take an UNSUBSCRIBE is a session on its way out for a reason of its own, and replacing the condition’s report with a transport error would lose the only account of why the track was withdrawn. The rest of the withdrawal is abandoned, because a stream that refused one message will refuse the next.

async fn note_received_framing( &self, alias: u64, seen: ObjectForwardingPreference, ) -> Result<(), ConnectionError>

Record the framing an arriving object was sent with, and withdraw from the track when it is the second framing that track has been sent.

The receiving half of a pair. The two writing paths call the endpoint directly and answer a mixed track by refusing to write it, because Section 2.5’s sentence is a subscriber’s: an endpoint about to send an object is that object’s Original Publisher, and a publisher has no subscription of its own to withdraw.

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 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_subgroup_stream( &self, ) -> Result<(AnySubgroupHeader, FramedRecvStream), ConnectionError>

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

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

Send an object via datagram.

The header goes through AnyDatagramHeader::encode, which refuses a header whose Object Status the framing it names cannot carry. Such a header errors here and nothing is sent, rather than going out as an ordinary payload datagram with the status quietly dropped.

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

Receive a datagram and decode its header.

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 negotiated_version(&self) -> Option<VarInt>

Get the negotiated MoQT version.

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

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

Every variant listed here comes from a sentence in this draft that names the consequence, and the list is per draft: answering a bound this draft does not state would close a session over traffic a conforming peer may send.

• Reason Phrase, maximum 1024 bytes: “If an endpoint receives a length exceeding the maximum, it MUST close the session with a Protocol Violation.”
• GOAWAY New Session URI, maximum 8,192 bytes, with the same sentence. Drafts 11 through 19 state it; 07 through 10 state no maximum for the field at all.
• Key-Value-Pair value, maximum 2^16-1 bytes, with the same sentence.
• Track Namespace tuple size: “If an endpoint receives a Track Namespace tuple with an N of 0 or more than 32, it MUST close the session with a Protocol Violation.” Note the lower bound - an empty tuple is refused here, where drafts 17 and later permit one.
• Full Track Name, maximum 4,096 bytes, “computed as the sum of the lengths of each Track Namespace tuple field and the Track Name length field”. This draft bounds the pair and not the namespace alone; draft-16 widened it.
• Duplicate parameters, a SHOULD rather than a MUST: “Receivers SHOULD check that there are no unauthorized duplicate parameters and close the session as a ‘Protocol Violation’ if found.” The rule is asymmetric here - “Receivers MUST allow duplicates of unknown parameters”, and one known type is granted repeats - and the codec reports only the repeats this draft actually forbids.
• Unknown control message type: “An endpoint that receives an unknown message type MUST close the session.”
• Extension headers on an Object whose status is Object Does Not Exist, Section 9.2.1.2. That one arrives on a data stream or a datagram, so Connection::close_for_data_stream is what carries it.

Not the zero-length Track Namespace Field, the delta-encoded parameter type overflow, or the Object ID delta wrap. Those enter the specification at drafts 16 and 18, and this draft states none of them.

Not the 2^16-1 control message length either. This draft states the limit - “the total length of a control message is limited to 2^16-1 bytes” - and states no consequence for exceeding it, so an oversized message is refused by the decoder and stops there.

Not CodecError::UnexpectedEnd, which reports no rule at all: the reader raises it whenever a message is still arriving, and read_control loops on it. Closing over it would end a session on an ordinary short read.

Not the unknown Message Parameter rule. Drafts 16 through 19 require a close for a Message Parameter whose type the negotiated version does not define. This draft states the opposite and states it about the same parameters: “Receivers MUST allow duplicates of unknown parameters”, which presumes an unknown parameter arrives and is carried. Refusing one here would close a session over an extension this draft leaves room for.

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 a session cannot be ended on it without ending sessions the draft does not ask to be ended. Splitting it is the way to bring the rest of those rules under this function; widening the match is not - the extension-header rule above reached this table by being split out of it.

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

Close the session on the wire when a decode failure is one draft-13 answers with a close, and hand the error back unchanged. 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” is a statement about the wire.

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 this draft ends 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 is purely internal: the local endpoint refuses to start anything new while the peer, which is the one that broke the rule, sees a session that is still open and goes on sending. A rule that names a session termination code 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 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.

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

Withdraw from a track a data stream found malformed, reporting whether it did.

The Malformed Track twin of Connection::close_for_data_stream, and separate from it for the same reason and one more. The same one: a FramedRecvStream holds no connection, so the reader that finds the fault is not the object that can send an UNSUBSCRIBE. The one more: the two answers are opposites — that call ends the session, this one gives up a track and leaves it running — and a single entry point would have to decide between them from the error alone, which is exactly the decision a caller reproducing a capture wants to make itself.

The datagram path needs none of this. It is read through the connection, so Connection::recv_datagram answers the condition where it finds it, and this is only for the objects that arrive on a stream the caller holds.

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

Close the session over a rule broken on a data stream, reporting whether it did.

A data stream cannot close for itself the way recv_control does: Connection::accept_subgroup_stream hands the caller a FramedRecvStream holding no connection, so the reader that finds the violation is not the object that can act on it. Keeping it a separate call is deliberate as well - a permissive caller, one reproducing a capture, can read a violating stream and report it without tearing the session down.

The rule this draft answers here is extension headers on an Object whose status is Object Does Not Exist, which reaches subgroup streams, fetch streams and status datagrams alike. It shares codec_session_error_code with the control path, so a rule is answered with one code whichever stream carried it.

Not every rule that reaches here is the decoder’s. A track whose objects mix forwarding preferences is the endpoint’s to notice — it takes the alias table to know which track an object belongs to — and it arrives on exactly these streams. Both kinds are asked for a code the same way, and a rule with no code is declined rather than guessed at.

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

Close the connection.