Struct Connection 

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

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

Fields

transport: Transport

endpoint: Endpoint

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.

Implementations

impl Connection

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

Connect to a draft-09 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( &mut 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_alias: VarInt, track_namespace: TrackNamespace, track_name: Vec<u8>, subscriber_priority: u8, group_order: GroupOrder, filter_type: FilterType, ) -> Result<VarInt, ConnectionError>

Send a SUBSCRIBE and return the allocated subscribe ID.

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

Send a SUBSCRIBE for a range of the track and return the allocated ID.

The Filter Type comes from the arguments, so the message cannot name a filter whose fields it does not carry.

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

Send an UNSUBSCRIBE for the given subscribe ID.

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

Accept a subscription the peer opened, sending SUBSCRIBE_OK.

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

Reject a subscription the peer opened, sending SUBSCRIBE_ERROR.

The Track Alias travels back with the refusal: under the ‘Retry Track Alias’ code it is the alias the peer should try again with, and under any other code it is ignored.

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

End a subscription this endpoint accepted, sending SUBSCRIBE_DONE.

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, ) -> Result<VarInt, ConnectionError>

Send a FETCH and return the allocated subscribe ID.

pub async fn joining_fetch( &mut self, subscriber_priority: u8, group_order: GroupOrder, joining_subscribe_id: VarInt, preceding_group_offset: VarInt, ) -> Result<VarInt, ConnectionError>

Send a Joining Fetch and return the allocated subscribe ID.

preceding_group_offset counts groups back from the live edge of the subscription it names rather than naming a group of its own. Section 7.7.1 has the publisher compute “Fetch StartGroup: Subscribe Largest Group - Preceding Group Offset”, so 0 asks for the current group and nothing before it.

This draft draws one joining kind, at Fetch Type 0x2, so there is nothing here for a caller to choose between and nothing to choose wrongly.

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

Send a FETCH_CANCEL for the given subscribe ID.

pub async fn fetch_ok( &mut self, subscribe_id: VarInt, group_order: GroupOrder, end_of_track: u8, largest_group_id: VarInt, largest_object_id: VarInt, 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, subscribe_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, so nothing is written on the wire when it does.

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

Send a SUBSCRIBE_ANNOUNCES.

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

Accept a namespace subscription the peer made, sending SUBSCRIBE_ANNOUNCES_OK.

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

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

Refuse a namespace subscription the peer made, sending SUBSCRIBE_ANNOUNCES_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, ) -> Result<(), ConnectionError>

Send an ANNOUNCE.

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

Send an UNANNOUNCE.

pub async fn announce_ok( &mut self, track_namespace: TrackNamespace, ) -> 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, track_namespace: TrackNamespace, 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_request( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, ) -> Result<(), ConnectionError>

Send a TRACK_STATUS_REQUEST.

pub async fn track_status( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, status_code: VarInt, last_group_id: VarInt, last_object_id: VarInt, ) -> Result<(), ConnectionError>

Answer a TRACK_STATUS_REQUEST the peer sent, sending TRACK_STATUS.

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

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-09 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.

• 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.
• Duplicate parameters, a SHOULD rather than a MUST: “Receivers SHOULD check that there are no duplicate parameters and close the session as a ‘Protocol Violation’ if found.” Unqualified on this draft, where later ones exempt a repeat the message authorizes.
• Unknown control message type: “An endpoint that receives an unknown message type MUST close the session.”
• End of Track at a non-zero Object ID, Section 8.1.1.1: “An object with this status that has a Group ID less than or equal to any other Group ID, or an Object ID other than zero, is a protocol error, and the receiver MUST terminate the session.” This one arrives on a data stream, so Connection::close_for_data_stream is what carries it.
• A parameter whose value does not match the length its type implies — the one rule here that is not answered with a Protocol Violation: “the receiver MUST terminate the session with error code ‘Parameter Length Mismatch’.”

Not the Full Track Name maximum, the Track Namespace byte cap, the zero-length namespace field, or the parameter value maximum. Those enter the specification at drafts 11 and 16 and this draft states none of them.

Not CodecError::UnexpectedEnd either, 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 key-value pair serialization rule. Drafts 11 and later require a close with KEY_VALUE_FORMATTING_ERROR when a value does not match the serialization its Type defines; this draft has no Key-Value Pair at all. What it states instead is the Parameter Length Mismatch rule above, over the Parameter framing it has in its place.

Not the unknown Message Parameter rule, which enters at draft-16 and requires a close for a Message Parameter type the negotiated version does not define. This draft states nothing of the kind, and its parameters are not Key-Value-Pairs at all.

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.

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

Close the session on the wire when a decode failure is one draft-09 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 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 the end-of-track Object ID, which arrives on a subgroup or fetch stream and nowhere else. 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.