Struct Endpoint 

pub struct Endpoint {

    role: Role,
    session: SessionStateMachine,
    request_ids: RequestIdAllocator,
    advertised_max_id: u64,
    subscriptions: HashMap<u64, Mutex<SubscriptionStateMachine>>,
    fetches: HashMap<u64, Mutex<FetchStateMachine>>,
    fetch_tracks: HashMap<u64, FetchTrack>,
    subscribe_namespace_prefixes: HashMap<u64, TrackNamespace>,
    subscribe_namespaces: HashMap<u64, SubscribeNamespaceStateMachine>,
    inbound_subscribe_namespaces: HashMap<u64, InboundSubscribeNamespace>,
    publish_namespaces: HashMap<u64, PublishNamespaceStateMachine>,
    publish_namespace_namespaces: HashMap<u64, TrackNamespace>,
    inbound_publish_namespaces: HashMap<u64, InboundPublishNamespace>,
    track_statuses: HashMap<u64, TrackStatusStateMachine>,
    inbound_track_statuses: HashMap<u64, InboundTrackStatus>,
    publishes: HashMap<u64, Mutex<PublishStateMachine>>,
    inbound_publishes: HashMap<u64, Publish>,
    inbound_subscribes: HashMap<u64, InboundSubscribe>,
    inbound_fetches: HashMap<u64, InboundFetch>,
    negotiated_version: Option<VarInt>,
    offered_versions: Vec<VarInt>,
    goaway_uri: Option<Vec<u8>>,
    forwarding_preferences: Mutex<TrackForwardingPreferences>,
    malformed: Mutex<MalformedTracks>,
    locations: Arc<Mutex<TrackLocations>>,
    track_bindings: HashMap<u64, TrackBinding>,
}

Unified MoQT endpoint wrapping session lifecycle, request ID allocation, and all per-request state machines (subscriptions, fetches, namespaces).

Fields

role: Role

session: SessionStateMachine

request_ids: RequestIdAllocator

advertised_max_id: u64

Tracks the MAX_REQUEST_ID we have advertised to the peer (for monotonic enforcement).

subscriptions: HashMap<u64, Mutex<SubscriptionStateMachine>>

Each subscription this endpoint opened, behind a lock apiece.

The lock is what lets the data plane end one. Section 2.5’s answer to a Malformed Track is a message per request, the conditions that make a track malformed are detected where objects arrive, and objects arrive through a shared reference to the connection carrying them.

fetches: HashMap<u64, Mutex<FetchStateMachine>>

Each fetch this endpoint made, behind a lock apiece, for the reason the subscriptions above are: the same sentence ends a fetch for a malformed track and ends it from the same place.

fetch_tracks: HashMap<u64, FetchTrack>

The track each fetch this endpoint made is for.

Why this is not in track_bindings

That table exists to answer questions about Track Aliases, and a fetch has none: its objects arrive on a stream that opens by naming the Request ID, so no alias rule can ever be about a fetch. An entry that can never hold an alias would be one every reader of that table had to learn to skip.

Why a Joining Fetch is resolved here rather than when it is read

A Joining Fetch names no track. It names the subscription it joins, and Section 9.16.2 takes the rest from there: “A publisher receiving a Joining Fetch uses properties of the associated Subscribe to determine the Track Namespace, Track Name and End Location such that it is contiguous with the associated Subscribe.” So its track is the joined subscription’s, and it is looked up once, as the fetch is made.

The other place to look it up is the withdrawal, through the request the fetch joined, and that fails in the case a Joining Fetch exists for: one fills a buffer behind the live edge, so it outlives the subscription it joined, and the lookup would come up empty exactly while there was still a fetch to cancel.

subscribe_namespace_prefixes: HashMap<u64, TrackNamespace>

The namespace prefix each SUBSCRIBE_NAMESPACE this endpoint sent asked about, which the state machine beside it does not hold.

Read by Endpoint::own_prefix_overlap and by nothing else. The withdrawal names the Request ID here, so until the rule about overlapping prefixes was judged there was nothing to remember a prefix for.

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.

publish_namespaces: HashMap<u64, PublishNamespaceStateMachine>

publish_namespace_namespaces: HashMap<u64, TrackNamespace>

The namespace each announcement this endpoint made names, so PUBLISH_NAMESPACE_DONE and PUBLISH_NAMESPACE_CANCEL, which carry a namespace and no Request ID, can find the one they are about.

inbound_publish_namespaces: HashMap<u64, InboundPublishNamespace>

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

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.

publishes: HashMap<u64, Mutex<PublishStateMachine>>

Both directions’ publish flows, behind a lock apiece.

A subscription the peer opened with PUBLISH is one of the two ways this endpoint receives a track, so the withdrawal a Malformed Track calls for has to reach these as well as the subscriptions above.

inbound_publishes: HashMap<u64, Publish>

The PUBLISH each offer the peer made arrived as, keyed by the Request ID it carries.

The state of each one stays in publishes beside this endpoint’s own offers, because every step after arrival names one Request ID and a Request ID the peer allocated can never be one this endpoint allocated. What a map of state machines cannot hold is the offer itself, and the offer is what the answer is decided from. Section 5.1: “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.”

track_bindings already keeps the Full Track Name and the Track Alias, because the rules about aliases read them. The delivery order, the largest location and the parameters are here and nowhere else.

inbound_subscribes: HashMap<u64, InboundSubscribe>

Subscriptions the peer opened with SUBSCRIBE, by Request ID.

Section 9.10 puts the update in the subscriber’s hands, so one 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. A PUBLISH needs no such record because it carries its track and its alias in the one message.

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

forwarding_preferences: Mutex<TrackForwardingPreferences>

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

Sections 9.8 and 9.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>>

Where each track this endpoint receives has ended, once an end-of-track object has said so.

Behind an Arc because the stream that reads a track’s objects holds a handle onto it and the endpoint cannot be reached from there.

track_bindings: HashMap<u64, TrackBinding>

Implementations

impl Endpoint

pub fn new(role: Role) -> Self

Create a new endpoint with 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 9.8 and 9.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 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 Section 2.5’s Malformed Track when it arrived after the place an end-of-track object put the end.

One rule and not two. The placement rule drafts 08 through 13 state about an end-of-track object is not in this draft, so an object that ends a track here is judged against nothing and only settles where the track stopped.

&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 10’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 answers a Malformed Track with a message per request, 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.

fn publish_flow(&self, id: u64) -> Option<MutexGuard<'_, PublishStateMachine>>

The publish flow id opened, locked for a read or a transition.

One map for both directions, which is what makes this one accessor: the offer this endpoint made and the offer the peer made are the same flow read from opposite ends, and Request ID parity keeps them apart.

fn fetch_flow(&self, id: u64) -> Option<MutexGuard<'_, FetchStateMachine>>

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

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 messages Section 2.5 asks for.

“When a subscriber detects a Malformed Track, it MUST UNSUBSCRIBE any subscription and FETCH_CANCEL any fetch for that Track from that publisher, 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 message for each request through which this endpoint is receiving the track: an UNSUBSCRIBE for a SUBSCRIBE it sent and for a PUBLISH the peer sent that it accepted, and a FETCH_CANCEL for each fetch it made for that track. 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 ending of that kind left in it.

Two records, one scan

Subscriptions are found through the alias table and fetches through their own, because a fetch never holds an alias. Both are keyed by Request ID and a Request ID names one request, so the two lists cannot overlap and the withdrawal below can tell from the id alone which message a request takes.

What makes the answer happen once

Not the record. The requests this withdraws end here, and a request that has ended has no second ending 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 message 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 ended that way.

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 a subscriber must close over is qualified - “the same Track Alias as a different track with an active subscription” - and the prohibition on the publisher is not: “The same Track Alias MUST NOT be used to refer to two different Tracks simultaneously.” Once a PUBLISH carrying an alias has been sent, giving that alias to a second track is what that sentence forbids, answered or not.

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 state machines because the two sequences Section 5.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 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_publish_namespace_count(&self) -> usize

Returns the number of active publish-namespace state machines.

pub fn active_track_status_count(&self) -> usize

Returns the number of active track status state machines.

pub fn active_publish_count(&self) -> usize

Returns the number of active publish 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 9.1 closes the session with INVALID_REQUEST_ID on a request ID “that is not valid for the peer”, and Section 9.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 9.5: a Request ID “equal to 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 9.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 (0x3) 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 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 9.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 send_requests_blocked(&self) -> Result<ControlMessage, EndpointError>

Generate a REQUESTS_BLOCKED message indicating that this endpoint wants to create a new request but is blocked by the current MAX_REQUEST_ID.

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

Process an incoming REQUESTS_BLOCKED message from the peer. This signals that the peer wants to issue new requests but is limited by the MAX_REQUEST_ID we advertised.

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.

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_location: Option<Location>, 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.

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 5.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 share a map and 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 receive_subscribe_update( &mut self, msg: &SubscribeUpdate, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBE_UPDATE.

Section 9.10 puts the message in the subscriber’s hands, so one that arrives is about a subscription the peer opened, and is looked for among those and not among this endpoint’s own.

Errors

EndpointError::UpdateForUnknownRequest when the Request ID names no request this session has carried, with the session already moved to Closed, and the subscription flow’s own InvalidTransition when it names one that has already ended.

pub fn subscribe_update( &mut self, subscription_request_id: VarInt, start_location: Location, end_group: VarInt, subscriber_priority: u8, forward: Forward, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

Send a SUBSCRIBE_UPDATE for an active subscription. Allocates a fresh request ID for the update message and returns it alongside the message.

pub fn receive_publish_done( &mut self, msg: &PublishDone, ) -> Result<(), EndpointError>

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

Section 5.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 9.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: Section 5.1 says “A publisher MUST send exactly one SUBSCRIBE_OK or SUBSCRIBE_ERROR in response to a SUBSCRIBE.”

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.

Errors

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

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 9.11: “A Subscriber issues an UNSUBSCRIBE message to a Publisher indicating it is no longer interested in receiving the specified Track, indicating 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 5.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::NotASubscription for one that names a track status, 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, asking for a range of Objects independently of any subscription. Allocates a Request ID.

Section 9.16.1 describes the range as two Locations: “Start Location: The start Location” and “End Location: The end Location, plus 1. A Location.Object value of 0 means the entire group is requested.” The end is the caller’s to name for the same reason the start is - a fetch that cannot say where it stops is a fetch for nothing - and this call used to send group 0, object 0 for every fetch it built.

Errors

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 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 9.16.2: “A Joining Fetch is associated with a Subscribe request by specifying the Request ID of an active subscription. A publisher receiving a Joining Fetch uses properties of the associated Subscribe to determine the Track Namespace, Track Name 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 9.16.2.1 has the publisher set “the Start Location to {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 9.16.2 answers that at the publisher, “it MUST respond with a Fetch Error with code Invalid Joining Request ID”, and this endpoint is the subscriber.

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 9.16.2.1: “For an Absolute Joining Fetch, the publisher sets the Start Location to Joining Start.” 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>

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 9.16.2 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 9.16.2: “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 5.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 9.16.2 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 9.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: TrackNamespace, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

Send a SUBSCRIBE_NAMESPACE message.

Section 9.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, request_id: VarInt, _track_namespace: 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 9.28: “The subscriber sends the SUBSCRIBE_NAMESPACE control message to a publisher to request the current set of matching published namespaces 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 9.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 6.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 6.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 publish_namespace( &mut self, track_namespace: TrackNamespace, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

Send a PUBLISH_NAMESPACE message.

pub fn receive_publish_namespace_ok( &mut self, msg: &PublishNamespaceOk, ) -> Result<(), EndpointError>

Process an incoming PUBLISH_NAMESPACE_OK.

pub fn receive_publish_namespace_error( &mut self, msg: &PublishNamespaceError, ) -> Result<(), EndpointError>

Process an incoming PUBLISH_NAMESPACE_ERROR.

pub fn receive_publish_namespace( &mut self, msg: &PublishNamespace, ) -> Result<(), EndpointError>

Process an incoming PUBLISH_NAMESPACE, recording the announcement it makes.

Section 9.23: “The publisher sends the PUBLISH_NAMESPACE control message to advertise that it has tracks available within a 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_publish_namespace( &self, request_id: VarInt, ) -> Option<&PublishNamespace>

The PUBLISH_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 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_publish_namespace_count(&self) -> usize

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

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

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

Section 9.26: “The publisher sends the PUBLISH_NAMESPACE_DONE control message to indicate its intent to stop serving new subscriptions for tracks within the provided Track Namespace.” and Section 9.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 PUBLISH_NAMESPACE 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_publish_namespace_ok( &mut self, request_id: VarInt, ) -> Result<ControlMessage, EndpointError>

Build the PUBLISH_NAMESPACE_OK accepting an announcement the peer made.

Section 6.2: “A subscriber MUST send exactly one PUBLISH_NAMESPACE_OK or PUBLISH_NAMESPACE_ERROR in response to a PUBLISH_NAMESPACE. 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_publish_namespace_error( &mut self, request_id: VarInt, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<ControlMessage, EndpointError>

Build the PUBLISH_NAMESPACE_ERROR refusing an announcement the peer made.

The same sentence in Section 6.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_publish_namespace_done( &mut self, msg: &PublishNamespaceDone, ) -> Result<(), EndpointError>

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

Section 9.26: “The publisher sends the PUBLISH_NAMESPACE_DONE 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::publish_namespace_done, 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 publish_namespace_cancel( &mut self, track_namespace: TrackNamespace, error_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<ControlMessage, EndpointError>

Build the PUBLISH_NAMESPACE_CANCEL revoking an acceptance.

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

Previously responded PUBLISH_NAMESPACE_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_publish_namespace_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 publish_namespace_done( &mut self, track_namespace: TrackNamespace, ) -> Result<ControlMessage, EndpointError>

Send the PUBLISH_NAMESPACE_DONE withdrawing an announcement this endpoint made.

Section 9.26: “The publisher sends the PUBLISH_NAMESPACE_DONE control message to indicate its intent to stop serving new subscriptions for tracks within the provided Track Namespace.” This endpoint is that publisher, so the record it ends is one Self::publish_namespace opened.

Errors

EndpointError::UnknownNamespace if this endpoint has announced nothing under that namespace, and the namespace flow’s own InvalidTransition for an announcement the peer has not accepted, or has already cancelled.

pub fn receive_publish_namespace_cancel( &mut self, msg: &PublishNamespaceCancel, ) -> Result<(), EndpointError>

Process an incoming PUBLISH_NAMESPACE_CANCEL, ending an announcement this endpoint made.

Section 8.4 names what the peer is revoking: a namespace “it previously responded PUBLISH_NAMESPACE_OK to”. What it responded to is an announcement this endpoint made, so the record this reads is the outbound one.

Errors

EndpointError::UnknownNamespace if this endpoint has announced nothing under that namespace, and the namespace flow’s own InvalidTransition for an announcement the peer never accepted.

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

The identifier of an announcement this endpoint made for namespace.

The mirror of Self::inbound_publish_namespace_id, over the other map: PUBLISH_NAMESPACE_DONE and PUBLISH_NAMESPACE_CANCEL name a namespace on this draft, and the announcements this endpoint made are filed under the Request IDs it allocated.

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 the publisher about a track. Allocates a Request ID.

TRACK_STATUS is shaped like a SUBSCRIBE on this draft, filter-dependent fields included, and this call used to carry none of them: it sent priority 128, Ascending, Forward and Largest Object for every request it built. The four the message names are the caller’s, as they are on draft-13, which words this message the same way.

The filter types that need a range are refused rather than sent with an absent one, which is what Self::subscribe does with the same question one message over.

Errors

EndpointError::FilterNeedsRange for AbsoluteStart or AbsoluteRange, the session error when the session is not established, and the request-id error when there is no identifier left to spend.

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

Process an incoming TRACK_STATUS_OK.

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

Process an incoming TRACK_STATUS_ERROR.

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

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

Section 9.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 9.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 9.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 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 5.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 9.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 5.1 gives to whichever end opened it: “The initiator of the subscription sets the initial Forward State in either PUBLISH or SUBSCRIBE.” Section 9.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 9.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_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 5.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 send_publish_done( &mut self, request_id: VarInt, status_code: VarInt, reason_phrase: Vec<u8>, ) -> Result<ControlMessage, EndpointError>

Send a PUBLISH_DONE message (publisher finishing).

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

Generate the PUBLISH_ERROR that rejects a PUBLISH the peer sent, 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 5.1 says “A subscriber MUST send exactly one PUBLISH_OK or PUBLISH_ERROR in response to a PUBLISH.”

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

This used to fall through to the subscriptions this endpoint opened with SUBSCRIBE, and then to Ok(()). Neither is right. A SUBSCRIBE is refused with SUBSCRIBE_ERROR, which arrives at Self::receive_subscribe_error and is answered there; and an identifier this session opened nothing under is one the peer had no reason to name, which the answer beside this one has always said.

Errors

As Self::receive_publish_ok.

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.

Section 5.1: “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.” The record made here outlives that answer, because everything the section gives the subscription afterwards names this same request: an UNSUBSCRIBE from this endpoint, a PUBLISH_DONE from the peer, and the Track Alias the offer spends for as long as it lasts.

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 PUBLISH 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 offer from the peer under – an offer of this endpoint’s own included, whose state is in the same map but whose message was never received. The record itself lives on past the answer, because the subscription the offer opened is not over until an UNSUBSCRIBE or a PUBLISH_DONE ends it.

pub fn pending_publish_count(&self) -> usize

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

pub fn send_publish_ok( &mut self, request_id: VarInt, forward: Forward, subscriber_priority: u8, group_order: GroupOrder, filter_type: FilterType, start_location: Option<Location>, end_group: Option<VarInt>, ) -> Result<ControlMessage, EndpointError>

Generate a PUBLISH_OK accepting a PUBLISH the peer sent, 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 5.1 says “A subscriber MUST send exactly one PUBLISH_OK or PUBLISH_ERROR in response to a PUBLISH.”

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

Hold a Request ID a peer allocated to the rules Section 9.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.

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. Section 9.1 names the set, and SUBSCRIBE_UPDATE is in it - it carries a Request ID of its own, alongside the separate field naming the request it modifies.

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.