Struct Endpoint 

pub struct Endpoint {

    role: Role,
    session: SessionStateMachine,
    request_ids: RequestIdAllocator,
    subscriptions: HashMap<u64, SubscriptionStateMachine>,
    fetches: HashMap<u64, FetchStateMachine>,
    unjoinable_fetches: HashMap<u64, u64>,
    subscribe_namespaces: HashMap<u64, SubscribeNamespaceStateMachine>,
    subscribe_tracks: HashMap<u64, SubscribeNamespaceStateMachine>,
    publish_namespaces: HashMap<u64, PublishNamespaceStateMachine>,
    track_statuses: HashMap<u64, TrackStatusStateMachine>,
    publishes: HashMap<u64, PublishStateMachine>,
    goaway_uri: Option<Vec<u8>>,
    goaway_request_streams: HashSet<u64>,
    peer_request_ids: HashSet<u64>,
    inbound_requests: HashMap<u64, ControlMessage>,
    overlapping_namespace_subscriptions: HashMap<u64, u64>,
    updated_namespace_prefixes: HashMap<u64, TrackNamespace>,
    overlapping_prefix_updates: HashMap<u64, u64>,
    unanswered_peer_updates: HashMap<u64, u64>,
    owed_update_failures: HashSet<u64>,
    answered_peer_requests: HashSet<u64>,
    track_bindings: HashMap<u64, TrackBinding>,
    fetch_tracks: HashMap<u64, FetchTrack>,
    malformed: Mutex<MalformedTracks>,
    locations: Arc<Mutex<TrackLocations>>,
}

Fields

role: Role

session: SessionStateMachine

request_ids: RequestIdAllocator

subscriptions: HashMap<u64, SubscriptionStateMachine>

fetches: HashMap<u64, FetchStateMachine>

unjoinable_fetches: HashMap<u64, u64>

Every Joining Fetch the peer sent naming a subscription this session had none live for, and the identifier each one named.

Judged as the FETCH arrives, because that is the moment the rule about it names, and kept for as long as the fetch is: a subscription that ends between the FETCH and its answer does not turn a fetch that could be joined into one that could not.

subscribe_namespaces: HashMap<u64, SubscribeNamespaceStateMachine>

subscribe_tracks: HashMap<u64, SubscribeNamespaceStateMachine>

publish_namespaces: HashMap<u64, PublishNamespaceStateMachine>

track_statuses: HashMap<u64, TrackStatusStateMachine>

publishes: HashMap<u64, PublishStateMachine>

goaway_uri: Option<Vec<u8>>

goaway_request_streams: HashSet<u64>

The peer’s requests whose stream has already carried a GOAWAY.

Section 10.4 makes the second GOAWAY on one request stream a session close while leaving a first one on every other stream legal, so the count cannot live on the session. Held apart from the per-kind request maps because a GOAWAY says nothing about which kind of request it migrates.

peer_request_ids: HashSet<u64>

Every Request ID the peer has spent, whether or not the request it opened is still live.

Draft-18 Section 10.1 makes a duplicate Request ID a session close, and “duplicate” is about the id ever having been used, not about the request still being open. Deriving it from the per-kind maps instead would answer wrongly the moment those maps are ever pruned, so the rule is stated once, here, and this set is never pruned.

inbound_requests: HashMap<u64, ControlMessage>

Every request the peer opened a stream with, as it arrived, keyed by the Request ID it carries.

One map for all seven kinds, because one entry point takes all seven: receive_request_on_stream is the only thing that writes here, so what is in here is a request, and it is the peer’s, by construction. How far each one has got stays in the per-kind map beside this endpoint’s own requests, which is where every later message on the stream reads it and where the parity of a Request ID keeps the two ends apart.

What a map of state machines cannot hold is the request. Section 5.1 has the subscriber “either accepts or rejects the subscription”, and what is being accepted or rejected – the track, the namespace prefix, the parameters – is named in the request and nowhere else once the message has been dropped.

overlapping_namespace_subscriptions: HashMap<u64, u64>

The request each namespace subscription of the peer’s overlapped when it arrived, keyed by the Request ID of the one that arrived.

The prefix is judged where the sentence says it is judged, on receipt, and the verdict is read again when the answer is written. An entry means this endpoint owes that request a REQUEST_ERROR and may send it nothing else.

updated_namespace_prefixes: HashMap<u64, TrackNamespace>

The Track Namespace Prefix a REQUEST_UPDATE asks one of the peer’s namespace subscriptions to move to, held until that update is answered.

Section 10.9.2 ties the move to the acceptance: “If the update is accepted, NAMESPACE and NAMESPACE_DONE messages following the REQUEST_OK will contain Track Namespace suffixes relative to the updated prefix.” Until then the subscription still selects what the peer opened it with, so a later request is weighed against the prefix in inbound_requests and not against this one.

overlapping_prefix_updates: HashMap<u64, u64>

The subscription an unanswered prefix update would collide with, when it would, keyed by the Request ID of the one being moved.

Separate from overlapping_namespace_subscriptions because one Request ID can be carrying both verdicts at once: the request that opened the stream has one and an update on it has another, and they are settled by different messages on that same stream.

unanswered_peer_updates: HashMap<u64, u64>

Per request stream, how many REQUEST_UPDATEs are still waiting for the answer Section 10.9 requires.

owed_update_failures: HashSet<u64>

The requests whose refused update has not been followed by the PUBLISH_DONE that ends them.

Emptied as each is written. A request is in here for exactly as long as this endpoint owes the peer the second half of a refusal.

answered_peer_requests: HashSet<u64>

The peer’s requests whose own response has already been written.

Section 10.9 gives an update the same two answers a request has, and on the kinds REQUEST_OK answers, the message that answers the request and the message that answers an update are the same message. Nothing on the wire tells them apart, so both endpoints resolve it by order: the first response on a stream answers the request that opened it and the ones after it answer updates. This set is that order, recorded.

track_bindings: HashMap<u64, TrackBinding>

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

Section 11.1 forbids 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 request’s own state machine, rather than keeping a second copy of it that every path ending a subscription would have to remember to prune.

fetch_tracks: HashMap<u64, FetchTrack>

The track each fetch this endpoint made is for.

Not in track_bindings, because 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. A Joining Fetch names no track either and takes the joined subscription’s, resolved as the fetch is made rather than at the withdrawal - one fills a buffer behind the live edge and outlives the subscription it joined, so a lookup through the join would come up empty exactly while there was still a fetch to cancel.

malformed: Mutex<MalformedTracks>

Which tracks this endpoint has given up on, and what for.

Behind a lock because the note is taken on the data plane, where this endpoint is reached through &self.

locations: Arc<Mutex<TrackLocations>>

How far each track’s objects have reached, and where a track ended.

An Arc because a subgroup stream measures its objects against one track for as long as it runs, and the handle it holds outlives any single call into this endpoint.

Implementations

impl Endpoint

pub fn new(role: Role) -> Self

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 Section 11.1 requires when alias already names a different track that still has an Established subscription, or None when it is free.

Why the set is read rather than kept

“Established” is a subscription state Section 5.1 defines, and both state machines here already hold it: a subscription reaches it on SUBSCRIBE_OK and a publish on PUBLISH_OK, and each leaves it on the message that ends the flow. Asking them 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 not: a second PUBLISH under a Request ID already bound is refused by the duplicate-Request-ID rule before it reaches here, and comparing a request against its own binding would answer the wrong rule if that one ever moved.

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 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 Established subscription” - and the prohibition on the publisher is not: “The same Track Alias MUST NOT be used by a publisher to refer to two different Tracks simultaneously in the same session.” 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 an Established subscription.

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

The track a live binding has given alias to.

Read rather than kept: a binding whose request has ended holds nothing, and an alias that is free again may name a different track next.

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

Whether this endpoint receives the track a request names.

The sentence to be answered is a subscriber’s - “cancel any corresponding subscription or fetches for that Track from that publisher” - so a request that makes this endpoint the publisher is not one of them. A SUBSCRIBE in the table is always this endpoint’s own, because a SUBSCRIBE the peer sends makes this endpoint the publisher and leaves no binding. A PUBLISH is in the table either way round, and Request ID parity is what separates them: the offer the peer made is the one this endpoint receives a track through.

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.4.2’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 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.

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. The record is keyed on the track, and so is this.

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

Note a track as malformed and name every request through which this endpoint is receiving it, so the caller can cancel them.

This is the whole of what this crate can do here, and the reason is structural rather than a shortfall. Section 2.4.2 asks a subscriber that detects a Malformed Track to “cancel any corresponding subscription or fetches for that Track from that publisher”, and on this draft cancelling a request means resetting the request’s own bidirectional stream - Section 3.3.2. Every request lives at the front of a stream of its own, and Connection::recv_on_request_stream hands that stream to the caller as a RequestStream. So the stream this answer operates on is not the endpoint’s to touch, and no amount of state here changes that. What the endpoint can do is say which requests they are.

Nothing here ends a request

Deliberately, and it is what keeps the answer to one. The flows are read and not moved, so a caller that passes each id to Connection::cancel_request_stream ends the request there - and the binding stops being in use, so a second object past the end finds no track for the alias and names nothing. The request ending is still what closes the loop, exactly as it is on the drafts that answer with a message; what changed is which side ends it. A caller that ignores the list gets named the same requests again, which is honest: the condition really did fire again.

What is named, and what is not

Requests through which this endpoint receives the track: a SUBSCRIBE this endpoint sent, a PUBLISH the peer sent, and a fetch this endpoint made. A request that makes this endpoint the publisher is not one of them. Empty for an alias no live binding names, and empty for a track whose only requests are ones this endpoint publishes.

Sorted, because a HashMap iterates in no order and two requests for one track is a shape a peer can produce.

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

pub fn peer_role(&self) -> Role

Returns the role of the peer, which is the other one.

fn require_update_failure_status( &mut self, id: u64, status_code: VarInt, ) -> Result<(), EndpointError>

Hold a refused update’s ending to the status the draft names, and retire the obligation once that ending is written.

Both routes to a PUBLISH_DONE come through here. A subscription this endpoint accepted is ended by a response on the peer’s stream; a PUBLISH this endpoint sent is ended on its own, which does not go through the response path at all. The rule is the same either way, so it is stated once and called twice rather than written where each route happened to need it.

pub fn has_unanswered_update(&self, request_id: VarInt) -> bool

Whether the peer has sent a REQUEST_UPDATE on this request that has not been answered yet.

Draft-18 Section 10.9: “A subscriber can also send REQUEST_UPDATE to modify parameters of a subscription established with PUBLISH”, and the receiver of one “MUST respond with exactly one REQUEST_OK or REQUEST_ERROR message indicating if the update was successful”.

Asked by the connection layer, which otherwise writes nothing on a stream this endpoint opened. An update the peer sent on a PUBLISH is the one thing on such a stream that this endpoint has to answer, and this is what tells it apart from a response to its own request.

fn publishes_a_subscription(&self, id: u64) -> bool

Whether the request id names a subscription this endpoint publishes.

Draft-18 Section 10.9.1 gives a refused REQUEST_UPDATE three different consequences and picks between them by what was being updated: “When a REQUEST_UPDATE is unsuccessful, the publisher MUST also terminate the subscription by sending a PUBLISH_DONE with error code UPDATE_FAILED. When a REQUEST_UPDATE fails for a FETCH, the publisher MUST reset the FETCH data stream. When a REQUEST_UPDATE fails for a SUBSCRIBE_NAMESPACE or PUBLISH_NAMESPACE, the responder MUST close the bidi stream.”

Only the first of the three is a message, and only the first is owed here. Recording it for the other two is what would hold their streams open past the close the third sentence requires.

Two requests leave this endpoint publishing: a SUBSCRIBE the peer sent, and a PUBLISH this endpoint sent. Either can carry an update from the other side, and either is ended by a PUBLISH_DONE written from here. A peer’s PUBLISH is neither, because the peer is the publisher on it, and it is told apart by having arrived rather than been sent.

pub fn owes_update_failure(&self, request_id: VarInt) -> bool

Whether this request’s refused update still owes the peer the PUBLISH_DONE that ends it.

Asked by the connection layer, which owns the stream that termination has to be written on and therefore has to know not to close it. A REQUEST_ERROR answering the request itself ends the exchange and takes the stream with it; one answering an update does not, and nothing in the message tells the two apart.

pub fn session_state(&self) -> SessionState

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

pub fn active_subscription_count(&self) -> usize

pub fn active_fetch_count(&self) -> usize

pub fn active_subscribe_namespace_count(&self) -> usize

pub fn active_subscribe_tracks_count(&self) -> usize

pub fn active_publish_namespace_count(&self) -> usize

pub fn active_track_status_count(&self) -> usize

pub fn active_publish_count(&self) -> usize

pub fn peer_request_count(&self) -> usize

How many Request IDs the peer has spent on this session.

Nothing here removes an entry, so this only grows. A responder that wants a ceiling on peer-created state has to impose one itself — see the note on receive_request_on_stream.

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

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

pub fn send_setup( &mut self, options: Vec<KeyValuePair>, ) -> Result<ControlMessage, EndpointError>

Generate a SETUP message. Both client and server use the same message type; only the role (and the order of send/receive) distinguishes them.

pub fn receive_setup(&mut self, msg: &Setup) -> Result<(), EndpointError>

Process an incoming SETUP message. Transitions the session to Active.

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

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

Process a GOAWAY that arrived on one request stream rather than on the control stream.

Draft-18 Table 5 gives GOAWAY the Stream value “Control, Request”, and Section 10.4 says why: “A GOAWAY MAY also be sent on a request stream to initiate migration of that individual request. Upon receiving a GOAWAY on a request stream, the endpoint SHOULD re-issue that specific request on a session at the specified URI (or the current session if no URI is provided), and close the old request stream using the appropriate mechanism (e.g. FIN, stream reset, or PUBLISH_DONE).”

The session keeps running — only this one request is being moved — so the session state machine is not touched and nothing drains. Routing this form to receive_goaway instead would take the whole session to Draining over something the draft scopes to a single request.

The server-side URI rule of the same section still applies here: a server may not be handed a migration URI on a request stream either.

Re-issuing the request and closing the old stream are the caller’s, since only the caller knows what the request was; this reports that the GOAWAY named a request that exists.

Errors

EndpointError::RepeatedGoAwayOnRequestStream if this request’s stream has already carried one. 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 fail_session(&mut self, err: EndpointError) -> EndpointError

Record that the session is over because the peer broke a rule the 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. The close on the wire is the connection layer’s job — see EndpointError::session_error_code for the code it should use.

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

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

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

Process an incoming SUBSCRIBE_OK. Draft-18: no request_id on wire; the caller supplies the request_id of the bidi stream on which the response arrived.

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

Process a REQUEST_UPDATE that arrived on the bidi request stream identified by request_id.

The stream names the request being modified, not the message. Draft-18 Section 10.9: “The sender of a request (SUBSCRIBE, PUBLISH, FETCH, PUBLISH_NAMESPACE, SUBSCRIBE_NAMESPACE, SUBSCRIBE_TRACKS) can later send a REQUEST_UPDATE on the same bidi stream as the request to modify it.” Section 10.1 lists REQUEST_UPDATE among the messages that consume a Request ID of their own, so the id in the message body is the update’s, not the target’s — reading it as the target means every conforming peer’s update names a request that by construction does not exist.

Every request kind in that list is accepted, not only subscriptions: the draft names six senders and only one of them is a SUBSCRIBE. SubscriptionStateMachine is the only one of the machines with an update edge, so a subscription walks it and the rest are acknowledged without a state move.

One parameter is read rather than acknowledged. Section 10.2.14 puts TRACK_NAMESPACE_PREFIX on an update for a namespace subscription, and it is the only thing an update on any draft can change about a request that this endpoint has to judge before answering.

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

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

parameters are the request’s own, as they are on every other request this endpoint makes. FETCH carried an empty list on all five of these drafts while SUBSCRIBE, TRACK_STATUS, PUBLISH_NAMESPACE and PUBLISH took the caller’s, which made it the only one an application could not attach an authorization token to.

pub fn joining_fetch( &mut self, joining_request_id: VarInt, joining_start: VarInt, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

Send a Relative Joining Fetch (Fetch Type 0x2). Allocates a request ID.

joining_start is an offset rather than a group number: draft-18 Section 10.12.2.1 has the publisher set the Start Location to “{Joining Location.Group - Joining Start, 0}”. To name the starting group outright, use absolute_joining_fetch.

parameters are the request’s own, as they are on every other request this endpoint makes.

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

Send an Absolute Joining Fetch (Fetch Type 0x3). Allocates a request ID.

Draft-18 Section 10.12.2.1: “For an Absolute Joining Fetch, the publisher sets the Start Location to {Joining Start, 0}.” So joining_start is the group to begin at rather than a count back from one, and Section 10.12.2 leaves the choice with the subscriber: “The subscriber can set the Start Location to an absolute Location or a Location relative to the Largest group.” Only the relative form could be sent before, so an application that knew which group it wanted had to express it as an offset from a largest group it may never have been told.

fn joining_fetch_of_type( &mut self, fetch_type: FetchType, joining_request_id: VarInt, joining_start: VarInt, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

The two above, which differ in the Fetch Type and in what the publisher then reads Joining Start as. Nothing else about the message changes.

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

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

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

pub fn subscribe_namespace( &mut self, namespace_prefix: TrackNamespace, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

pub fn subscribe_tracks( &mut self, namespace_prefix: TrackNamespace, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

pub fn publish_namespace( &mut self, track_namespace: TrackNamespace, parameters: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

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

pub fn publish( &mut self, track_namespace: TrackNamespace, track_name: Vec<u8>, track_alias: VarInt, parameters: Vec<KeyValuePair>, track_properties: Vec<KeyValuePair>, ) -> Result<(VarInt, ControlMessage), EndpointError>

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

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

Process an incoming REQUEST_OK on the bidi stream identified by request_id. Draft-18: PUBLISH_OK is now a REQUEST_OK alias, so this handler also resolves outstanding PUBLISH requests.

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

Process an incoming REQUEST_ERROR on the bidi stream identified by request_id.

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

Record that a request was cancelled at its stream.

This draft withdraws a request by terminating the bidirectional stream it was made on rather than by sending a message. Section 3.3.2: “Once a request stream has been opened, the request MAY be cancelled by either endpoint. Senders cancel requests if the response is no longer of interest; Receivers cancel requests if they are unable to or choose not to respond.”

Both of those reach here. A request the peer opened and one this endpoint opened share a map and cannot collide, because their Request IDs have opposite least significant bits, so one method records a cancel from whichever side performed it.

The request moves to its end state, which is what makes the record worth keeping: a response arriving afterwards is refused rather than applied to a request that is over. The Request ID is not released — nothing here reuses one — and the per-request bookkeeping keyed by it is left alone, since a cancelled request’s stream carries nothing more.

Nothing is written on the wire. The reset that goes with this is Connection::cancel_request_stream, which calls this first and terminates the stream only if it returns Ok.

Errors

EndpointError::UnknownRequest when no request of any kind carries that id, and the kind’s own InvalidTransition when the request has not been written yet.

pub fn receive_namespace( &mut self, _msg: &Namespace, ) -> Result<(), EndpointError>

pub fn receive_namespace_done( &mut self, _msg: &NamespaceDone, ) -> Result<(), EndpointError>

pub fn receive_publish_blocked( &mut self, _msg: &PublishBlocked, ) -> Result<(), EndpointError>

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

Dispatch a message that arrived on the control stream.

Draft-18 Table 5 gives every message a Stream value, and only two of them name the control stream: SETUP is “Control”, GOAWAY is “Control, Request”. Everything else is “Request”, so this method’s job is to take those two and refuse the messages that identify a request they have no stream to name.

Four are refused for that reason. REQUEST_UPDATE modifies the request its stream carries (Section 10.9). NAMESPACE and NAMESPACE_DONE report namespaces on the SUBSCRIBE_NAMESPACE request stream that asked for them (Sections 10.16 and 10.17), and PUBLISH_BLOCKED names a track that cannot be published on the SUBSCRIBE_TRACKS stream that asked for it (Section 10.20). All four route through receive_response_on_stream, which has the request ID they need.

fn require_the_first_response_first( &mut self, id: u64, msg: &ControlMessage, ) -> Result<(), EndpointError>

Hold the response half of a namespace-scoped request to the rule that its first message answers the request.

Sections 10.18 and 10.19 state it once each, for SUBSCRIBE_NAMESPACE and for SUBSCRIBE_TRACKS: “The publisher will respond with REQUEST_OK or REQUEST_ERROR on the response half of the stream. If the subscriber receives any message other than a REQUEST_OK or a REQUEST_ERROR as the first message on the response half of the stream, then it MUST close the session with a PROTOCOL_VIOLATION.” Draft-18’s own change log records it as new work rather than as a clarification, so drafts 17 and earlier are deliberately not held to it: they say nothing about which message comes first, and refusing one there would close a session over traffic those drafts permit.

Why only these two requests

They are the two whose response half carries more than an answer. SUBSCRIBE_NAMESPACE goes on to carry NAMESPACE and NAMESPACE_DONE and SUBSCRIBE_TRACKS goes on to carry PUBLISH_BLOCKED, and those are exactly the messages that could arrive before the answer and be taken for it. A SUBSCRIBE has SUBSCRIBE_OK as its own first message and no such ambiguity, which is why the rule is written where it is.

What “first” is read from

The request’s own state machine. Pending means the request went out and nothing has come back, so it is the same question asked of the state rather than of a second counter that could disagree with it. A request this endpoint did not open, or one already answered, is not this rule’s subject and passes through.

Errors

EndpointError::ResponseBeforeTheFirstResponse, which answers Some(ProtocolViolation), and the session is failed before it returns.

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

Dispatch a message that arrived on the bidi request stream identified by request_id.

Beyond the five responses this also takes the messages draft-18 Table 5 places on a request stream without their being answers to it.

NAMESPACE (0x8) and NAMESPACE_DONE (0xE) arrive on the SUBSCRIBE_NAMESPACE request stream that asked for the namespaces they report — Section 10.16 says NAMESPACE “is sent on the response stream of a SUBSCRIBE_NAMESPACE request” — and PUBLISH_BLOCKED (0xF) on the SUBSCRIBE_TRACKS stream that asked for the track it says cannot be published: “All PUBLISH_BLOCKED messages are in response to a SUBSCRIBE_TRACKS” (Section 10.20). Table 5 marks all three “Request”, so this is where they land; the control stream refuses them.

Without these three arms a peer that follows the draft renders SUBSCRIBE_NAMESPACE useless: every namespace it reports would fall through to EndpointError::ResponseOnControlStream, on the one stream the draft says to report it on.

pub fn refuse_non_request(&mut self, ty: MessageType) -> EndpointError

Refuse a bidirectional stream the peer opened with a message type that does not begin a request, and end the session.

Draft-18 Section 3.3: “Bidirectional streams MUST NOT begin with any other message type unless negotiated. If they do, the peer MUST close the Session with a PROTOCOL_VIOLATION.” The returned error answers Some(ProtocolViolation) from EndpointError::session_error_code, which is what tells the connection layer to put the close on the wire.

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

Register a request the peer opened a bidirectional stream with, and hand back the Request ID it carries.

The mirror of receive_response_on_stream: that one is fed what comes back on a stream this endpoint opened, this one is fed the first message on a stream the peer opened.

What it enforces, and with which code

Draft-18 Section 10.1: “If an endpoint receives a Request ID where the least significant bit is incorrect for the sender, or a duplicate Request ID, it MUST close the session with INVALID_REQUEST_ID.” Both halves are checked here and both are returned as errors that answer Some(InvalidRequestId) from EndpointError::session_error_code. A message that opens no request stream at all is a different rule with a different code — see refuse_non_request.

Seven kinds, not six

Draft-18 added SUBSCRIBE_TRACKS (0x51) to the set Section 3.3 lets a bidirectional stream begin with, and renumbered SUBSCRIBE_NAMESPACE from 0x11 to 0x50. Both share the namespace-subscription lifecycle, so they share a state machine type while keeping separate maps: Section 10.19 gives them “independent overlap spaces”, so a prefix registered by one is not registered by the other.

Why there are no separate inbound maps

The peer’s ids and this endpoint’s ids have opposite least significant bits, so they cannot collide. A peer’s request goes into the same HashMap its outbound twin would, keyed the same way, and only the transition names differ — on_subscribe_received where the outbound path calls on_subscribe_sent.

Peer-controlled growth

Every accepted request adds an entry that nothing removes, and the peer chooses how many to open. peer_request_count is what a responder can watch to impose its own ceiling; this method imposes none.

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, for an identifier this session has carried no SUBSCRIBE under, and for one this endpoint spent on a request of its own – whose state is in the same map, but which never arrived here. The record itself lives on past the answer, because the subscription it opened runs on after it, for as long as the request stream does.

pub fn pending_subscribe_count(&self) -> usize

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

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, for an identifier this session has carried no PUBLISH under, and for one this endpoint spent on a request of its own – whose state is in the same map, but which never arrived here. The record itself lives on past the answer, because the subscription the offer opened runs on after it, for as long as the request stream does.

pub fn pending_publish_count(&self) -> usize

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

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, for an identifier this session has carried no FETCH under, and for one this endpoint spent on a request of its own – whose state is in the same map, but which never arrived here. The record itself lives on past the answer, because the fetch is not over until its response stream is.

pub fn pending_fetch_count(&self) -> usize

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

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, for an identifier this session has carried no PUBLISH_NAMESPACE under, and for one this endpoint spent on a request of its own – whose state is in the same map, but which never arrived here. The record itself lives on past the answer, because an announcement that was accepted stands until it is withdrawn.

pub fn pending_publish_namespace_count(&self) -> usize

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

fn peer_namespace_overlap( &self, prefix: &TrackNamespace, except: Option<u64>, ) -> 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 “established” one, 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 established yet, but this endpoint is the one about to establish it, and accepting both would leave the session holding exactly the pair the sentence exists to prevent.

The record of what the peer sent is what tells the two directions apart. The state machines live in one map per kind whichever end opened the request, so a prefix this endpoint asked about would be indistinguishable there; only requests that arrived are written into inbound_requests.

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.

except is the subscription a REQUEST_UPDATE is moving, which Section 10.2.14 weighs against “another active subscription of the same type” and therefore not against the prefix it is leaving behind.

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

The earliest track 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 “established” one, 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 established yet, but this endpoint is the one about to establish it, and accepting both would leave the session holding exactly the pair the sentence exists to prevent.

The record of what the peer sent is what tells the two directions apart. The state machines live in one map per kind whichever end opened the request, so a prefix this endpoint asked about would be indistinguishable there; only requests that arrived are written into inbound_requests.

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.

except is the subscription a REQUEST_UPDATE is moving, which Section 10.2.14 weighs against “another active subscription of the same type” and therefore not against the prefix it is leaving behind.

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, for an identifier this session has carried no SUBSCRIBE_NAMESPACE under, and for one this endpoint spent on a request of its own – whose state is in the same map, but which never arrived here. The record itself lives on past the answer, because a namespace subscription lasts as long as its stream does.

pub fn pending_subscribe_namespace_count(&self) -> usize

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

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

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

None once it has been answered, for an identifier this session has carried no SUBSCRIBE_TRACKS under, and for one this endpoint spent on a request of its own – whose state is in the same map, but which never arrived here. The record itself lives on past the answer, because the subscription it opened lasts as long as its stream does.

pub fn pending_subscribe_tracks_count(&self) -> usize

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

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, for an identifier this session has carried no TRACK_STATUS under, and for one this endpoint spent on a request of its own – whose state is in the same map, but which never arrived here. The record itself lives on past the answer, because an update may still name it once it has been answered.

pub fn pending_track_status_count(&self) -> usize

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

fn answers_an_update(&self, id: u64, msg: &ControlMessage) -> bool

Drive the state machine for a message this endpoint is about to write on a request stream the peer opened.

The mirror of receive_response_on_stream, and the reason the transitions are named *_sent rather than reusing the received-side ones: the state edges coincide, so a mis-dispatch would otherwise succeed silently instead of naming the wrong event in an InvalidTransition.

More than the five responses

Draft-18’s Table 5 marks NAMESPACE (0x8), NAMESPACE_DONE (0xE) and PUBLISH_BLOCKED (0xF) “Request”, and each is written by the responder on the stream of the request that asked for it: Section 10.16 says NAMESPACE “is sent on the response stream of a SUBSCRIBE_NAMESPACE request”, Section 10.17 says the same of NAMESPACE_DONE, and Section 10.20 says “All PUBLISH_BLOCKED messages are in response to a SUBSCRIBE_TRACKS”. None of the three ends its request, so each is a self-transition on an Active namespace subscription; sending one before the REQUEST_OK that made it Active is refused.

Draft-18 folded PUBLISH_OK into REQUEST_OK (Section 10.5), so REQUEST_OK is also what answers a peer’s PUBLISH here, where draft-17 had a message of its own for it.

The caller writes msg only after this returns Ok. What it cannot undo is the opposite order: a write that fails afterwards leaves the state machine one step ahead of the wire, the same asymmetry the outbound request path already carries. Whether this response answers a REQUEST_UPDATE rather than the request that opened the stream.

Section 10.9 gives an update the same two answers a request has: “The receiver of a REQUEST_UPDATE MUST respond with exactly one REQUEST_OK or REQUEST_ERROR message indicating if the update was successful.” Nothing in either message says which of the two it is answering, so the question is settled twice over.

A SUBSCRIBE is answered with SUBSCRIBE_OK and a FETCH with FETCH_OK, so a REQUEST_OK on one of those streams has no other message it could be answering. Draft-17 has a third, PUBLISH, which this draft folded into REQUEST_OK.

Everywhere else it is ordering: the first REQUEST_OK or REQUEST_ERROR on a stream answers the request that opened it, and the ones after it answer updates. Both endpoints have to resolve it the same way and neither has anything else to resolve it with.

fn answer_an_update( &mut self, id: u64, msg: &ControlMessage, ) -> Result<(), EndpointError>

Record a response as the answer to an outstanding update.

A REQUEST_OK answers exactly one: “The receiver MUST still send a REQUEST_OK for each successful update”. A REQUEST_ERROR may answer every update still waiting, because Section 10.9.1 permits the receiver to coalesce them — “If the coalesced REQUEST_UPDATE results in REQUEST_ERROR, only a single REQUEST_ERROR will be sent and the sender of the REQUEST_UPDATEs will not always be able to determine which caused an error.” Draft-17 has no such paragraph and answers one update per message in both directions.

No state machine moves. An update changes a request’s parameters and not its lifecycle, so a subscription that was Active before its update was answered is Active after it, whichever answer went out.

Errors

EndpointError::NoUpdateToAnswer, and nothing is written.

EndpointError::PeerPrefixOverlap and EndpointError::WrongOverlapRefusal when the update asked to move a namespace subscription onto a prefix that overlaps another of its kind and the answer is not the REQUEST_ERROR the sentence names. The same two the request itself is refused with, because it is the same rule under the same code; which of the two moments raised it is told apart by which call returned.

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 10.12.2: “If a publisher receives a Joining Fetch with a Request ID that does not correspond to a subscription in the same session in the Established or Pending (subscriber) states, it MUST return a REQUEST_ERROR with error code INVALID_JOINING_REQUEST_ID.” A standalone fetch names none and answers None, and so does a joining one whose subscription is live. Either message can establish the one it joins: Section 5.1 says the Largest Location a Joining FETCH works from is the one saved “communicated in SUBSCRIBE_OK, PUBLISH or REQUEST_UPDATE_OK that changes the Forward State from 0 to 1”.

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

Drive the state machines for a response this endpoint is about to write on a request stream the peer opened.

The caller writes the message only after this returns Ok, so a response the endpoint refuses never reaches the wire.

Errors

EndpointError::UnknownRequest when no request of the answering kind carries that identifier, EndpointError::NotAResponse when the message is not one, EndpointError::UnjoinableSubscription and EndpointError::WrongJoiningRefusal for a Joining Fetch that named no live subscription, EndpointError::PeerPrefixOverlap and EndpointError::WrongOverlapRefusal for a namespace subscription that overlapped one already open or that a REQUEST_UPDATE asked to move onto an overlapping prefix, and each flow’s own InvalidTransition for a request already answered.

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

Dispatch a message that arrived on a request stream the peer opened, after the request that opened it.

Nothing that arrives here is a response: this endpoint is the responder on such a stream, so a SUBSCRIBE_OK or REQUEST_ERROR turning up is the peer answering its own request, and it is refused with EndpointError::UnexpectedOnPeerRequestStream. The same goes for NAMESPACE, NAMESPACE_DONE and PUBLISH_BLOCKED, which Table 5 puts on a request stream but which only ever travel from the responder to the requester — see send_response_on_stream.

Two messages are expected instead.

REQUEST_UPDATE, because draft-18 Section 10.9 puts it on the request’s own stream: “The sender of a request (SUBSCRIBE, PUBLISH, FETCH, PUBLISH_NAMESPACE, SUBSCRIBE_NAMESPACE, SUBSCRIBE_TRACKS) can later send a REQUEST_UPDATE on the same bidi stream as the request to modify it.” The request it modifies is therefore the stream’s, which is why request_id here comes from the stream and the message’s own Request ID field is not consulted — Section 10.1 says REQUEST_UPDATE consumes a Request ID of its own, so that field does not name the request being updated. Only a subscription carries a transition for an update; for the other kinds draft-18 defines no state change, so the message is accepted and the state left alone.

PUBLISH_DONE, because a peer that sent PUBLISH is the publisher and ends the publication it opened.