Struct StreamRegistry 

pub(crate) struct StreamRegistry {
    live: Mutex<HashMap<StreamKey, StreamEntry>>,
}

Every forwarded stream that is still live, and the Gate each one releases when it ends.

This is what StreamAction::SerializeAfter waits on. It is always constructed, independently of whether the session has a ShapeProfile: SerializeAfter is gated by Interest::STREAMS, and the capability table publishes it as an unconditional Yes at both stream sites, so a registry that only existed when a profile was configured would make that published cell a lie. Empty, it is one HashMap header behind an Arc and allocates nothing until a stream registers.

Engine-internal despite living in a pub module: a caller names a StreamKey, never a registry.

Why a lookup miss is not an error

Self::gate_for answers None for a key that never existed and for one whose stream has already ended, because an entry is removed as it is released. Both mean the same thing to a waiter — there is nothing left to wait for — and the two are deliberately not distinguished: a serialize that resolves immediately is correct in both cases, and the caller reports SerializeTargetUnknown once so the run says which streams did it.

Fields

live: Mutex<HashMap<StreamKey, StreamEntry>>

Implementations

impl StreamRegistry

pub(crate) fn new() -> Self

An empty registry.

pub(crate) fn register( self: &Arc<Self>, key: StreamKey, inbox: Sender<StreamCommand>, ) -> StreamGuard

Register key as live, and hand back the guard that ends it.

The guard is the whole release mechanism, and it is a guard rather than a call at each teardown site on purpose. The gate has to be released on every termination path without exception — FIN, mirrored reset, synthesized reset, STOP_SENDING, cancellation — and a missed release is not a loud failure but a max_hold stall on some other stream. An enumerated list is only as complete as the reader; Drop is complete by construction, and it additionally covers the paths no enumeration would have listed: a ? return, a panicking forwarding task, and the task future being dropped wholesale by JoinSet::shutdown at session teardown. inbox is where a request naming this key is delivered; it is the sending half of a channel whose receiving half the forwarding task holds, so the entry going away and the task stopping are one event.

pub(crate) fn gate_for(&self, target: StreamKey) -> Option<Gate>

The gate to wait on for target, or None when there is nothing to wait for — see the type’s own doc for why those are one answer.

pub(crate) fn inbox_for( &self, target: StreamKey, ) -> Option<Sender<StreamCommand>>

Where to deliver a request aimed at target, or None when no such stream is live.

The same “never existed” / “already ended” conflation Self::gate_for makes, and for the same reason: the entry is removed as the stream ends, so nothing is left to tell the two apart. A caller reports both as “no such stream”, which is what a request aimed at either can act on.

fn end(&self, key: StreamKey)

Retire key: remove it and release its gate.

Idempotent, and the removal is what makes a later gate_for on the same key answer None rather than handing out an already-released gate that would look live.

Trait Implementations

impl Debug for StreamRegistry

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.