Struct Endpoint 

pub struct Endpoint {

    role: Role,
    session: SessionStateMachine,
    request_ids: RequestIdAllocator,
    advertised_max_id: u64,
    subscriptions: HashMap<u64, SubscriptionStateMachine>,
    inbound_subscribes: HashMap<u64, InboundSubscribe>,
    inbound_fetches: HashMap<u64, InboundFetch>,
    forwarding_preferences: Mutex<TrackForwardingPreferences>,
    locations: Arc<Mutex<TrackLocations>>,
    track_bindings: HashMap<u64, TrackBinding>,
    fetches: HashMap<u64, FetchStateMachine>,
    subscribe_announces: HashMap<u64, SubscribeAnnouncesStateMachine>,
    inbound_subscribe_announces: HashMap<u64, InboundSubscribeAnnounces>,
    announces: HashMap<u64, AnnounceStateMachine>,
    inbound_announces: HashMap<u64, InboundAnnounce>,
    announce_ids: HashMap<Vec<Vec<u8>>, u64>,
    subscribe_announces_ids: HashMap<Vec<Vec<u8>>, u64>,
    track_statuses: HashMap<u64, TrackStatusStateMachine>,
    inbound_track_statuses: HashMap<u64, InboundTrackStatus>,
    negotiated_version: Option<VarInt>,
    offered_versions: Vec<VarInt>,
    goaway_uri: Option<Vec<u8>>,
    peer_reported_max_request_id: Option<VarInt>,
}

Unified draft-11 MoQT endpoint wrapping session lifecycle, request ID allocation, and all per-flow state machines (subscriptions, fetches, announces, subscribe-announces, track statuses).

Fields

role: Role

Which end of the session this is, which fixes the parity of the request IDs it allocates.

session: SessionStateMachine

request_ids: RequestIdAllocator

advertised_max_id: u64

Tracks the MAX_REQUEST_ID we have advertised to the peer.

subscriptions: HashMap<u64, SubscriptionStateMachine>

inbound_subscribes: HashMap<u64, InboundSubscribe>

Subscriptions the peer opened with SUBSCRIBE, each from the moment its message arrived to the end of the flow.

Separate from subscriptions, which holds the ones this endpoint opened. The two never collide on an identifier - Request IDs carry the allocating end’s parity - but a subscription’s state is read from whichever end drives it, and the two ends drive different messages.

inbound_fetches: HashMap<u64, InboundFetch>

Every FETCH the peer has sent, from arrival to the end of the fetch.

Separate from fetches, which holds the ones this endpoint made. The record holds the FETCH itself and not only its state, because the answer is built out of the request: a Joining Fetch has to be judged against the subscription it names, and that name is nowhere else.

forwarding_preferences: Mutex<TrackForwardingPreferences>

Every Track Alias in use in this session, and the track each one names.

“Already being used” is what makes this a table rather than a set: an alias whose subscription has ended is free again. The table records the binding and reads liveness back off the subscription’s own state machine, rather than keeping a second copy of it that every path ending a subscription would have to remember to prune. What each track’s objects have been framed as, so far.

Behind a lock because this is the one endpoint fact a data stream settles, and the data plane reaches the endpoint through &Connection: a caller may hold one across tasks while it reads streams and datagrams, so there is no &mut to reach the rest of this struct with.

locations: Arc<Mutex<TrackLocations>>

How far each track’s objects have reached, so far.

Behind an Arc rather than beside the rest of this struct because the objects that settle it are read off stream handles the caller owns, one at a time, with no way back to the endpoint. Each such stream is handed a clone of the handle, and every clone measures against this one record.

track_bindings: HashMap<u64, TrackBinding>

fetches: HashMap<u64, FetchStateMachine>

subscribe_announces: HashMap<u64, SubscribeAnnouncesStateMachine>

inbound_subscribe_announces: HashMap<u64, InboundSubscribeAnnounces>

Namespace subscriptions the peer made, keyed by the Request ID it opened each under.

Kept apart from subscribe_announces, which holds the ones this endpoint made: one Request ID names one request whichever end opened it, and the two ends answer opposite halves of the flow.

announces: HashMap<u64, AnnounceStateMachine>

inbound_announces: HashMap<u64, InboundAnnounce>

Announcements the peer made, keyed by the Request ID it opened each under.

announce_ids: HashMap<Vec<Vec<u8>>, u64>

Maps namespace tuple -> request_id, so callers can UNANNOUNCE / cancel by namespace without threading the id through every API.

subscribe_announces_ids: HashMap<Vec<Vec<u8>>, u64>

Maps namespace prefix tuple -> request_id for subscribe-announces.

track_statuses: HashMap<u64, TrackStatusStateMachine>

inbound_track_statuses: HashMap<u64, InboundTrackStatus>

Track statuses the peer asked about, keyed by the Request ID it asked under.

Kept apart from track_statuses, which holds the ones this endpoint asked about: the two are answered by opposite ends, and one Request ID belongs to one request whichever end opened it.

negotiated_version: Option<VarInt>

offered_versions: Vec<VarInt>

goaway_uri: Option<Vec<u8>>

peer_reported_max_request_id: Option<VarInt>

The most recent maximum_request_id reported by the peer via a REQUESTS_BLOCKED message.

Implementations

impl Endpoint

pub fn new(role: Role) -> Self

Create a new draft-11 endpoint for the given role.

fn alias_holder( &self, alias: u64, namespace: &TrackNamespace, name: &[u8], ) -> Option<u64>

The request already using alias for a track other than (namespace, name), or None when the alias is free for that track.

Why the set is read rather than kept

Section 8.7 says “already being used”, and a subscription that has ended is not using anything. Asking each binding’s own state machine is what makes an alias free again the instant its track’s subscription ends, with nothing to prune on the way out - and a path that ended a subscription without telling this table would otherwise hold the alias forever and refuse the peer’s next, conforming, use of it.

Why a binding for the same track is not a conflict

The rule is about a Track Alias naming two tracks, not about naming one track twice. A second subscription to the track an alias already names breaks nothing this section states.

fn track_for_alias(&self, alias: u64) -> Option<(&TrackNamespace, &[u8])>

The track a live binding has given alias to.

Read rather than kept, for the reason the alias table beside it gives: a binding whose request has ended holds nothing, and an alias that is free again may name a different track next. That is exactly why the forwarding-preference record below is keyed on the track this returns and never on the alias itself.

pub fn track_objects(&self, alias: u64) -> Option<TrackObjects>

The record a stream carrying alias’s objects measures them against.

None for an alias no live binding names: an object for one breaks a different rule, and measuring it against a track this endpoint never asked for would answer that one with the wrong sentence.

pub fn note_received_object( &self, alias: u64, at: ObjectLocation, role: ObjectRole, ) -> Result<(), EndpointError>

Record or judge one object that arrived outside a subgroup stream, and report Section 9.1.1.1’s protocol error when it says the track ended somewhere the track has already passed.

&self, because the call site is the data plane’s.

pub fn note_object_forwarding_preference( &self, alias: u64, seen: ObjectForwardingPreference, ) -> Result<(), EndpointError>

Record how a track’s object was framed, and report Section 9’s “MUST NOT mix” when it disagrees with what that track’s earlier objects used.

&self, because the call sites are the data plane’s: a subgroup header arriving, a datagram arriving, and the two writers that produce them.

An alias no live binding names records nothing and reports nothing. An object for such an alias breaks a different rule — the one about objects nobody asked for — and answering that one here would answer it with the wrong sentence.

fn binding_is_live(&self, id: u64, kind: BindingKind) -> bool

Whether the subscription that owns a binding is still standing. Subscribing counts as well as Active, which is what separates this draft from the next one: draft-12 Section 8.8 reads “a different track with an active subscription” and the alias arrives in the answer, so only an answered request holds one. Here the alias is in the SUBSCRIBE itself, and the sentence puts no qualifier on “already being used” - so it is in use from the moment that message is sent or received, and stays in use until the subscription ends.

fn conflicting_track_alias( &self, id: u64, alias: u64, namespace: &TrackNamespace, name: &[u8], ) -> Option<EndpointError>

The close Section 8.7 requires of an arriving SUBSCRIBE whose Track Alias is spoken for, or None when it is free.

fn conflicting_retry_alias(&self, id: u64, alias: u64) -> Option<EndpointError>

The close Section 8.9 requires of a SUBSCRIBE_ERROR offering a Track Alias to retry with, or None when the offer can be taken up.

The track is not in the SUBSCRIBE_ERROR: it is the one this endpoint’s own SUBSCRIBE asked for, so the request has to be looked up before the alias offered for it can be judged. An offer of an alias this endpoint already holds for that same track is the retry succeeding, not a conflict.

pub fn role(&self) -> Role

Returns the role (client or server) of this endpoint.

pub fn session_state(&self) -> SessionState

Returns the current session state.

pub fn negotiated_version(&self) -> Option<VarInt>

Returns the negotiated MoQT version, if setup is complete.

pub fn goaway_uri(&self) -> Option<&[u8]>

Returns the URI from a received GOAWAY message, if any.

pub fn is_blocked(&self) -> bool

Returns whether this endpoint is blocked on request ID allocation.

pub fn active_subscription_count(&self) -> usize

Returns the number of active subscription state machines.

pub fn active_fetch_count(&self) -> usize

Returns the number of active fetch state machines.

pub fn active_subscribe_announces_count(&self) -> usize

Returns the number of active subscribe-announces state machines.

pub fn active_announce_count(&self) -> usize

Returns the number of active announce state machines.

pub fn active_track_status_count(&self) -> usize

Returns the number of active track status state machines.

pub fn connect(&mut self) -> Result<(), EndpointError>

Transition from Connecting to SetupExchange.

pub fn close(&mut self) -> Result<(), EndpointError>

Close the session (SetupExchange, Active or Draining -> Closed).

pub fn send_client_setup( &mut self, versions: Vec<VarInt>, parameters: Vec<KeyValuePair>, ) -> Result<ControlMessage, EndpointError>

Generate a CLIENT_SETUP message (client-side).

pub fn receive_server_setup( &mut self, msg: &ServerSetup, ) -> Result<(), EndpointError>

Process a SERVER_SETUP message (client-side). Transitions to Active. If the server includes a MAX_REQUEST_ID parameter (key 0x02), the request ID allocator is initialized with that value.

pub fn receive_client_setup_and_respond( &mut self, client_setup: &ClientSetup, selected_version: VarInt, ) -> Result<ControlMessage, EndpointError>

Process CLIENT_SETUP and generate SERVER_SETUP (server-side).

pub fn validate_peer_request_id(&mut self, id: u64) -> Result<(), EndpointError>

Take a Request ID from a new request the peer sent, holding it to all three halves of the rule.

Section 8.1 closes the session with Invalid Request ID on a request ID “that is not valid for the peer”, and Section 8.5 closes it with ‘Too Many Requests’ on one at or above the ceiling this endpoint advertised. The parity half belongs to the allocator; the ceiling half needs advertised_max_id, which is this endpoint’s grant to the peer and not the peer’s grant to it, so the two numbers are different and only one of them answers this question.

The third is the sequence. The same sentence closes the session on “a new request with a Request ID that is not expected”, and what is expected is fixed by the two lines above it: each endpoint starts at 0 or 1 by role and steps by 2 per request. So a repeat and a skip are both refused, and neither is caught by parity or by the ceiling - a repeat has the right parity and sits below the ceiling by construction, having been accepted once already.

Taking the ID is what advances the sequence, so this is for a new request only. A response, a cancellation or an update that names the request it modifies all carry an ID that has already been spent, and passing one here would refuse it.

The ceiling is a session rule, not a request rule

Section 8.5: a Request ID “equal or larger than this” received by the endpoint that sent the MAX_REQUEST_ID in any request message - the list Self::receive_request carries - closes the session, and ‘Too Many Requests’ is the code. Which is also why the number measured against is the one this endpoint sent. An id that reaches the ceiling is not a request to refuse with an error message: the session is over, so this moves the endpoint’s own state to Closed and leaves the code to EndpointError::session_error_code.

Errors

RequestIdError::WrongParity if the ID belongs to this endpoint’s half of the space, RequestIdError::ExceedsMax if it is not below the advertised ceiling, or RequestIdError::OutOfSequence if it is not the one the peer’s sequence called for. A ceiling that was never raised is 0, which refuses every ID, matching a default that reads “the peer MUST NOT send requests”.

All three end the session rather than the one request, so all three move this endpoint’s own session to Closed on the way out and all three answer Some from EndpointError::session_error_code. Returning one of them without ending the session would leave a peer that broke the rule free to keep sending, and never told.

fn record_advertised_max(&mut self, parameters: &[KeyValuePair])

Record a MAX_REQUEST_ID parameter this endpoint is about to send as the ceiling it has advertised to the peer.

pub fn receive_max_request_id( &mut self, msg: &MaxRequestId, ) -> Result<(), EndpointError>

Process an incoming MAX_REQUEST_ID message, ending the session if the ceiling it carries does not increase. Section 8.5: “The Maximum Request ID MUST only increase within a session, and receipt of a MAX_REQUEST_ID message with an equal or smaller Request ID value is a ‘Protocol Violation’.” Section 3.4 lists Protocol Violation among the codes for terminating the session - “The remote endpoint performed an action that was disallowed by the specification” - so naming it of a receipt is this draft saying the session ends, and with which code. Draft-16 Section 9.5 states the same rule with the verb in it: “it MUST close the session with a PROTOCOL_VIOLATION”.

Errors

RequestIdError::Decreased if the value does not increase, with the session already moved to Closed.

pub fn send_max_request_id( &mut self, max_id: VarInt, ) -> Result<ControlMessage, EndpointError>

Generate a MAX_REQUEST_ID message (typically server-side).

Section 8.5: “The Maximum Request ID MUST only increase within a session”, and a peer that receives an equal or smaller value closes the session. The ceiling starts at 0 and 0 is not greater than 0, so the first value that may go on the wire is 1 and there is no opening case where a repeat is allowed.

Errors

The decrease error if the value does not strictly increase.

pub fn receive_goaway(&mut self, msg: &GoAway) -> Result<(), EndpointError>

Process an incoming GOAWAY message. Transitions to Draining.

Errors

EndpointError::GoAwayUriAtServer if this endpoint is the server and the GOAWAY carries a New Session URI. The session is over: this endpoint’s own state has moved to Closed and the code the transport should close with is in EndpointError::session_error_code.

EndpointError::RepeatedGoAway if a GOAWAY has already been received. The session is over: this endpoint’s own state has moved to Closed and the code the transport should close with is in EndpointError::session_error_code.

fn require_active_or_err(&self) -> Result<(), EndpointError>

fn fail_session(&mut self, err: EndpointError) -> EndpointError

Record that the session is over because the peer broke a rule this draft answers with a session close, and hand the error back unchanged.

The state move is what makes the violation stick: every request entry point goes through require_active_or_err, so a caller that ignores the returned error still cannot start anything new. Closing on the wire is the connection layer’s job - see EndpointError::session_error_code for the code it should use.

pub fn subscribe( &mut self, track_alias: VarInt, track_namespace: TrackNamespace, track_name: Vec<u8>, subscriber_priority: u8, group_order: GroupOrder, filter_type: VarInt, ) -> Result<(VarInt, ControlMessage), EndpointError>

Send a SUBSCRIBE message. Allocates an ID and creates a subscription state machine.

AbsoluteStart and AbsoluteRange name a start location, which this call has no way to supply, and are answered with EndpointError::FilterNeedsRange - use Self::subscribe_range for those. Without the refusal this call would hand back a message whose filter announces fields the message does not carry, and the frame that goes on the wire is short by exactly those fields.

pub 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, ControlMessage), EndpointError>

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

The Filter Type is derived from the arguments rather than taken beside them, so the message cannot name a filter whose fields it does not carry.

fn subscribe_inner( &mut self, track_alias: VarInt, track_namespace: TrackNamespace, track_name: Vec<u8>, subscriber_priority: u8, group_order: GroupOrder, filter_type: VarInt, start_location: Option<Location>, end_group: Option<VarInt>, ) -> Result<(VarInt, ControlMessage), EndpointError>

pub fn receive_subscribe_ok( &mut self, msg: &SubscribeOk, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE_OK.

pub fn receive_subscribe_error( &mut self, msg: &SubscribeError, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE_ERROR.

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

Send an UNSUBSCRIBE message for an active subscription.

pub 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<ControlMessage, EndpointError>

Send a SUBSCRIBE_UPDATE narrowing a subscription this endpoint opened.

Section 8.10 gives the message to the subscriber, which is what this endpoint is for every subscription in subscriptions. No identifier is spent: the update’s one identifier field names the subscription being modified rather than opening a request of its own.

The narrowing rules the same section states are the caller’s to keep.

Errors

EndpointError::UnknownRequest when this endpoint opened no subscription under that identifier, and the subscription flow’s own InvalidTransition when the one it names has already ended.

pub fn receive_subscribe_update( &mut self, msg: &SubscribeUpdate, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE_UPDATE.

Section 8.10: “A subscriber issues a SUBSCRIBE_UPDATE to a publisher to request a change to an existing subscription.” One that arrives is therefore about a subscription the peer opened, which is why it is looked for among those and not among this endpoint’s own.

Errors

EndpointError::UpdateForUnknownRequest when the identifier names no subscription the peer has opened in this session, and the subscription flow’s own InvalidTransition when it names one that has already ended. Neither ends the session: Section 8.10 says SHOULD.

pub fn receive_subscribe_done( &mut self, msg: &SubscribeDone, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE_DONE (subscriber side — publisher finished).

pub fn receive_subscribe( &mut self, msg: &Subscribe, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE, judging the Track Alias it carries and recording the subscription it opens.

Section 8.7: “If the Track Alias is already being used for a different track, the publisher MUST close the session with a Duplicate Track Alias error”. This endpoint is the publisher of a SUBSCRIBE that arrives, so this is where that close is raised. Judged before anything is written down, so a refused SUBSCRIBE leaves no binding behind.

The Request ID has already been checked by Self::receive_request, which every request message passes through before its own handler.

Errors

EndpointError::DuplicateTrackAlias when the alias names a second track, with the session already moved to Closed.

pub fn pending_subscribe(&self, request_id: VarInt) -> Option<&Subscribe>

The SUBSCRIBE the peer sent under request_id and this endpoint has not answered yet.

None once it has been answered, and for an identifier this session has no inbound subscription for.

pub fn pending_subscribe_count(&self) -> usize

How many SUBSCRIBEs the peer has sent that are still waiting for an answer.

pub fn send_subscribe_ok( &mut self, request_id: VarInt, expires: VarInt, group_order: GroupOrder, parameters: Vec<KeyValuePair>, ) -> Result<ControlMessage, EndpointError>

Build the SUBSCRIBE_OK accepting a subscription the peer opened.

Errors

EndpointError::UnknownRequest if the peer opened no subscription under that identifier, and EndpointError::Subscription if it has already been answered.

pub fn send_subscribe_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, track_alias: VarInt, ) -> Result<ControlMessage, EndpointError>

Build the SUBSCRIBE_ERROR rejecting a subscription the peer opened.

The Track Alias goes back out with the refusal because Section 8.9 gives the field a use: an alias to retry with, when the code is ‘Retry Track Alias’. Under any other code the peer reads nothing from it.

Errors

EndpointError::UnknownRequest if the peer opened no subscription under that identifier, and EndpointError::Subscription if it has already been answered.

pub fn send_subscribe_done( &mut self, request_id: VarInt, status_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<ControlMessage, EndpointError>

Build the SUBSCRIBE_DONE ending a subscription this endpoint accepted.

Errors

EndpointError::UnknownRequest if the peer opened no subscription under that identifier, and EndpointError::Subscription if it is not one this endpoint accepted and has not already ended.

pub fn receive_unsubscribe( &mut self, msg: &Unsubscribe, ) -> Result<(), EndpointError>

Process an incoming UNSUBSCRIBE, ending the subscription the peer opened and freeing the Track Alias it held.

Errors

EndpointError::UnknownRequest if the peer opened no subscription under that identifier, and EndpointError::Subscription if it is not one this endpoint accepted and has not already ended.

pub 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, ControlMessage), EndpointError>

Send a standalone FETCH message. Allocates a request ID and creates a fetch state machine.

pub fn joining_fetch( &mut self, subscriber_priority: u8, group_order: GroupOrder, joining_subscribe_id: VarInt, joining_start: VarInt, ) -> Result<(VarInt, ControlMessage), EndpointError>

Send a Relative Joining Fetch (Fetch Type 0x2), attaching a fetch to a subscription this session already holds. Allocates a Request ID.

Section 8.13 calls it “A Fetch joined together with a Subscribe by specifying the Request ID of an active subscription and a relative starting offset”, and has “A publisher receiving a Joining Fetch uses properties of the associated Subscribe to determine the Track Namespace, Track, Start Group, Start Object, End Group, and End Object such that it is contiguous with the associated Subscribe.” So a joining fetch names neither the namespace nor the name, and joining_start is read against the subscription rather than against the track: Section 8.13.1 has the publisher compute “Fetch Start Group: Subscribe Largest Group - Joining start”, which makes it a count of groups back from the live edge.

The subscription is named by joining_subscribe_id, which is what this draft’s field list calls the field although the sentence under it already reads “The Request ID of the existing subscription to be joined”.

Errors

The session error when the session is not established, and the request-id error when this endpoint has no identifier left to spend. A Request ID naming no subscription is not refused here — Section 8.13 answers that at the publisher, “it MUST respond with a Fetch Error with code Invalid Joining Subscribe ID”, and this endpoint is the subscriber.

pub fn absolute_joining_fetch( &mut self, subscriber_priority: u8, group_order: GroupOrder, joining_subscribe_id: VarInt, joining_start: VarInt, ) -> Result<(VarInt, ControlMessage), EndpointError>

Send an Absolute Joining Fetch (Fetch Type 0x3).

Section 8.13.2 is the whole of the difference: “Identical to the Relative Joining fetch except that Fetch Start Group is the Joining Start value.” So joining_start is the group to begin at rather than a count of groups back, which is what an application that knows the group it wants actually has. Asking for the same range relatively would need the Largest Group, and a subscriber that has not yet been told one cannot compute the offset.

Errors

As Endpoint::joining_fetch.

fn joining_fetch_of_type( &mut self, fetch_type: FetchType, subscriber_priority: u8, group_order: GroupOrder, joining_subscribe_id: VarInt, joining_start: VarInt, ) -> Result<(VarInt, ControlMessage), EndpointError>

What both calls above are, and why neither of them takes the type.

Section 8.13 admits three Fetch Types and this payload carries two of them. A type taken as an argument here would leave a third value a caller could pass and this call would have to answer for; naming the two rules it out instead, so there is no answer left to get wrong.

pub fn receive_fetch_ok(&mut self, msg: &FetchOk) -> Result<(), EndpointError>

Process an incoming FETCH_OK.

pub fn receive_fetch_error( &mut self, msg: &FetchError, ) -> Result<(), EndpointError>

Process an incoming FETCH_ERROR.

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

Send a FETCH_CANCEL message.

pub fn on_fetch_stream_fin( &mut self, request_id: VarInt, ) -> Result<(), EndpointError>

Notify that a fetch data stream received FIN.

It may arrive before the FETCH_OK or FETCH_ERROR answering the request, which leaves the fetch in FetchState::Unanswered until the answer lands.

pub fn on_fetch_stream_reset( &mut self, request_id: VarInt, ) -> Result<(), EndpointError>

Notify that a fetch data stream was reset.

As with a FIN, it may arrive before the answer to the request.

pub fn receive_fetch(&mut self, msg: &Fetch) -> Result<(), EndpointError>

Process an incoming FETCH, recording the fetch it opens.

The Request ID has already been checked by Self::receive_request, which every request message passes through before its own handler.

A Joining Fetch is recorded like any other. Section 8.13 answers one naming a subscription this session cannot join with a refusal and not a session close, and a refusal is a message this endpoint has to build, so the request it refuses has to be on record first.

Errors

The session error when the session is not established, and the fetch flow’s own InvalidTransition for a second FETCH under an identifier already carrying one.

pub fn pending_fetch(&self, request_id: VarInt) -> Option<&Fetch>

The FETCH the peer sent under request_id and this endpoint has not answered yet.

None once it has been answered, and for an identifier this session has no inbound fetch for. The record itself lives on past the answer, because the fetch is not over until its data stream is.

pub fn pending_fetch_count(&self) -> usize

How many FETCHes the peer has sent that are still waiting for an answer.

fn joining_subscription_missing(&self, msg: &Fetch) -> Option<u64>

The identifier an arriving Joining Fetch names, when this session has no subscription it may join.

Section 8.13: “If a publisher receives a Joining Fetch with a Request ID that does not correspond to an existing Subscribe in the same session, it MUST respond with a Fetch Error with code Invalid Joining Subscribe ID.”

The verdict is taken as the FETCH arrives, because that is the moment the sentence names, and it is kept. A subscription that ends between the FETCH and its answer does not turn a fetch that could be joined into one that could not.

A standalone fetch names none and answers None, and so does a joining one whose subscription is live. The subscription is one the peer opened, because the peer is the end that fetches and this endpoint is the one answering. “Existing” is read as “has not ended”. Draft-16 Section 9.16.2 states the same rule with the states named - “in the Established or Pending (subscriber) states” - which is the same set read the same way.

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

Build the FETCH_OK accepting a fetch the peer opened.

Errors

EndpointError::UnknownRequest if the peer opened no fetch under that identifier, EndpointError::UnjoinableSubscription for a Joining Fetch naming a subscription this session cannot join, and the fetch flow’s own InvalidTransition for a second answer: Section 4 says the publisher “MUST send exactly one FETCH_OK or FETCH_ERROR in response to a FETCH”.

pub fn send_fetch_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<ControlMessage, EndpointError>

Build the FETCH_ERROR refusing a fetch the peer opened.

Section 8.13 names the code a Joining Fetch naming an unjoinable subscription is refused with, so that refusal cannot go out under any other: a subscriber told the wrong reason retries the wrong thing.

Errors

EndpointError::UnknownRequest if the peer opened no fetch under that identifier, and the fetch flow’s own InvalidTransition if it has already been answered.

pub fn receive_fetch_cancel( &mut self, msg: &FetchCancel, ) -> Result<(), EndpointError>

Process an incoming FETCH_CANCEL, ending the fetch the peer opened.

Section 8.16: the subscriber sends it to stop a fetch it no longer wants, so the record this endpoint serves the fetch from is the one it ends.

Errors

EndpointError::UnknownRequest if the peer opened no fetch under that identifier, and the fetch flow’s own InvalidTransition for a fetch that has already ended.

pub fn on_peer_fetch_stream_fin( &mut self, request_id: VarInt, ) -> Result<(), EndpointError>

Note that this endpoint finished the data stream serving a fetch the peer opened.

A fetch is over when its answer and its data stream have both settled, and this is the second of those for the end that serves it.

Errors

EndpointError::UnknownRequest if the peer opened no fetch under that identifier, and the fetch flow’s own InvalidTransition from a state the stream cannot close from.

pub fn subscribe_announces( &mut self, track_namespace_prefix: TrackNamespace, ) -> Result<(VarInt, ControlMessage), EndpointError>

Send a SUBSCRIBE_ANNOUNCES message. Returns the allocated request ID alongside the control message so the caller can correlate replies.

Section 8.24 addresses the first half of the overlap rule to this end of the session: “A subscriber cannot make overlapping namespace subscriptions on a single session.” So a prefix overlapping one this endpoint has already asked about is refused here rather than built and sent for the peer to refuse.

Errors

The session error when the session is not established, EndpointError::OwnPrefixOverlap when the prefix overlaps one this endpoint is already subscribed to, and the Request ID allocator’s own error when the peer has granted no room for another request.

pub fn receive_subscribe_announces_ok( &mut self, msg: &SubscribeAnnouncesOk, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE_ANNOUNCES_OK.

pub fn receive_subscribe_announces_error( &mut self, msg: &SubscribeAnnouncesError, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE_ANNOUNCES_ERROR.

pub fn unsubscribe_announces( &mut self, track_namespace_prefix: TrackNamespace, ) -> Result<ControlMessage, EndpointError>

Send an UNSUBSCRIBE_ANNOUNCES message.

pub fn receive_subscribe_announces( &mut self, msg: &SubscribeAnnounces, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE_ANNOUNCES, recording the namespace subscription it opens.

Section 8.24: “The subscriber sends the SUBSCRIBE_ANNOUNCES control message to a publisher to request the current set of matching announcements, as well as future updates to the set.”

The set it asks for is this endpoint’s to decide, and deciding needs both the request and somewhere to answer from. The record holds the message and not only its state, because every message that answers this request or ends it names the prefix the request carried, and nothing else here has it.

Errors

The session error when the session is not established.

fn peer_prefix_overlap(&self, prefix: &TrackNamespace) -> Option<u64>

The earliest namespace subscription the peer has made whose prefix overlaps prefix, and None when there is none.

Every one the peer has made counts, including one it has since withdrawn and one this endpoint refused: the sentence weighs the arriving prefix against “an earlier SUBSCRIBE_ANNOUNCES”, and one that has ended was still earlier. Draft-12 changes that word to “active”, and there a subscription that has ended stops counting.

Namespace subscriptions this endpoint made are a separate set and are not consulted. This endpoint is the subscriber for those, so a prefix it asked about says nothing about what the peer may ask about.

The lowest Request ID wins when more than one overlaps, so the answer does not depend on the order a map happens to iterate in. Identifiers are handed out in increasing order, so the lowest of them is the earliest request.

fn own_prefix_overlap(&self, prefix: &TrackNamespace) -> Option<u64>

The earliest namespace subscription this endpoint has made whose prefix overlaps prefix, and None when there is none.

The subscriber’s half of the same sentence reads over this endpoint’s own requests, and takes the same view of which of them are past as Self::peer_prefix_overlap takes of the peer’s.

pub fn pending_subscribe_announces( &self, request_id: VarInt, ) -> Option<&SubscribeAnnounces>

The SUBSCRIBE_ANNOUNCES the peer sent under request_id and this endpoint has not answered yet.

None once it has been answered, and for an identifier the peer has subscribed to nothing under. The record itself lives on past the answer, because a namespace subscription that was accepted is not over until it is withdrawn.

pub fn pending_subscribe_announces_count(&self) -> usize

How many namespace subscriptions the peer has made that are still waiting for an answer.

fn inbound_subscribe_announces_id(&self, prefix: &TrackNamespace) -> Option<u64>

The identifier of a live namespace subscription the peer made for prefix.

Section 8.27 names a Track Namespace Prefix where the SUBSCRIBE_ANNOUNCES it ends named a Request ID, so one record has to be reachable both ways. It is stored under the identifier, which is unique, and found by prefix with a scan of the same map. A second map from prefix to identifier would be quicker and could fall out of step with the first; there is nothing here for it to disagree with.

One that has ended is skipped, so a prefix subscribed again after being withdrawn finds the live one.

pub fn send_subscribe_announces_ok( &mut self, request_id: VarInt, ) -> Result<ControlMessage, EndpointError>

Build the SUBSCRIBE_ANNOUNCES_OK accepting a namespace subscription the peer made.

Section 5.1: “A publisher MUST send exactly one SUBSCRIBE_ANNOUNCES_OK or SUBSCRIBE_ANNOUNCES_ERROR in response to a SUBSCRIBE_ANNOUNCES.”

One answer and no second one: the flow moves on the first, and a second call finds a record that has left Pending.

Errors

EndpointError::UnknownRequest if the peer has subscribed to nothing under that identifier, EndpointError::PeerPrefixOverlap if the prefix it asked about overlaps one the peer is already subscribed to, and the namespace flow’s own InvalidTransition for a request already answered.

pub fn send_subscribe_announces_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<ControlMessage, EndpointError>

Build the SUBSCRIBE_ANNOUNCES_ERROR refusing a namespace subscription the peer made.

The other half of the same sentence: one message back, whichever of the two it is.

Errors

EndpointError::UnknownRequest if the peer has subscribed to nothing under that identifier, EndpointError::WrongOverlapRefusal if the request overlaps another and the code named is not the one the draft assigns to that refusal, and the namespace flow’s own InvalidTransition for a request already answered.

pub fn receive_unsubscribe_announces( &mut self, msg: &UnsubscribeAnnounces, ) -> Result<(), EndpointError>

Process an incoming UNSUBSCRIBE_ANNOUNCES, ending the namespace subscription the peer made.

Section 5.1: “An UNSUBSCRIBE_ANNOUNCES withdraws a previous SUBSCRIBE_ANNOUNCES.”

The subscription it ends is the peer’s, so the record it reads is the one this endpoint keeps of what the peer subscribed to. One this endpoint made is withdrawn by Endpoint::unsubscribe_announces, which is the same message travelling the other way.

Errors

EndpointError::UnknownPeerNamespaceSubscription if the peer has no live namespace subscription for that prefix, and the namespace flow’s own InvalidTransition for one this endpoint never accepted.

pub fn announce( &mut self, track_namespace: TrackNamespace, ) -> Result<(VarInt, ControlMessage), EndpointError>

Send an ANNOUNCE message. Returns the allocated request ID alongside the control message.

pub fn receive_announce_ok( &mut self, msg: &AnnounceOk, ) -> Result<(), EndpointError>

Process an incoming ANNOUNCE_OK.

pub fn receive_announce_error( &mut self, msg: &AnnounceError, ) -> Result<(), EndpointError>

Process an incoming ANNOUNCE_ERROR.

pub fn receive_announce_cancel( &mut self, msg: &AnnounceCancel, ) -> Result<(), EndpointError>

Process an incoming ANNOUNCE_CANCEL.

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

Send an UNANNOUNCE message (publisher withdrawing).

pub fn receive_announce(&mut self, msg: &Announce) -> Result<(), EndpointError>

Process an incoming ANNOUNCE, recording the announcement it makes.

Section 8.19: “The publisher sends the ANNOUNCE control message to advertise that it has tracks available within the announced Track Namespace. The receiver verifies the publisher is authorized to publish tracks under this namespace.”

Verifying is the application’s to do, and it needs both the message to verify and somewhere to answer from. The Request ID has already been checked by Self::receive_request, which every request message passes through before its own handler, so a repeat of one the peer has already spent never reaches here.

Errors

The session error when the session is not established.

pub fn pending_announce(&self, request_id: VarInt) -> Option<&Announce>

The ANNOUNCE the peer sent under request_id and this endpoint has not answered yet.

None once it has been answered, and for an identifier the peer has announced nothing under. The record itself lives on past the answer, because an announcement that was accepted is not over until it is withdrawn or cancelled.

pub fn pending_announce_count(&self) -> usize

How many announcements the peer has made that are still waiting for an answer.

fn inbound_announce_id(&self, namespace: &TrackNamespace) -> Option<u64>

The identifier of a live announcement the peer made for namespace.

Section 8.22: “The publisher sends the UNANNOUNCE control message to indicate its intent to stop serving new subscriptions for tracks within the provided Track Namespace.” and Section 8.23 says what a cancellation is for: the subscriber “will stop sending new subscriptions for tracks within the provided Track Namespace”. Both name a namespace where the ANNOUNCE they are about named a Request ID, so one record has to be reachable both ways.

It is stored under the identifier, which is unique, and found by namespace with a scan of the same map. A second map from namespace to identifier would be quicker and could fall out of step with the first; there is nothing here for it to disagree with.

An announcement that has ended is skipped, so a namespace announced again after being withdrawn finds the live one. Which of two live announcements for the same namespace is found is not decided here, because no sentence in the draft makes a second one for a namespace already announced an error.

pub fn send_announce_ok( &mut self, request_id: VarInt, ) -> Result<ControlMessage, EndpointError>

Build the ANNOUNCE_OK accepting an announcement the peer made.

Section 5.2: “A subscriber MUST send exactly one ANNOUNCE_OK or ANNOUNCE_ERROR in response to an ANNOUNCE. The publisher SHOULD close the session with a protocol error if it receives more than one.”

One answer and no second one: the flow moves on the first, and a second call finds a record that has left Pending.

Errors

EndpointError::UnknownRequest if the peer has announced nothing under that identifier, and the namespace flow’s own InvalidTransition for an announcement already answered.

pub fn send_announce_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<ControlMessage, EndpointError>

Build the ANNOUNCE_ERROR refusing an announcement the peer made.

The same sentence in Section 5.2 answers both ways: one message back and no second one, whichever of the two it is.

Errors

EndpointError::UnknownRequest if the peer has announced nothing under that identifier, and the namespace flow’s own InvalidTransition for an announcement already answered.

pub fn receive_unannounce( &mut self, msg: &Unannounce, ) -> Result<(), EndpointError>

Process an incoming UNANNOUNCE, ending the announcement the peer made.

Section 8.22: “The publisher sends the UNANNOUNCE control message to indicate its intent to stop serving new subscriptions for tracks within the provided Track Namespace.”

The announcement it ends is the peer’s, so the record it reads is the one this endpoint keeps of what the peer announced. An announcement this endpoint made is withdrawn by Self::unannounce, which is the same message travelling the other way.

Errors

EndpointError::UnknownPeerNamespace if the peer has no live announcement for that namespace, and the namespace flow’s own InvalidTransition for one this endpoint never accepted.

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

Build the ANNOUNCE_CANCEL revoking an acceptance.

Section 7.3 names what a cancellation revokes: a namespace “it previously responded ANNOUNCE_OK to”. Section 8.23 says what it does: the subscriber “will stop sending new subscriptions for tracks within the provided Track Namespace”.

Previously responded ANNOUNCE_OK to is a state, and it is Active: an announcement reaches it by being accepted and no other way. One still waiting for an answer, one refused and one already ended are all refused here rather than sent.

The announcement is the peer’s. An announcement this endpoint made is not cancelled by its own publisher; the peer cancels it, and that arrives at Self::receive_announce_cancel.

Errors

EndpointError::UnknownPeerNamespace if the peer has no live announcement for that namespace, and the namespace flow’s own InvalidTransition for one this endpoint never accepted.

pub fn track_status_request( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, ) -> Result<(VarInt, ControlMessage), EndpointError>

Send a TRACK_STATUS_REQUEST message. Returns the allocated request ID alongside the control message so the caller can correlate replies.

pub fn receive_track_status( &mut self, msg: &TrackStatus, ) -> Result<(), EndpointError>

Process an incoming TRACK_STATUS reply.

pub fn receive_track_status_request( &mut self, msg: &TrackStatusRequest, ) -> Result<(), EndpointError>

Process an incoming TRACK_STATUS_REQUEST, recording what the peer asked about.

Section 8.17: “A potential subscriber sends a ‘TRACK_STATUS_REQUEST’ message on the control stream to obtain information about the current status of a given track.”

Answering is the application’s to do, and it needs both the request and somewhere to answer from. This draft gave the request a Request ID, so that is what the record is filed under and what the answer names.

Errors

The session error when the session is not established.

pub fn pending_track_status_request( &self, request_id: VarInt, ) -> Option<&TrackStatusRequest>

The TRACK_STATUS_REQUEST the peer sent under request_id and this endpoint has not answered yet.

None once it has been answered, and for an identifier this session has carried no track status request under.

pub fn pending_track_status_request_count(&self) -> usize

How many track statuses the peer has asked about that are still waiting for an answer.

pub fn send_track_status( &mut self, request_id: VarInt, status_code: VarInt, largest_location: Location, parameters: Vec<KeyValuePair>, ) -> Result<ControlMessage, EndpointError>

Build the TRACK_STATUS answering a request the peer sent.

Section 8.17 leaves the answering end no discretion about whether to answer: “A TRACK_STATUS message MUST be sent in response to each TRACK_STATUS_REQUEST.” What it bounds is how many, and that half is what the record carries: the request leaves Pending on the first answer, so a second call finds nothing left to answer.

Section 8.18 says the identifier this message carries is “The Request ID of the TRACK_STATUS_REQUEST this message is replying to”, which is why the answer is asked for by the identifier the request arrived under and not by the track it named.

Errors

EndpointError::UnknownRequest if the peer has asked nothing under that identifier, and the flow’s own InvalidTransition for a request already answered.

pub fn receive_requests_blocked( &mut self, msg: &RequestsBlocked, ) -> Result<(), EndpointError>

Process an incoming REQUESTS_BLOCKED message.

Draft-11 renames draft-10’s SUBSCRIBES_BLOCKED to REQUESTS_BLOCKED so the peer can explicitly report that a new request id would exceed our advertised maximum. The endpoint records the peer’s reported maximum; acting on it (issuing a new MAX_REQUEST_ID) is up to the caller.

pub fn peer_reported_max_request_id(&self) -> Option<VarInt>

The maximum request id that the peer most recently reported in a REQUESTS_BLOCKED message, if any.

pub fn receive_request( &mut self, msg: &ControlMessage, ) -> Result<(), EndpointError>

Hold a Request ID a peer allocated to the rules Section 8.1 states about it.

“The client’s Request ID starts at 0 and are even and the server’s Request ID starts at 1 and are odd. The Request ID increments by 2 … If an endpoint receives a Request ID that is not valid for the peer, it MUST close the session with Invalid Request ID.” Parity says which end may have chosen it; the ceiling this endpoint advertised says how far the peer may go.

This is the call site Self::validate_peer_request_id did not have. The rule was implemented and then applied to nothing, so a peer could open requests with ids from this endpoint’s own half of the space, or past the ceiling it had advertised, and neither was noticed.

A message that is a response rather than a request carries the id of a request this endpoint made, so it is not checked here - it is checked by finding the state machine it names. SUBSCRIBE_UPDATE is the same case on this draft: its one Request ID field names the subscription it modifies rather than opening a request of its own, which is why Section 8.1 does not list it among the messages that step the sequence.

The list below is the rule, not a convenience

Every message named here spends one of the peer’s Request IDs, and nothing else does. That makes the list load-bearing in a way it was not before the sequence was tracked: a request left out of it spends an ID this endpoint never counts, so the peer’s next request looks like a skip and a conforming session is closed over it.

Errors

The request-id errors, for a wrong parity, an id at or above the advertised ceiling, or one that is not the next in the peer’s sequence.

pub fn receive_message( &mut self, msg: ControlMessage, ) -> Result<(), EndpointError>

Dispatch an incoming control message to the appropriate handler.