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>,
    advertised_max_request_updates: u64,
    outstanding_peer_updates: HashMap<u64, u64>,
    unanswered_peer_updates: HashMap<u64, u64>,
    owed_update_failures: HashSet<u64>,
    answered_peer_requests: HashSet<u64>,
    advertised_max_filter_ranges: u64,
    peer_filter_rejections: HashMap<u64, FilterRejection>,
    peer_request_filters: HashMap<u64, Vec<KeyValuePair>>,
    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-19 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.

advertised_max_request_updates: u64

The MAX_REQUEST_UPDATES this endpoint put in its own SETUP.

The peer’s value is a different number and belongs to the sending side, which this type has no path for: draft-19 gives the endpoint no REQUEST_UPDATE builder, so there is nothing here to hold back.

Zero means no limit rather than none allowed, which is the opposite of how MAX_REQUEST_ID reads. Section 10.3.1.7 says so outright - “A value of 0 means the endpoint does not limit REQUEST_UPDATE concurrency. If not present, the default value is 0” - so an endpoint that never sends the option is not limiting anything, and a check that read zero as a ceiling would refuse the first update of every session.

outstanding_peer_updates: HashMap<u64, u64>

Per request stream, how many REQUEST_UPDATEs the peer has sent that this endpoint has not yet answered.

“Outstanding” is Section 10.3.1.7’s word and it is per stream, not per session: “Each REQUEST_OK or REQUEST_ERROR response restores one credit on that stream.”

unanswered_peer_updates: HashMap<u64, u64>

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

A second count rather than a reading of the one above, because the two sentences count different things and disagree on exactly one response. Section 10.3.1.7 says “Each REQUEST_OK or REQUEST_ERROR response restores one credit on that stream” — every response, including the REQUEST_OK that answers a SUBSCRIBE_NAMESPACE rather than an update. That is the sender’s accounting rule as much as the receiver’s, and an endpoint that credited more carefully than the peer does would close a conforming session, so the credit above stays literal.

This one is about which response answers which message, and there the request’s own REQUEST_OK answers the request. Keeping them apart is what lets a response be recognised as an update’s answer without changing what the peer is allowed to send.

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

advertised_max_filter_ranges: u64

The MAX_FILTER_RANGES this endpoint put in its own SETUP.

Section 10.3.1.6: “The MAX_FILTER_RANGES option (Type 0x06) limits the peer’s total number of Ranges (Start/End pairs) allowed concurrently in all Range filter parameters for a given subscription or fetch. The default value is 0, so if not specified, the peer MUST NOT send any such filter parameters.”

So zero is none allowed, and it is the default. The option in the subsection after this one, MAX_REQUEST_UPDATES, reads its zero the other way — no limit — and the two are otherwise the same shape. Reading either with the other’s rule is a working implementation that is wrong in one direction or the other for every session.

peer_filter_rejections: HashMap<u64, FilterRejection>

Requests whose Range Filters this endpoint owes the peer a REQUEST_ERROR about.

Keyed by Request ID, because that is what the reply names. Entries are spent by the REQUEST_ERROR that answers them, and a REQUEST_OK offered for one is refused: taking the request in and then accepting it would leave the peer with a subscription whose filters were never agreed.

peer_request_filters: HashMap<u64, Vec<KeyValuePair>>

The Range Filter parameters currently in force on each of the peer’s requests.

Section 5.1.3 makes the ceiling a property of the request rather than of the message that carried it — “the total number of Ranges allowed concurrently in all Range filter parameters for a given subscription or fetch” — and lets a REQUEST_UPDATE rewrite the set: “Length can be 0 to remove a filter parameter or non-zero to replace that entire filter parameter including all sets and Property Types. If a filter parameter is omitted from REQUEST_UPDATE, the value is unchanged.”

So the count that matters is of what is in force after the update, and the parameters an update leaves alone are part of it. Held as the parameters rather than as decoded filters because replacement is by Parameter Type, which is the key of the pair.

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.3. 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-19 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-19 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, SUBSCRIBE_TRACKS or PUBLISH_NAMESPACE, the responder MUST close the bidi stream (see Section 3.3.2).”

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>

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-19: no request_id on wire; the caller supplies the request_id of the bidi stream on which the response arrived.

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

Take one of the REQUEST_UPDATE credits this endpoint advertised for id’s stream.

Section 10.3.1.7 puts the limit on the number outstanding, so the count is a running balance rather than a total: it goes up on each REQUEST_UPDATE received and down on each REQUEST_OK or REQUEST_ERROR this endpoint writes back, and a peer that keeps pace never reaches the ceiling however many updates it sends.

Errors

EndpointError::TooManyRequestUpdates, which answers Some(TooManyRequestUpdates) - its own close code, not the general PROTOCOL_VIOLATION - and the session is failed before it returns.

fn restore_update_credit(&mut self, id: u64)

Give back the credit a REQUEST_OK or REQUEST_ERROR restores.

Called for every response this endpoint writes, whether or not the stream ever carried an update: a stream with no outstanding updates has nothing to restore and the saturating subtraction says so, which is cheaper than deciding first whether the response is answering an update or the original request.

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.

Draft-19 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. A subscriber can also send REQUEST_UPDATE to modify parameters of a subscription established with PUBLISH.” Anything else “MUST close the session with a PROTOCOL_VIOLATION”, which is what EndpointError::UnexpectedRequestUpdate carries — TRACK_STATUS most of all, since Section 10.14 says outright that “the subscriber cannot send REQUEST_UPDATE”.

The message carries a Request ID of its own and the stream carries one too. They name the same request when the peer is conforming; a disagreement means the update was sent on a stream that is not its request’s, which is the same violation, so it is refused rather than silently resolved to one of the two.

Only SUBSCRIBE-established subscriptions have a state-machine event for this. That is not an omission: an update changes a request’s parameters and not its lifecycle, so for the other five kinds the update is a self-transition with nothing to record.

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

joining_start counts groups back from the subscription’s Largest Group. To name the group directly, 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).

Draft-19 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, not an offset — which is what an application that knows the group it wants actually has. Expressing it as a relative fetch would need the Largest Group, which the subscriber may not know.

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

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-19: PUBLISH_OK is a REQUEST_OK alias, so this handler also resolves outstanding PUBLISH requests.

Track Properties are refused except on TRACK_STATUS_OK

Draft-19 Section 10.5: Track Properties “are populated in TRACK_STATUS_OK; they are empty in PUBLISH_OK, REQUEST_UPDATE_OK, SUBSCRIBE_NAMESPACE_OK and PUBLISH_NAMESPACE_OK. If an endpoint receives Track Properties in one of these messages it MUST close the session with a PROTOCOL_VIOLATION.” The codec cannot make that check — REQUEST_OK is one wire form and only the request stream says which of the five shapes it is. This is the layer that knows, because finding the request id in one of the maps below is what names the shape.

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.3: “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_skipped( &mut self, _msg: &PublishSkipped, ) -> Result<(), EndpointError>

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

Dispatch a message that arrived on the control stream.

Draft-19 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_SKIPPED names a track that will not 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_SKIPPED, 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-19 Table 5 places on a request stream without their being answers to it.

REQUEST_UPDATE modifies the request the stream carries (Section 10.9). GOAWAY is listed “Control, Request” because “A GOAWAY MAY also be sent on a request stream to initiate migration of that individual request” (Section 10.4); draft-19 is the draft where its receive path matters, since it removed GOAWAY’s Request ID and left one wire form for both places.

NAMESPACE (0x8) and NAMESPACE_DONE (0xE) arrive on the SUBSCRIBE_NAMESPACE request stream that asked for the namespaces they report, and PUBLISH_SKIPPED (0xF) on the SUBSCRIBE_TRACKS stream that asked for the track it says will not be published. Table 5 marks all three “Request”, so this is where they land; the control stream refuses them.

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-19 Section 10.4: “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”. The session keeps running — only this request is being moved — so the session state machine is not touched and no draining event follows. The server-side URI rule of the same section still applies.

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.

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

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.19 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.19 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 filter_verdict(&self, parameters: &[KeyValuePair]) -> Option<FilterRejection>

Whether the Range Filters in parameters are ones this endpoint may accept.

The three rules that need more than one parameter to see, in the order that reports the most specific fault: whether the filters read at all, then whether any budget was advertised, then whether the request stays inside it, then whether two filters share a key. A request with no Range Filter at all is not measured against the ceiling, so an endpoint that advertised nothing still takes ordinary requests — which is the whole of the traffic today, since draft-19 is the first draft with these parameters.

pub fn filter_rejection(&self, id: VarInt) -> Option<&FilterRejection>

Why request id must be answered with a REQUEST_ERROR, if it must.

The caller builds the message; FilterRejection::request_error_code gives the code and the variant gives the reason phrase. Answering it clears the record.

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

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, unless it is coalescing failed updates to produce just one REQUEST_ERROR for multiple REQUEST_UPDATE messages.” 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. That is the half that needs no ordering.

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 one or more outstanding updates.

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 its endpoint answers one update per message in both directions.

The credit MAX_REQUEST_UPDATES counts is restored once whatever the answer covered, which is Section 10.3.1.7 read literally: “Each REQUEST_OK or REQUEST_ERROR response restores one credit on that stream.” The peer restores by the same sentence, so an endpoint that gave back one per coalesced update would be crediting a peer that is not.

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 or spent.

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>

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.

Beyond the four responses this also takes the three messages draft-19 Table 5 places on a request stream that a responder writes after its response: NAMESPACE and NAMESPACE_DONE on a SUBSCRIBE_NAMESPACE stream (Sections 10.16 and 10.17) and PUBLISH_SKIPPED on a SUBSCRIBE_TRACKS stream (Section 10.20). All three require the request to have been accepted first, because each machine only leaves Pending on its REQUEST_OK.

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. 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, EndpointError::FilterMustBeRejected for a request whose Range Filters this draft says to reject, 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.

Three messages are expected instead.

REQUEST_UPDATE, because draft-19 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.” It goes through the same receive_request_update the requester side uses, so the peer’s update is held to the same rule its own would be: the message’s Request ID must name the stream’s request, and a kind the section does not allow to be updated — TRACK_STATUS, by Section 10.14 — closes the session.

GOAWAY, because Section 10.4 lets one arrive on a request stream to migrate that request alone, in either direction.

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