o_sfu_core/engine/room/media_graph/
mod.rs

1use std::collections::{BTreeMap, BTreeSet};
2
3use o_sfu_router::{ConsumerId, MediaKind, rtp, topology::RoutedProducerId};
4
5use crate::engine::{
6    UserId,
7    media_transport::{
8        SourceActivityRevision, TransportConsumerRoute, TransportMediaId, TransportSourceKey,
9    },
10    source_model::{ConsumerSourceSelection, PublishedSourceDescriptor, UserStreamId},
11};
12
13mod consumer_setup;
14mod producer;
15mod route_graph;
16mod source_index;
17mod subscription;
18mod topology;
19
20#[cfg(test)]
21#[expect(non_snake_case, reason = "test modules map to local TESTS directories")]
22mod TESTS;
23#[cfg(test)]
24#[path = "TESTS/route_graph.rs"]
25mod route_graph_tests;
26
27#[cfg(any(test, feature = "testing-transport"))]
28pub use self::subscription::ConsumerRouteState;
29pub(super) use self::{
30    consumer_setup::{
31        CommittedConsumerSetup, ConsumerSetupOrigin, ConsumerSetupOutcome, ConsumerSetupTarget,
32        DeclaredConsumerSetup, PendingConsumerSetup,
33    },
34    producer::{ProducerActivityCommit, PublishCommit, PublishIntentPlan, ValidatedPublish},
35    route_graph::PendingUpgrade,
36    subscription::{ReceiverRouteActivity, ReceiverRouteCommit, ReceiverRouteWork},
37    topology::{RoomTopology, SessionPlacementCommit, SessionPlacementRejection},
38};
39
40#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
41pub(super) struct SubscriptionKey {
42    pub receiver: UserId,
43    pub publisher: UserId,
44    pub stream: UserStreamId,
45}
46
47impl SubscriptionKey {
48    pub fn new(receiver: &UserId, publisher: &UserId, stream: &UserStreamId) -> Self {
49        Self {
50            receiver: receiver.clone(),
51            publisher: publisher.clone(),
52            stream: stream.clone(),
53        }
54    }
55}
56
57#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
58pub(super) struct SourceKey {
59    owner_user_id: UserId,
60    stream_id: UserStreamId,
61}
62
63#[derive(Debug)]
64pub(super) struct PublishedSource {
65    pub descriptor: PublishedSourceDescriptor,
66    pub transport: TransportSourceKey,
67    pub rtp: rtp::MediaStream,
68    pub routed: RoutedProducerId,
69    pub active: bool,
70    pub activity_revision: SourceActivityRevision,
71}
72
73#[derive(Debug, Clone)]
74pub(super) struct ConsumerRouteView<'a> {
75    pub key: &'a SubscriptionKey,
76    pub route: &'a TransportConsumerRoute,
77    pub mid: &'a str,
78    /// Borrow holds so policy snapshots do not copy deadlines for every route.
79    pub pending_upgrade: Option<&'a PendingUpgrade>,
80    pub source: &'a PublishedSource,
81    pub selection: ConsumerSourceSelection,
82}
83
84impl ConsumerRouteView<'_> {
85    pub fn target(&self) -> ConsumerRouteTarget {
86        ConsumerRouteTarget::new(
87            self.route.clone(),
88            self.key.stream.clone(),
89            self.source.descriptor.media_kind(),
90        )
91    }
92}
93
94#[derive(Debug, Clone, Copy)]
95pub(super) struct PendingConsumerRouteView<'a> {
96    pub source: &'a PublishedSource,
97    pub selection: ConsumerSourceSelection,
98}
99
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub struct ConsumerRouteTarget {
102    transport_route: TransportConsumerRoute,
103    stream_id: UserStreamId,
104    kind: MediaKind,
105}
106
107impl ConsumerRouteTarget {
108    fn new(
109        transport_route: TransportConsumerRoute,
110        stream_id: UserStreamId,
111        kind: MediaKind,
112    ) -> Self {
113        Self {
114            transport_route,
115            stream_id,
116            kind,
117        }
118    }
119
120    pub const fn transport_route(&self) -> &TransportConsumerRoute {
121        &self.transport_route
122    }
123
124    pub const fn consumer_media_id(&self) -> TransportMediaId {
125        self.transport_route.consumer_transport_media_id()
126    }
127
128    pub fn producer_user_id(&self) -> &UserId {
129        self.transport_route.source_session_key().user_id()
130    }
131
132    pub const fn source_media_id(&self) -> TransportMediaId {
133        self.transport_route.source_transport_media_id()
134    }
135
136    pub fn stream_id(&self) -> &UserStreamId {
137        &self.stream_id
138    }
139
140    pub fn request_keyframe_after_activity(&self, active: bool) -> bool {
141        active && self.kind == MediaKind::Video
142    }
143}
144
145impl SourceKey {
146    pub fn new(owner_user_id: &UserId, stream_id: &UserStreamId) -> Self {
147        Self {
148            owner_user_id: owner_user_id.clone(),
149            stream_id: stream_id.clone(),
150        }
151    }
152}
153
154fn remove_from_index_set<K, V>(index: &mut BTreeMap<K, BTreeSet<V>>, key: &K, value: &V)
155where
156    K: Ord,
157    V: Ord,
158{
159    let Some(values) = index.get_mut(key) else {
160        return;
161    };
162    values.remove(value);
163    if values.is_empty() {
164        index.remove(key);
165    }
166}
o_sfu_core/engine/room/media_graph/mod.rs

o_sfu_core/engine/room/media_graph/
mod.rs
