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#[allow(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    subscription::{ReceiverRouteActivity, ReceiverRouteCommit, ReceiverRouteWork},
36    topology::{
37        CommittedTransportReceipt, RoomTopology, SessionPlacementCommit, SessionPlacementRejection,
38    },
39};
40
41#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
42pub(super) struct SubscriptionKey {
43    pub receiver: UserId,
44    pub publisher: UserId,
45    pub stream: UserStreamId,
46}
47
48impl SubscriptionKey {
49    pub fn new(receiver: &UserId, publisher: &UserId, stream: &UserStreamId) -> Self {
50        Self {
51            receiver: receiver.clone(),
52            publisher: publisher.clone(),
53            stream: stream.clone(),
54        }
55    }
56}
57
58#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
59pub(super) struct SourceKey {
60    owner_user_id: UserId,
61    stream_id: UserStreamId,
62}
63
64#[derive(Debug)]
65pub(super) struct PublishedSource {
66    pub descriptor: PublishedSourceDescriptor,
67    pub transport: TransportSourceKey,
68    pub rtp: rtp::MediaStream,
69    pub routed: RoutedProducerId,
70    pub active: bool,
71    pub activity_revision: SourceActivityRevision,
72}
73
74#[derive(Debug, Clone)]
75pub(super) struct ConsumerRouteView<'a> {
76    pub key: &'a SubscriptionKey,
77    pub route: &'a TransportConsumerRoute,
78    pub mid: &'a str,
79    pub source: &'a PublishedSource,
80    pub selection: ConsumerSourceSelection,
81}
82
83impl ConsumerRouteView<'_> {
84    pub fn target(&self) -> ConsumerRouteTarget {
85        ConsumerRouteTarget::new(
86            self.route.clone(),
87            self.key.stream.clone(),
88            self.source.descriptor.media_kind(),
89        )
90    }
91}
92
93#[derive(Debug, Clone, Copy)]
94pub(super) struct PendingConsumerRouteView<'a> {
95    pub source: &'a PublishedSource,
96    pub selection: ConsumerSourceSelection,
97}
98
99#[derive(Debug, Clone, PartialEq, Eq)]
100pub struct ConsumerRouteTarget {
101    transport_route: TransportConsumerRoute,
102    stream_id: UserStreamId,
103    kind: MediaKind,
104}
105
106impl ConsumerRouteTarget {
107    fn new(
108        transport_route: TransportConsumerRoute,
109        stream_id: UserStreamId,
110        kind: MediaKind,
111    ) -> Self {
112        Self {
113            transport_route,
114            stream_id,
115            kind,
116        }
117    }
118
119    pub const fn transport_route(&self) -> &TransportConsumerRoute {
120        &self.transport_route
121    }
122
123    pub const fn consumer_media_id(&self) -> TransportMediaId {
124        self.transport_route.consumer_transport_media_id()
125    }
126
127    pub fn producer_user_id(&self) -> &UserId {
128        self.transport_route.source_session_key().user_id()
129    }
130
131    pub const fn source_media_id(&self) -> TransportMediaId {
132        self.transport_route.source_transport_media_id()
133    }
134
135    pub fn stream_id(&self) -> &UserStreamId {
136        &self.stream_id
137    }
138
139    pub fn request_keyframe_after_activity(&self, active: bool) -> bool {
140        active && self.kind == MediaKind::Video
141    }
142}
143
144impl SourceKey {
145    pub fn new(owner_user_id: &UserId, stream_id: &UserStreamId) -> Self {
146        Self {
147            owner_user_id: owner_user_id.clone(),
148            stream_id: stream_id.clone(),
149        }
150    }
151}
152
153fn remove_from_index_set<K, V>(index: &mut BTreeMap<K, BTreeSet<V>>, key: &K, value: &V)
154where
155    K: Ord,
156    V: Ord,
157{
158    let Some(values) = index.get_mut(key) else {
159        return;
160    };
161    values.remove(value);
162    if values.is_empty() {
163        index.remove(key);
164    }
165}
o_sfu_core/engine/room/media_graph/mod.rs

o_sfu_core/engine/room/media_graph/
mod.rs
