Struct Endpoint 

pub struct Endpoint {

    role: Role,
    session: SessionStateMachine,
    request_ids: RequestIdAllocator,
    advertised_max_id: u64,
    subscriptions: HashMap<u64, Mutex<SubscriptionStateMachine>>,
    fetches: HashMap<u64, FetchStateMachine>,
    subscribe_namespaces: HashMap<u64, SubscribeNamespaceStateMachine>,
    inbound_subscribe_namespaces: HashMap<u64, InboundSubscribeNamespace>,
    announces: HashMap<u64, AnnounceStateMachine>,
    inbound_announces: HashMap<u64, InboundAnnounce>,
    announce_ids: HashMap<Vec<Vec<u8>>, u64>,
    subscribe_namespace_ids: HashMap<Vec<Vec<u8>>, u64>,
    track_statuses: HashMap<u64, TrackStatusStateMachine>,
    inbound_track_statuses: HashMap<u64, InboundTrackStatus>,
    inbound_publishes: HashMap<u64, InboundPublish>,
    publishes: HashMap<u64, PublishStateMachine>,
    inbound_subscribes: HashMap<u64, InboundSubscribe>,
    inbound_fetches: HashMap<u64, InboundFetch>,
    negotiated_version: Option<VarInt>,
    offered_versions: Vec<VarInt>,
    goaway_uri: Option<Vec<u8>>,
    peer_reported_max_request_id: Option<VarInt>,
    forwarding_preferences: Mutex<TrackForwardingPreferences>,
    malformed: Mutex<MalformedTracks>,
    locations: Arc<Mutex<TrackLocations>>,
    track_bindings: HashMap<u64, TrackBinding>,
}

Unified draft-13 MoQT endpoint wrapping session lifecycle, request ID allocation, and all per-flow state machines (subscriptions, fetches, announces, subscribe-namespaces, 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, Mutex<SubscriptionStateMachine>>

fetches: HashMap<u64, FetchStateMachine>

subscribe_namespaces: HashMap<u64, SubscribeNamespaceStateMachine>

inbound_subscribe_namespaces: HashMap<u64, InboundSubscribeNamespace>

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

Kept apart from subscribe_namespaces, 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_namespace_ids: HashMap<Vec<Vec<u8>>, u64>

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

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.

inbound_publishes: HashMap<u64, InboundPublish>

The subscriptions the peer opened with PUBLISH, keyed by the request id it opened them under.

The record outlives the answer. Section 4.1 makes a PUBLISH the start of a subscription rather than a question to be answered and forgotten: “Once either of these sequences is successful, the subscription can be updated by the subscriber using SUBSCRIBE_UPDATE, terminated by the subscriber using UNSUBSCRIBE, or terminated by the publisher using SUBSCRIBE_DONE.” Everything after the answer needs the record - which Track Alias is spoken for, and whether a message naming this request is still about anything.

The offers this endpoint makes itself are in publishes beside this, which holds their state and not their message: the two are answered by opposite ends.

publishes: HashMap<u64, PublishStateMachine>

The subscriptions this endpoint opened with PUBLISH, keyed by the Request ID it allocated for each.

The state and nothing else, because the offer was built here and the caller was handed it. That is the one asymmetry with inbound_publishes and it is the same one every other pair of records here has.

inbound_subscribes: HashMap<u64, InboundSubscribe>

Subscriptions the peer opened with SUBSCRIBE, by Request ID.

Section 8.10 puts SUBSCRIBE_UPDATE in the subscriber’s hands - “A subscriber sends a SUBSCRIBE_UPDATE to a publisher to modify an existing subscription” - so an update that arrives names a subscription in here and never one in subscriptions.

The record holds the SUBSCRIBE itself and not only its state, because the answer is built out of the request: the track a SUBSCRIBE_OK is about is named nowhere else, and the alias it hands out has to be judged against that name.

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.

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.

forwarding_preferences: Mutex<TrackForwardingPreferences>

What Full Track Name the peer has attached each Track Alias to, per Request ID.

Sections 8.8 and 8.13 forbid one alias naming two tracks at once, and the “at once” is what makes this a table rather than a set: an alias the peer used for a track whose subscription has ended is free again. The table therefore 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.

malformed: Mutex<MalformedTracks>

The tracks this endpoint has found malformed and withdrawn from.

Behind a lock for the same reason the record above it is: it is written from the data plane, which holds a shared reference.

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>

Implementations

impl Endpoint

pub fn new(role: Role) -> Self

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

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

The Track Alias the peer attached to request_id, once it has named one.

Answers for a subscription this endpoint asked for from the moment its SUBSCRIBE_OK arrives, and for one the peer offered from the moment its PUBLISH does. None before that, and for a Request ID this session has no track for.

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

The refusal Sections 8.8 and 8.13 require when alias already names a different track whose subscription is still live, or None when it is free.

Why the set is read rather than kept

The draft says “an active subscription”, and this endpoint’s subscription state machine has exactly that state: a subscription reaches Active on SUBSCRIBE_OK and leaves it on the message that ends the flow. Asking it is what makes an alias free again the moment 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 leave the alias held forever and refuse the peer’s next, conforming, use of it.

Why the request’s own binding is skipped

A SUBSCRIBE_OK is judged before its own alias is written down, so the skip is not what keeps it from finding itself. A PUBLISH is: it carries its alias and its track in the one message, and a second PUBLISH under a Request ID already bound is refused by the duplicate-Request-ID rule before it reaches here.

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

Whether the request that owns a binding still has a live subscription.

The two kinds are answered by two different records because the two sequences Section 4.1 names end in different places: a SUBSCRIBE this endpoint made is live once its SUBSCRIBE_OK arrives, a PUBLISH the peer made once this endpoint has answered PUBLISH_OK.

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

The request already using alias for a track other than (namespace, name), or None when this endpoint may give the alias to that track.

Separate from Self::conflicting_track_alias because the two answer different questions about the same table. That one judges a message that has arrived and ends the session over it; this one judges one that has not been built and declines to build it.

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 the two rules its track’s record answers: Section 9.2.1.1’s protocol error when an end-of-track object says the track ended somewhere the track has already passed, and Section 2.5’s Malformed Track when an object arrives after the place an end-of-track object put the end.

&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 subscription( &self, id: u64, ) -> Option<MutexGuard<'_, SubscriptionStateMachine>>

The subscription id opened, locked for a read or a transition.

&self, which is the whole point of the lock. Section 2.5’s answer to a Malformed Track is “it MUST UNSUBSCRIBE from the Track”, and the conditions that make a track malformed are detected on the data plane, where this endpoint is reached through a shared reference. A flow that could only be moved through &mut self would leave that sentence unanswerable from the only place it is ever read.

Hold one guard at a time. Nothing here takes a second while a first is live, and Self::withdraw_malformed_track collects the requests it is going to move before it moves any of them for that reason.

pub fn malformed_track( &self, namespace: &TrackNamespace, name: &[u8], ) -> Option<MalformedTrackCondition>

The condition a track was withdrawn for, or None for a track this endpoint has found nothing wrong with.

The withdrawal happens without the application asking for it, so this is how an application that missed the error learns why a subscription it never ended has ended.

Why this takes a track and not the alias the object carried

An alias only means anything through a live binding, and the withdrawal ends the binding it would have been resolved through. An accessor taking an alias would therefore answer None from the instant it had something to say, which is worse than not existing. The record is keyed on the track, and so is this.

pub fn withdraw_malformed_track( &self, alias: u64, condition: MalformedTrackCondition, ) -> Vec<ControlMessage>

Withdraw from a track a condition has just shown to be malformed, and give back the UNSUBSCRIBEs Section 2.5 asks for.

“When a subscriber detects a Malformed Track, it MUST UNSUBSCRIBE from the Track and SHOULD deliver an error to the application.” This is the first half. The second is the error the detecting path returns, which is why nothing here reports anything: a caller that gets an empty list back is not being told the track was fine.

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

What comes back, and what does not

One UNSUBSCRIBE for each request through which this endpoint is receiving the track — a SUBSCRIBE it sent, and a PUBLISH the peer sent that it accepted, which Section 4.1 ends with the same message. A request that names the track the other way round is not one of those: a peer subscribing to this endpoint makes it the publisher, and a publisher does not unsubscribe from what it is serving.

Empty for an alias no live binding names — there is no track to withdraw from — and for a track whose only requests are in a state that has no UNSUBSCRIBE in it. A subscription still waiting for its SUBSCRIBE_OK is the case that reaches that last one, and it holds no alias for an object to have arrived under.

What makes the answer happen once

Not the record. The requests this withdraws end here, and a request that has ended has no second UNSUBSCRIBE in it, so a publisher that goes on mixing a track’s framing is answered once however many objects it sends. The record is read afterwards, by an application asking why a track it never gave up was given up.

Which leaves the case the two answers differ on: an application that subscribes to the same track again. That request has never been withdrawn from, and a publisher mixing its framing again has broken the sentence again, so it is withdrawn from too.

fn withdraw_one(&self, id: u64) -> Option<ControlMessage>

The UNSUBSCRIBE ending one request, or None when that request is not one this endpoint receives a track through or is not in a state that can be unsubscribed from.

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

Whether a binding’s request has put its alias in play at all.

Broader than Self::binding_is_established, and the two sentences are why. What an endpoint must close over is qualified - “a different track with an active subscription” - and the prohibition Section 8.8 opens with is not: “The same Track Alias MUST NOT be used to refer to two different Tracks simultaneously.” Once an alias has been handed out, giving it to a second track is what that sentence forbids, answered or not.

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

The conflict a SUBSCRIBE_OK’s alias has with the tracks already bound.

Separate from Self::conflicting_track_alias because the track a SUBSCRIBE_OK is about is not in the SUBSCRIBE_OK: it is the one this endpoint’s own SUBSCRIBE asked for, which is why the request has to be looked up before the alias can be judged.

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_namespace_count(&self) -> usize

Returns the number of active subscribe-namespace 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_namespace: TrackNamespace, track_name: Vec<u8>, subscriber_priority: u8, group_order: GroupOrder, filter_type: FilterType, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

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

LargestObject is a reasonable default. 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 that refusal this call would hand back a message whose filter announces fields the message does not carry, which the encoder rejects.

Draft-13 note: the track alias is no longer chosen by the subscriber — it is returned by the publisher in SUBSCRIBE_OK. Callers that need the alias should inspect track_alias_for after the OK arrives.

pub 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, 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: end_group present means AbsoluteRange and absent means AbsoluteStart. So the message cannot name a filter whose fields it does not carry.

fn subscribe_inner( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, subscriber_priority: u8, group_order: GroupOrder, filter_type: FilterType, start: Option<(VarInt, VarInt)>, end_group: Option<VarInt>, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

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

Process an incoming SUBSCRIBE_OK.

Draft-13 carries the publisher-assigned track_alias on SUBSCRIBE_OK, which is recorded here so callers can retrieve it via Endpoint::track_alias_for.

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.

Section 4.1 gives the subscriber this for a subscription that came either way round, so a request the peer opened with PUBLISH ends here too. The two kinds cannot collide: a Request ID belongs to whichever endpoint allocated it, and the two halves of the space have opposite least significant bits.

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: this draft’s SUBSCRIBE_UPDATE has one Request ID field and it names the subscription being modified, which is why Section 8.1 leaves the message out of the list that steps the peer’s sequence.

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 sends a SUBSCRIBE_UPDATE to a publisher to modify 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 Request ID names no subscription the peer has opened in this session, with the session already moved to Closed, and the subscription flow’s own InvalidTransition when it names one that has already ended.

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

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

Section 4.1 gives the publisher this for a subscription that came either way round, so one the peer opened with PUBLISH ends here too.

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

Process an incoming SUBSCRIBE, recording the subscription it opens.

The Request ID has already been checked by Self::receive_request, which every request message passes through before its own handler. There is no Track Alias to judge here: Section 8.8 carries it in the SUBSCRIBE_OK, so this endpoint chooses it, and it is judged when the answer is built.

Errors

The session error when the session is not established, and the subscription flow’s own InvalidTransition for a second SUBSCRIBE under a Request ID already carrying one.

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. The record itself lives on for as long as the session does, which is what lets an update say whether the request it names has ever existed.

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, track_alias: VarInt, expires: VarInt, group_order: GroupOrder, parameters: Vec<KeyValuePair>, ) -> Result<ControlMessage, EndpointError>

Build the SUBSCRIBE_OK accepting a subscription the peer opened, giving its track a Track Alias.

Errors

EndpointError::UnknownRequest if the peer opened no subscription under that identifier, EndpointError::TrackAliasInUse if a live track of this endpoint’s already holds the alias, and EndpointError::Subscription if the request has already been answered.

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

Build the SUBSCRIBE_ERROR rejecting a subscription the peer opened.

No Track Alias goes back with it. This draft moved the field out of SUBSCRIBE_ERROR along with the retry the earlier ones offered, so a refusal here says only that the request failed.

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, stream_count: VarInt, reason_phrase: Vec<u8>, ) -> Result<ControlMessage, EndpointError>

Build the SUBSCRIBE_DONE ending a subscription this endpoint publishes, whichever of the two sequences opened it.

Errors

EndpointError::UnknownRequest if this session is carrying no such subscription under that identifier, EndpointError::Subscription if it is one the peer opened and this endpoint has not accepted or has already ended, and EndpointError::PublishFlow for the same of one this endpoint opened.

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

Process an incoming UNSUBSCRIBE, ending a subscription this endpoint publishes and freeing the Track Alias it held.

Section 8.11: “A subscriber issues a UNSUBSCRIBE message to a publisher indicating it is no longer interested in receiving media for the specified track and requesting that the publisher stop sending Objects as soon as possible.” The message travels from subscriber to publisher, so what it can end is whatever this endpoint publishes - and Section 4.1 gives that two sources, not one: “A subscription can be initiated by either a publisher or a subscriber.”

Errors

EndpointError::UnknownRequest if this endpoint publishes no subscription under that identifier, EndpointError::Subscription if it is one the peer opened that this endpoint never accepted or has already ended, and EndpointError::PublishFlow for the same of one this endpoint opened.

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, parameters: Vec<KeyValuePair>, ) -> 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_request_id: VarInt, joining_start: VarInt, parameters: Vec<KeyValuePair>, ) -> 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.16 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 Location, and End Location 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.16.1 has the publisher set “Fetch Start Location: {Subscribe Largest Location.Group - Joining Start, 0}”, which makes it a count of groups back from the live edge.

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.16 answers that at the publisher, “it MUST respond with a Fetch Error with code Invalid Joining Request ID”, and this endpoint is the subscriber. The code is 0x7 whichever of its two names is read: the FETCH_ERROR table in Section 8.18 lists it under the name above and then describes it one line down as Invalid Joining Subscribe ID.

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

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

Section 8.16.2 is the whole of the difference: “Identical to the Relative Joining fetch except that Fetch Start Location.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 Location, 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_request_id: VarInt, joining_start: VarInt, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

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

Section 8.16 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.16 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.16: “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 Request 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.1 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.16 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.19: 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_namespace( &mut self, track_namespace_prefix: TrackNamespace, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

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

Section 8.28 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_namespace_ok( &mut self, msg: &SubscribeNamespaceOk, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE_NAMESPACE_OK.

pub fn receive_subscribe_namespace_error( &mut self, msg: &SubscribeNamespaceError, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE_NAMESPACE_ERROR.

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

Send an UNSUBSCRIBE_NAMESPACE message.

pub fn receive_subscribe_namespace( &mut self, msg: &SubscribeNamespace, ) -> Result<(), EndpointError>

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

Section 8.28: “The subscriber sends the SUBSCRIBE_NAMESPACE control message to a publisher to request the current set of matching announcements and established subscriptions, 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.

Only ones that have not ended count: the sentence weighs the arriving prefix against “an active SUBSCRIBE_NAMESPACE”, so one the peer has withdrawn and one this endpoint refused are both past. Drafts 07 through 11 say “an earlier” instead and count those too.

One that has arrived and has not been answered does count. It is not active yet, but this endpoint is the one about to make it so, and accepting both would leave the session holding exactly the pair the sentence exists to prevent.

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_namespace( &self, request_id: VarInt, ) -> Option<&SubscribeNamespace>

The SUBSCRIBE_NAMESPACE 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_namespace_count(&self) -> usize

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

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

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

Section 8.31 names a Track Namespace Prefix where the SUBSCRIBE_NAMESPACE 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_namespace_ok( &mut self, request_id: VarInt, ) -> Result<ControlMessage, EndpointError>

Build the SUBSCRIBE_NAMESPACE_OK accepting a namespace subscription the peer made.

Section 5.1: “A publisher MUST send exactly one SUBSCRIBE_NAMESPACE_OK or SUBSCRIBE_NAMESPACE_ERROR in response to a SUBSCRIBE_NAMESPACE.”

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_namespace_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<ControlMessage, EndpointError>

Build the SUBSCRIBE_NAMESPACE_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_namespace( &mut self, msg: &UnsubscribeNamespace, ) -> Result<(), EndpointError>

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

Section 5.1: “An UNSUBSCRIBE_NAMESPACE withdraws a previous SUBSCRIBE_NAMESPACE.”

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_namespace, 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, parameters: Vec<KeyValuePair>, ) -> 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.23: “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.26: “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.27 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.26: “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.27 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( &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, ControlMessage), EndpointError>

Send a TRACK_STATUS message asking after a track, and record it.

Named for the message this draft carries. TRACK_STATUS_REQUEST was the earlier drafts’ name for the request and is not a message here, so a method carrying that name would say the wrong thing about which of the two shapes goes on the wire.

Returns the allocated request ID alongside the control message.

pub fn receive_track_status_ok( &mut self, msg: &TrackStatusOk, ) -> Result<(), EndpointError>

Process an incoming TRACK_STATUS_OK reply.

Section 8.21 gives this draft a pair of answers where the earlier ones had a single TRACK_STATUS, so the acceptance and the refusal walk separate edges: a machine told only that an answer arrived cannot say which of the two it was.

pub fn receive_track_status_error( &mut self, msg: &TrackStatusErrorMsg, ) -> Result<(), EndpointError>

Process an incoming TRACK_STATUS_ERROR reply.

Section 8.22: “The publisher sends a TRACK_STATUS_ERROR control message in response to a failed TRACK_STATUS message.”

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

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

Section 8.20: the receiver of one “treats it identically as if it had received a SUBSCRIBE message, except it does not create downstream subscription state or send any Objects”. The exception is why this does not reach for the subscriptions the peer has opened: nothing that names a subscription is to find this request, and the surest way to hold to that is for it never to be one.

Errors

The session error when the session is not established.

pub fn pending_track_status(&self, request_id: VarInt) -> Option<&TrackStatus>

The TRACK_STATUS 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 under.

pub fn pending_track_status_count(&self) -> usize

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

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

Build the TRACK_STATUS_OK accepting a track status the peer asked for.

Section 8.21: “The publisher sends a TRACK_STATUS_OK control message in response to a successful TRACK_STATUS message”, populating it “exactly as it would have populated a SUBSCRIBE_OK, setting Track Alias to 0”.

The alias is not a parameter for that reason: the one value the draft allows is the one this builds, so no caller can put another on the wire. The sentence after it is what keeps the alias out of the table this endpoint judges aliases against - “It is not considered an error if Track Alias 0 is already in use by an active subscription” - so nothing here consults that table and nothing here adds to it. An alias that names no track cannot collide with one that does.

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 send_track_status_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<ControlMessage, EndpointError>

Build the TRACK_STATUS_ERROR refusing a track status the peer asked for.

Section 8.22: “The publisher sends a TRACK_STATUS_ERROR control message in response to a failed TRACK_STATUS message.”

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.

The peer explicitly reports 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 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, ControlMessage), EndpointError>

Offer the peer a subscription to a track this endpoint publishes. Allocates a Request ID.

Section 4.1: “A subscription can be initiated by either a publisher or a subscriber. A publisher initiates a subscription to a track by sending the PUBLISH message. The subscriber either accepts or rejects the subscription using PUBLISH_OK or PUBLISH_ERROR.”

There is no content_exists parameter because it is not a choice. Section 8.13 makes it a flag for whether the field after it is there at all - “1 if an object has been published on this track, 0 if not. If 0, then the Largest Group ID and Largest Object ID fields will not be present” - so it is derived from largest_location, and the pair cannot be built disagreeing.

forward is the subscription’s initial Forward State, which Section 4.1 gives to whichever end opened it: “The initiator of the subscription sets the initial Forward State in either PUBLISH or SUBSCRIBE.” Section 8.13 says what the peer may then assume of it: “1 indicates the publisher will start transmitting objects immediately, even before PUBLISH_OK.”

Errors

EndpointError::TrackAliasInUse when some live request of this session already holds track_alias for a different track, which Section 8.13 forbids outright - “The same Track Alias MUST NOT be used to refer to two different Tracks simultaneously” - and which the subscriber answers by closing the session. Judged before the Request ID is allocated, so a refused offer spends nothing. Also the session error when the session is not established, and the request-id error when this endpoint has no identifier left to spend.

pub fn receive_publish(&mut self, msg: &Publish) -> Result<(), EndpointError>

Process an incoming PUBLISH message, which opens a subscription this endpoint is the subscriber of.

The message is kept so the application can answer it with Endpoint::send_publish_ok or Endpoint::send_publish_error under the same request id, and the record it is kept in outlives that answer: Section 4.1 gives the subscription that follows an UNSUBSCRIBE and a SUBSCRIBE_DONE, and both name this request.

Errors

EndpointError::DuplicateTrackAlias when the Track Alias offered is one a different live track already holds, which ends the session, and the publish flow’s own InvalidTransition for a second PUBLISH under a request id already carrying one.

pub fn pending_publish(&self, request_id: VarInt) -> Option<&Publish>

The inbound PUBLISH for a request id that is still waiting to be answered.

Stops answering once the PUBLISH has been answered, which is the whole of what “pending” means: the record itself lives on for as long as the subscription does.

pub fn pending_publish_count(&self) -> usize

Number of PUBLISH requests received but not yet responded to.

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

Generate a PUBLISH_OK response for a previously received PUBLISH, which establishes the subscription it opened.

Errors

EndpointError::UnknownRequest when no PUBLISH arrived under that id, and the publish flow’s own InvalidTransition for a second answer: Section 4.1 says “A subscriber MUST send exactly one PUBLISH_OK or PUBLISH_ERROR in response to a PUBLISH.”

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

Generate a PUBLISH_ERROR response for a previously received PUBLISH, which ends the subscription it opened before it was established.

Errors

EndpointError::UnknownRequest when no PUBLISH arrived under that id, and the publish flow’s own InvalidTransition for a second answer: Section 4.1 says “A subscriber MUST send exactly one PUBLISH_OK or PUBLISH_ERROR in response to a PUBLISH.”

pub fn receive_publish_ok( &mut self, msg: &PublishOk, ) -> Result<(), EndpointError>

Process an incoming PUBLISH_OK, which establishes the subscription this endpoint offered under that Request ID.

Errors

EndpointError::UnknownRequest when this endpoint has offered nothing under that identifier, and the publish flow’s own InvalidTransition for an offer that has been answered already: Section 4.1 says “A subscriber MUST send exactly one PUBLISH_OK or PUBLISH_ERROR in response to a PUBLISH. The peer SHOULD close the session with a protocol error if it receives more than one.” The verb there is SHOULD, so the second answer is reported rather than acted on, and the caller decides.

pub fn receive_publish_error( &mut self, msg: &PublishError, ) -> Result<(), EndpointError>

Process an incoming PUBLISH_ERROR, which ends the subscription this endpoint offered under that Request ID before it was established.

Section 4.1: “Objects MUST NOT be sent for requests that end with an error.” The Track Alias the offer named is free again from here, because a binding reads liveness off this record rather than keeping a second copy of it.

Errors

As Self::receive_publish_ok.

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. PUBLISH is here and not in Section 8.1’s enumeration, which predates the message; its own Request ID field refers the reader to that section like every other request’s does.

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.