Struct RoomTopology 

pub struct RoomTopology {
    instance: RoomInstanceId,
    sources: PublishedSources,
    route_graph: RouteGraph,
    router: Router,
    next_producer_id: u64,
}

Keeps source records and route realization beside Router so transport effects resolve from the same committed placement.

Transport-facing mutations return resolved work for execution after the room state lock is released.

Fields

instance: RoomInstanceId

sources: PublishedSources

route_graph: RouteGraph

router: Router

next_producer_id: u64

Implementations

impl RoomTopology

pub fn new( runtime_context: &RoomRuntimeContext, router_rtp_capabilities: MediaCapabilities, ) -> Self

pub(in engine::room) fn router( &self, ) -> &Router

pub fn committed_transport_user_key( &self, user_id: impl Into<Arc<UserId>>, connection_id: ConnectionId, ) -> Option<TransportSessionKey>

Returns None unless the exact user connection remains committed.

pub fn transport_user_key( &self, user_id: impl Into<Arc<UserId>>, connection_id: ConnectionId, ) -> TransportSessionKey

Requires the exact user connection to remain committed.

Use Self::committed_transport_user_key for stale callbacks or teardown races where the placement may already be retired.

Lookup choice

Receiver work may carry a stale connection while relay effects from committed graph state use the strict lookup:

let Some(consumer_session) =
    topology.committed_transport_user_key(user_id.clone(), connection_id)
else {
    return Vec::new();
};

let source_session =
    topology.transport_user_key(route.source_user, route.source_connection);

Panics

Panics when no committed router placement exists.

fn transport_session_key( &self, user_id: Arc<UserId>, connection_id: ConnectionId, media_worker_id: MediaWorkerId, ) -> TransportSessionKey

pub fn retire_committed_placement( &mut self, user_id: &UserId, connection_id: ConnectionId, ) -> Option<TransportSessionKey>

Returns None when connection_id is not the user’s committed placement.

pub fn media_counts(&self) -> RoomMediaCounts

Counts each logical subscription once while pending or committed.

pub(in engine::room) fn source_descriptor( &self, source_id: PublishedSourceId, ) -> Option<&PublishedSourceDescriptor>

pub(in engine::room) fn source_for_transport_media( &self, transport_media_id: TransportMediaId, ) -> Option<&PublishedSource>

pub(in engine::room) fn source_id_for_owner_stream( &self, owner_user_id: &UserId, stream_id: &UserStreamId, ) -> Option<PublishedSourceId>

pub(in engine::room) fn published_source_id( &self, owner: &UserId, connection: ConnectionId, stream_id: &UserStreamId, ) -> Option<PublishedSourceId>

Returns the source ID only when owner, connection and stream remain current.

pub(in engine::room) fn published_sources( &self, ) -> impl Iterator<Item = &PublishedSource>

Iterates committed sources in source-ID order.

pub(in engine::room) fn active_stream_user_counts( &self, ) -> BTreeMap<UserStreamId, u64>

Counts distinct users with active publications for each logical stream.

pub(in engine::room) fn active_speaker_detector_owner( &self, transport_media_id: TransportMediaId, ) -> Option<UserId>

Returns a detector owner with an active promotable source in the same group.

pub(in engine::room) fn committed_consumer_routes( &self, ) -> impl Iterator<Item = ConsumerRouteView<'_>>

pub(in engine::room) fn committed_consumer_routes_for_user( &self, user_id: &UserId, ) -> impl Iterator<Item = ConsumerRouteView<'_>>

pub(super) fn detach_declined_consumers( &mut self, session: &TransportSessionKey, declined: &[TransportMediaId], ) -> (Vec<TransportRelayRouteEffect>, Vec<TransportTeardown>, bool)

pub(in engine::room) fn pending_consumer_routes_for_user( &self, user_id: &UserId, ) -> impl Iterator<Item = PendingConsumerRouteView<'_>>

pub(in engine::room) fn consumer_source_selection( &self, key: &SubscriptionKey, source_id: PublishedSourceId, ) -> Option<ConsumerSourceSelection>

Returns None when the attached source differs from source_id.

pub(in engine::room) fn merge_subscription_intent( &mut self, key: SubscriptionKey, intent: SourceSubscriptionIntent, )

Ignores empty updates and applies active to an attached selection.

pub(in engine::room) fn subscription_intent( &self, key: &SubscriptionKey, ) -> SourceSubscriptionIntent

Returns merged receiver intent or the default for a missing subscription.

pub(in engine::room) fn committed_consumer_user_ids_for_source( &self, source_id: PublishedSourceId, ) -> BTreeSet<UserId>

pub(in engine::room) fn committed_consumer_user_ids_for_owner_sources( &self, user_id: &UserId, ) -> BTreeSet<UserId>

pub(in engine::room) fn committed_consumer_route_for_key( &self, key: &SubscriptionKey, ) -> Option<ConsumerRouteView<'_>>

pub(in engine::room) fn published_source( &self, source_id: PublishedSourceId, ) -> Option<&PublishedSource>

pub(super) fn missing_consumer_targets_for_source<'a>( &mut self, source_id: PublishedSourceId, receivers: impl IntoIterator<Item = (&'a UserId, ConnectionId)>, ) -> Vec<ConsumerSetupTarget>

Attaches source_id to each eligible receiver lacking a route realization.

fn consumer_target( &mut self, user_id: &UserId, connection_id: ConnectionId, source_id: PublishedSourceId, ) -> Option<ConsumerSetupTarget>

fn attach_consumer_target( route_graph: &mut RouteGraph, consumer_session: &TransportSessionKey, source: &PublishedSource, ) -> Option<ConsumerSetupTarget>

Skips self-consumption and attaches only an absent realization.

pub(super) fn missing_consumer_targets( &mut self, user_id: &UserId, connection_id: ConnectionId, include_source: impl Fn(&PublishedSource) -> bool, ) -> Vec<ConsumerSetupTarget>

Attaches each matching source with no realization to the exact committed receiver.

Returns an empty list when connection_id is stale.

pub fn update_consumer_source_selection( &mut self, key: &SubscriptionKey, source_id: PublishedSourceId, route: &TransportConsumerRoute, update: impl FnOnce(&mut ConsumerSourceSelection), ) -> bool

Updates selection while subscription, source and exact route still match.

Returns false when async transport work refers to a displaced route.

fn detach_user_sources( &mut self, user_id: &UserId, ) -> (Vec<TransportSourceKey>, RemovedRoutes)

fn remove_source( &mut self, source_id: PublishedSourceId, ) -> Option<(PublishedSource, RemovedRoutes)>

fn consumer_route<'a>( &'a self, key: &'a SubscriptionKey, current: &'a CurrentPublication, ) -> Option<ConsumerRouteView<'a>>

pub(in engine::room) fn commit_publication( &mut self, publish: ValidatedPublish, rtp: RouterRtpParameters, encodings: &[SessionUploadEncoding], media: TransportMediaId, ) -> Result<PublishedSourceId, PublicationCommitError>

Commits a negotiated source only after its router producer succeeds.

Errors

Returns PublicationCommitError::Source when descriptor allocation or validation fails. Returns PublicationCommitError::Router when the router rejects the producer dependency.

fn media_teardowns( sources: impl IntoIterator<Item = TransportSourceKey>, routes: impl IntoIterator<Item = TransportConsumerRoute>, ) -> impl Iterator<Item = TransportTeardown>

fn resolve_relay_effects( &self, effects: impl IntoIterator<Item = RelayRouteEffect>, ) -> Vec<TransportRelayRouteEffect>

Panics

Panics if effects violates the topology invariant that every relay source has a committed router placement.

fn resolve_relay_effects_with_displaced( &self, effects: impl IntoIterator<Item = RelayRouteEffect>, user_id: &UserId, session_key: &TransportSessionKey, ) -> Vec<TransportRelayRouteEffect>

Resolves relay effects while retaining a displaced source’s transport key.

Panics

Panics if any non-displaced relay source lacks a committed placement.

pub fn set_published_source_activity( &mut self, source_id: PublishedSourceId, connection_id: ConnectionId, active: bool, ) -> Option<SourceActivityRevision>

Advances the revision only when the exact source connection changes activity.

Returns None when the source is missing, the connection is stale or the requested activity already matches.

pub(super) fn source_activity_effects( &self, source: &TransportSourceKey, update: SourceActivityUpdate, ) -> Vec<TransportSourceActivityEffect>

pub fn commit_session_placement( &mut self, user_id: &UserId, connection_id: ConnectionId, previous_connection: Option<ConnectionId>, home_placement: RouterPlacement, ) -> Result<SessionPlacementCommit, SessionPlacementRejection>

Commits the new placement before returning cleanup for previous_connection.

Replacement joins must pass the currently committed previous_connection.

Errors

Returns SessionPlacementRejection::MissingPreviousSession when the expected replacement target is no longer committed. Returns SessionPlacementRejection::Router when the router rejects the new connection or placement.

Panics

Panics when existing relay state refers to another uncommitted source placement.

pub fn remove_session(&mut self, user_id: &UserId) -> RoomTransportPlan

Returns the cleanup plan even if router removal fails.

Panics

Panics when detached relay state refers to an uncommitted source placement.

pub(super) fn commit_consumer_setup( &mut self, setup: DeclaredConsumerSetup, selection: ConsumerSourceSelection, ) -> Result<CommittedConsumerSetup, (TransportConsumerRoute, Vec<TransportRelayRouteEffect>)>

Selects MID from the transport declaration, negotiated RTP MID then the consumer identity.

Errors

Returns the declared route and relay release effects when the reservation is stale or the router rejects the consumer dependency.

pub(super) fn reserve_consumer_setup( &mut self, target: ConsumerSetupTarget, consumer: ConsumerId, selection: ConsumerSourceSelection, sender: UserOutboundSender, rtp: RouterRtpParameters, ) -> Option<PendingConsumerSetup>

Returns None unless the route realization remains absent.

A cross-worker reservation also claims relay ownership.

pub(super) fn release_consumer_setup( &mut self, setup: PendingConsumerSetup, ) -> Vec<TransportRelayRouteEffect>

A stale reservation releases no relay ownership.

pub(super) fn set_consumer_activity( &mut self, user_id: &UserId, connection_id: ConnectionId, target_user_id: &UserId, stream_id: &UserStreamId, active: bool, receiver_deafened: bool, ) -> Option<ConsumerActivityCommit>

Returns None when the source is missing, its attachment changed or the committed consumer belongs to another receiver connection.

Trait Implementations

impl Debug for RoomTopology

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

Formats the value using the given formatter.