Struct Endpoint 

pub struct Endpoint {

    role: Role,
    session: SessionStateMachine,
    subscribe_ids: SubscribeIdAllocator,
    advertised_max_id: u64,
    peer_highest_subscribe_id: Option<u64>,
    subscriptions: HashMap<u64, SubscriptionStateMachine>,
    inbound_subscribes: HashMap<u64, InboundSubscribe>,
    inbound_fetches: HashMap<u64, InboundFetch>,
    forwarding_preferences: Mutex<TrackForwardingPreferences>,
    locations: Arc<Mutex<TrackLocations>>,
    track_bindings: HashMap<(SubscribeSide, u64), TrackBinding>,
    fetches: HashMap<u64, FetchStateMachine>,
    subscribe_announces: HashMap<Vec<Vec<u8>>, SubscribeAnnouncesStateMachine>,
    inbound_subscribe_announces: HashMap<Vec<Vec<u8>>, InboundSubscribeAnnounces>,
    announces: HashMap<Vec<Vec<u8>>, AnnounceStateMachine>,
    inbound_announces: HashMap<Vec<Vec<u8>>, InboundAnnounce>,
    track_statuses: HashMap<(Vec<Vec<u8>>, Vec<u8>), TrackStatusStateMachine>,
    inbound_track_statuses: HashMap<(Vec<Vec<u8>>, Vec<u8>), InboundTrackStatus>,
    negotiated_version: Option<VarInt>,
    offered_versions: Vec<VarInt>,
    goaway_uri: Option<Vec<u8>>,
    peer_reported_max_subscribe_id: Option<VarInt>,
}

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

Fields

role: Role

Which side of the session this is. Read only by the GOAWAY rule.

session: SessionStateMachine

subscribe_ids: SubscribeIdAllocator

advertised_max_id: u64

Tracks the MAX_SUBSCRIBE_ID we have advertised to the peer.

peer_highest_subscribe_id: Option<u64>

The highest Subscribe ID the peer has used, once it has used one.

subscriptions: HashMap<u64, SubscriptionStateMachine>

inbound_subscribes: HashMap<u64, InboundSubscribe>

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

Separate from subscriptions, which holds the ones this endpoint opened, because the identifiers do not separate themselves: see SubscribeSide.

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, because the identifiers do not separate themselves: both ends allocate from zero, so one number can name a fetch at each end at once.

forwarding_preferences: Mutex<TrackForwardingPreferences>

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

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

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

locations: Arc<Mutex<TrackLocations>>

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

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

track_bindings: HashMap<(SubscribeSide, u64), TrackBinding>

fetches: HashMap<u64, FetchStateMachine>

subscribe_announces: HashMap<Vec<Vec<u8>>, SubscribeAnnouncesStateMachine>

inbound_subscribe_announces: HashMap<Vec<Vec<u8>>, InboundSubscribeAnnounces>

Namespace subscriptions the peer made, keyed by the prefix each one names.

The prefix is the whole of a request’s name on this draft: the acceptance, the refusal and the withdrawal all carry a Track Namespace Prefix and nothing else, so one prefix has one record here. A second SUBSCRIBE_ANNOUNCES under a prefix already subscribed replaces it, which is a case Section 4.1 forbids rather than one this map decides.

announces: HashMap<Vec<Vec<u8>>, AnnounceStateMachine>

inbound_announces: HashMap<Vec<Vec<u8>>, InboundAnnounce>

Announcements the peer made, keyed by the namespace each names, which is all this draft’s ANNOUNCE carries to name it by.

track_statuses: HashMap<(Vec<Vec<u8>>, Vec<u8>), TrackStatusStateMachine>

inbound_track_statuses: HashMap<(Vec<Vec<u8>>, Vec<u8>), InboundTrackStatus>

Track statuses the peer asked about, keyed by the track each names.

Kept apart from track_statuses, which holds the ones this endpoint asked about: the two are answered by opposite ends. This draft’s TRACK_STATUS_REQUEST carries no identifier of its own, so the track it names is the only thing an answer can be matched to, which is the key the outbound map is under for the same reason.

negotiated_version: Option<VarInt>

offered_versions: Vec<VarInt>

goaway_uri: Option<Vec<u8>>

peer_reported_max_subscribe_id: Option<VarInt>

The most recent maximum_subscribe_id reported by the peer via a SUBSCRIBES_BLOCKED message (draft-09 only).

Implementations

impl Endpoint

pub fn new(role: Role) -> Self

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

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

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

Why the set is read rather than kept

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

Why a binding for the same track is not a conflict

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

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

The track a live binding has given alias to.

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

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

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

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

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

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

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

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

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

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

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

fn binding_is_live(&self, key: (SubscribeSide, u64)) -> bool

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

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

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

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

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

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

pub fn role(&self) -> Role

Returns which side of the session this endpoint is.

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 subscribe ID allocation.

pub fn active_subscription_count(&self) -> usize

Returns the number of active subscription state machines.

pub fn active_fetch_count(&self) -> usize

Returns the number of active fetch state machines.

pub fn active_subscribe_announces_count(&self) -> usize

Returns the number of active subscribe-announces state machines.

pub fn active_announce_count(&self) -> usize

Returns the number of active announce state machines.

pub fn active_track_status_count(&self) -> usize

Returns the number of active track status state machines.

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

Transition from Connecting to SetupExchange.

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

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

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

Generate a CLIENT_SETUP message (client-side).

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

Process a SERVER_SETUP message (client-side). Transitions to Active. If the server includes a MAX_SUBSCRIBE_ID parameter (key 0x02), the subscribe 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 receive_client_setup_and_respond_with( &mut self, client_setup: &ClientSetup, selected_version: VarInt, parameters: Vec<KeyValuePair>, ) -> Result<ControlMessage, EndpointError>

Process CLIENT_SETUP and generate SERVER_SETUP carrying parameters.

The form that can answer with a MAX_SUBSCRIBE_ID. Section 7.2.2.2 describes the parameter as communicating “an initial value for the Maximum Subscribe ID to the receiving subscriber. The default value is 0, so if not specified, the peer MUST NOT create subscriptions” - so a server that never sends it has told the client it may not subscribe, and every SUBSCRIBE the client tries is answered Blocked until a MAX_SUBSCRIBE_ID message arrives.

A MAX_SUBSCRIBE_ID among parameters is recorded as the ceiling this endpoint has advertised, which is the number a peer’s Subscribe IDs are measured against.

Errors

The setup errors, and a malformed MAX_SUBSCRIBE_ID in the CLIENT_SETUP.

fn read_granted_max( &mut self, parameters: &[KeyValuePair], ) -> Result<(), EndpointError>

Take the budget a peer’s setup parameters grant this endpoint.

An explicit 0 is the same as the parameter’s absence - Section 7.2.2.2 gives it a default of 0 - so it is not put through the only-increase rule, which belongs to the MAX_SUBSCRIBE_ID message.

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

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

The peer’s Subscribe IDs are bound by this number, and this endpoint’s own by the one the peer advertised. The two are different values and measuring against the wrong one accepts ids a conforming peer would never send and refuses ids it may.

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

Hold a Subscribe ID the peer chose to the rules Section 7.4 states about it.

“Subscribe ID is a variable length integer that MUST be unique and monotonically increasing within a session and MUST be less than the session’s Maximum Subscribe ID”, and Section 7.7 repeats the first half for FETCH - so the two share one sequence and are checked together here.

The ceiling is the one this endpoint advertised, not the one the peer granted us: those are different numbers, and either may be the larger. Strictly increasing gives uniqueness as well, so one high-water mark answers both halves of the sentence.

The ceiling is a session rule, not a request rule

Section 7.20: “If a Subscribe ID equal or larger than this is received by the publisher that sent the MAX_SUBSCRIBE_ID, the publisher MUST close the session with an error of ‘Too Many Subscribes’.” Which is also why the number measured against is the one this endpoint sent. An id that reaches the ceiling is not a SUBSCRIBE to refuse with a SUBSCRIBE_ERROR: the session is over, so this moves the endpoint’s own state to Closed and leaves the code to EndpointError::session_error_code.

Errors

SubscribeIdError::ExceedsMax if the id reaches the advertised ceiling, and EndpointError::PeerSubscribeIdNotIncreasing if it does not increase on the last one the peer used.

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

Process an incoming SUBSCRIBE, checking the Subscribe ID the peer chose.

Errors

Whatever Self::validate_peer_subscribe_id answers.

pub fn receive_max_subscribe_id( &mut self, msg: &MaxSubscribeId, ) -> Result<(), EndpointError>

Process an incoming MAX_SUBSCRIBE_ID message, ending the session if the ceiling it carries does not increase. Section 7.20: “The Maximum Subscribe Id MUST only increase within a session, and receipt of a MAX_SUBSCRIBE_ID message with an equal or smaller Subscribe ID value is a ‘Protocol Violation’.” Section 3.5 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

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

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

Generate a MAX_SUBSCRIBE_ID message (typically server-side).

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

Errors

The decrease error if the value does not strictly increase.

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

Process an incoming GOAWAY message. Transitions to Draining.

Errors

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Process an incoming SUBSCRIBE_OK.

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

Process an incoming SUBSCRIBE_ERROR.

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

Send an UNSUBSCRIBE message for an active subscription.

pub fn subscribe_update( &mut self, subscribe_id: VarInt, start_group: VarInt, start_object: VarInt, end_group: VarInt, subscriber_priority: u8, parameters: Vec<KeyValuePair>, ) -> Result<ControlMessage, EndpointError>

Send a SUBSCRIBE_UPDATE narrowing a subscription this endpoint opened.

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

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

Errors

EndpointError::UnknownSubscribe 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 7.5: “A subscriber issues a SUBSCRIBE_UPDATE to a publisher to request a change to an existing subscription.” One that arrives is therefore about a subscription the peer opened, which is why it is looked for among those and not among this endpoint’s own.

Errors

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

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

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

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

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

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

pub fn pending_subscribe_count(&self) -> usize

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

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

Build the SUBSCRIBE_OK accepting a subscription the peer opened.

Errors

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

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

Build the SUBSCRIBE_ERROR rejecting a subscription the peer opened.

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

Errors

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

Build the SUBSCRIBE_DONE ending a subscription this endpoint accepted.

Errors

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

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

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

Errors

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

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

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

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

Send a joining FETCH message that attaches to an existing subscription. Allocates a new subscribe ID for the fetch and tracks it in its own fetch state machine.

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, subscribe_id: VarInt, ) -> Result<ControlMessage, EndpointError>

Send a FETCH_CANCEL message.

pub fn on_fetch_stream_fin( &mut self, subscribe_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, subscribe_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 Subscribe ID is checked here, in the same sequence a SUBSCRIBE draws from: Section 7.7 gives it the same “unique and monotonically increasing within a session” requirement.

A Joining Fetch is recorded like any other. Section 7.7 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

Whatever Self::validate_peer_subscribe_id answers, and the fetch flow’s own InvalidTransition for a second FETCH under an identifier already carrying one.

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

The FETCH the peer sent under subscribe_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 7.7: “If a publisher receives a Joining Fetch with a Subscribe ID that does not correspond to an existing Subscribe, it MUST respond with a Fetch Error.”

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, subscribe_id: VarInt, group_order: GroupOrder, end_of_track: u8, largest_group_id: VarInt, largest_object_id: VarInt, parameters: Vec<KeyValuePair>, ) -> Result<ControlMessage, EndpointError>

Build the FETCH_OK accepting a fetch the peer opened.

Errors

EndpointError::UnknownSubscribe 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.3 says the publisher “MUST send exactly one FETCH_OK or FETCH_ERROR in response to a FETCH”.

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

Build the FETCH_ERROR refusing a fetch the peer opened.

Errors

EndpointError::UnknownSubscribe 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 7.8: 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::UnknownSubscribe 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, subscribe_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::UnknownSubscribe if the peer opened no fetch under that identifier, and the fetch flow’s own InvalidTransition from a state the stream cannot close from.

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

Send a SUBSCRIBE_ANNOUNCES message.

Section 7.13: “A subscriber cannot make overlapping namespace subscriptions on a single session.”

Errors

The session error when the session is not established, and EndpointError::OwnPrefixOverlap when the prefix overlaps one this endpoint has already subscribed to.

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

Process an incoming SUBSCRIBE_ANNOUNCES_OK.

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

Process an incoming SUBSCRIBE_ANNOUNCES_ERROR.

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

Send an UNSUBSCRIBE_ANNOUNCES message.

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

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

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

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

Errors

The session error when the session is not established, and EndpointError::PeerPrefixOverlap when the prefix overlaps one the peer has already subscribed to.

fn peer_prefix_overlap(&self, prefix: &TrackNamespace) -> bool

Whether prefix overlaps a namespace subscription the peer has already made on this session.

Every one of them counts, including a subscription the peer has since withdrawn: the sentence weighs the arriving prefix against “an earlier SUBSCRIBE_ANNOUNCES”, and one that has ended was still earlier.

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.

pub fn pending_subscribe_announces( &self, prefix: &TrackNamespace, ) -> Option<&SubscribeAnnounces>

The SUBSCRIBE_ANNOUNCES the peer sent for prefix and this endpoint has not answered yet.

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

pub fn pending_subscribe_announces_count(&self) -> usize

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

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

Build the SUBSCRIBE_ANNOUNCES_OK accepting a namespace subscription the peer made.

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

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

Errors

EndpointError::UnknownPeerNamespaceSubscription if the peer has subscribed to nothing under that prefix, and the namespace flow’s own InvalidTransition for a request already answered.

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

Build the SUBSCRIBE_ANNOUNCES_ERROR refusing a namespace subscription the peer made.

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

Errors

EndpointError::UnknownPeerNamespaceSubscription if the peer has subscribed to nothing under that prefix, and the namespace flow’s own InvalidTransition for a request already answered.

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

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

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

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

Errors

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

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

Send an ANNOUNCE 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 7.22: “The publisher sends the ANNOUNCE control message to advertise where the receiver can route SUBSCRIBEs for tracks 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. This draft’s ANNOUNCE carries no Request ID, so the namespace it names is what the record is filed under, and a second one naming a namespace already held replaces it. No sentence makes a repeat an error, and the newest advertisement is the one an answer has to be built from.

Errors

The session error when the session is not established.

pub fn pending_announce( &self, track_namespace: &TrackNamespace, ) -> Option<&Announce>

The ANNOUNCE the peer sent for track_namespace and this endpoint has not answered yet.

None once it has been answered, and for a namespace 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.

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

Build the ANNOUNCE_OK accepting an announcement the peer made.

Section 4.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::UnknownPeerNamespace if the peer has announced nothing under that namespace, and the namespace flow’s own InvalidTransition for an announcement already answered.

pub fn send_announce_error( &mut self, track_namespace: TrackNamespace, 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 4.2 answers both ways: one message back and no second one, whichever of the two it is.

Errors

EndpointError::UnknownPeerNamespace if the peer has announced nothing under that namespace, 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 7.23: “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 6.2 names what a cancellation revokes: a namespace “it previously responded ANNOUNCE_OK to”. Section 7.11 says what it does: the subscriber “will stop sending new subscriptions for tracks within the provided Track Namespace”.

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

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

Errors

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

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

Send a TRACK_STATUS_REQUEST message.

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

Process an incoming TRACK_STATUS reply.

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

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

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

Answering is the application’s to do, and it needs both the request and somewhere to answer from. This draft’s request carries no Request ID, so the track it names is what the record is filed under, and a second request for a track already asked about replaces it. No sentence makes a repeat an error, and the newest request is the one an answer has to be built from.

Errors

The session error when the session is not established.

pub fn pending_track_status_request( &self, track_namespace: &TrackNamespace, track_name: &[u8], ) -> Option<&TrackStatusRequest>

The TRACK_STATUS_REQUEST the peer sent about this track and this endpoint has not answered yet.

None once it has been answered, and for a track the peer has asked nothing about.

pub fn pending_track_status_request_count(&self) -> usize

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

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

Build the TRACK_STATUS answering a request the peer sent.

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

Section 7.24 says which track the answer is about, and this draft has no identifier to say it with, so the caller names the track and the message repeats it.

Errors

EndpointError::UnknownPeerTrackStatus if the peer has asked nothing about that track, and the flow’s own InvalidTransition for a request already answered.

pub fn receive_subscribes_blocked( &mut self, msg: &SubscribesBlocked, ) -> Result<(), EndpointError>

Process an incoming SUBSCRIBES_BLOCKED.

Draft-09 adds this message so the peer can explicitly report that a new subscribe id would exceed our advertised maximum. The endpoint records the peer’s reported maximum; acting on it (issuing a new MAX_SUBSCRIBE_ID) is up to the caller.

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

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

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

Dispatch an incoming control message to the appropriate handler.

Trait Implementations

impl Default for Endpoint

fn default() -> Self

Returns the “default value” for a type.