o_sfu_core/engine/room/state/
shared.rs

1use std::{
2    collections::{BTreeMap, BTreeSet},
3    sync::Arc,
4};
5
6use o_sfu_router::rtp::{MediaCapabilities, MediaCapabilities as RouterRtpCapabilities};
7
8use super::super::{
9    RoomAdmissionPolicy, RoomMediaCounts, RoomUserPermissions,
10    media_graph::{ConsumerRouteView, RoomTopology},
11    outbound::{
12        OutboundSender, RemoteTrackProjection, RemoteTrackSnapshot, VersionedRemoteTrackSnapshot,
13    },
14    transition::StagedPublishes,
15};
16use crate::{
17    RoomMediaLimits, VideoAdaptationTuning,
18    engine::{
19        ConnectionId, MediaWorkerId, PeerSnapshot, RecordingState, UserId, UserInfo,
20        media_transport::TransportSessionKey, room::placement::PlacementSnapshot,
21        source_model::UserStreamId,
22    },
23};
24
25#[derive(Debug)]
26pub struct RoomState {
27    pub(super) admission_policy: RoomAdmissionPolicy,
28    pub media_limits: RoomMediaLimits,
29    pub video_adaptation_tuning: VideoAdaptationTuning,
30    pub users: BTreeMap<UserId, ActiveUser>,
31    /// rejects stale async callbacks from previous connections
32    pub(super) next_connection_id: u64,
33    pub next_consumer_id: u64,
34    next_track_snapshot_revision: u64,
35    pub(super) recording_state: RecordingState,
36    pub(in crate::engine::room) staged_publishes: StagedPublishes,
37    pub(in crate::engine::room) topology: RoomTopology,
38}
39
40#[derive(Debug)]
41pub struct ActiveUser {
42    pub(super) user_id: Arc<UserId>,
43    pub(super) permissions: RoomUserPermissions,
44    pub(super) info: UserInfo,
45    pub(super) server_featured: Option<bool>,
46    pub parsed_client_rtp_capabilities: Option<RouterRtpCapabilities>,
47    pub connection_id: ConnectionId,
48    pub sender: OutboundSender,
49}
50
51impl ActiveUser {
52    pub(super) fn reset_presentation(&mut self) {
53        self.info = UserInfo::default();
54        self.server_featured = None;
55    }
56
57    pub(super) fn apply_info_update(&mut self, info: &UserInfo) {
58        self.info.apply_partial_update(info);
59    }
60
61    pub(in crate::engine::room) const fn featured(&self) -> Option<bool> {
62        self.server_featured
63    }
64
65    pub(in crate::engine::room) const fn is_deaf(&self) -> bool {
66        matches!(self.info.is_deaf, Some(true))
67    }
68
69    pub(in crate::engine::room) const fn is_screensharing(&self) -> bool {
70        matches!(self.info.is_screen_sharing_on, Some(true))
71    }
72
73    pub(in crate::engine::room) fn set_featured(&mut self, featured: Option<bool>) {
74        self.server_featured = featured;
75    }
76
77    pub(super) fn project_info(&self) -> UserInfo {
78        self.info
79            .clone()
80            .with_featured(self.server_featured)
81            .snapshot_complete()
82    }
83}
84
85impl RoomState {
86    pub fn new(
87        runtime_context: &super::super::RoomRuntimeContext,
88        admission_policy: RoomAdmissionPolicy,
89        media_limits: RoomMediaLimits,
90        video_adaptation_tuning: VideoAdaptationTuning,
91        router_rtp_capabilities: MediaCapabilities,
92    ) -> Self {
93        Self {
94            admission_policy,
95            media_limits,
96            video_adaptation_tuning,
97            users: BTreeMap::new(),
98            next_connection_id: 0,
99            next_consumer_id: 1,
100            next_track_snapshot_revision: 1,
101            recording_state: RecordingState {
102                recording: Some(false),
103                audio: Some(false),
104                transcription: Some(false),
105                video: Some(false),
106            },
107            staged_publishes: StagedPublishes::default(),
108            topology: RoomTopology::new(runtime_context, router_rtp_capabilities),
109        }
110    }
111
112    pub fn user_for_connection(
113        &self,
114        user_id: &UserId,
115        connection_id: ConnectionId,
116    ) -> Option<&ActiveUser> {
117        let user = self.users.get(user_id)?;
118        if user.connection_id != connection_id {
119            return None;
120        }
121        Some(user)
122    }
123
124    pub fn user_mut_for_connection(
125        &mut self,
126        user_id: &UserId,
127        connection_id: ConnectionId,
128    ) -> Option<&mut ActiveUser> {
129        let user = self.users.get_mut(user_id)?;
130        if user.connection_id != connection_id {
131            return None;
132        }
133        Some(user)
134    }
135
136    pub fn recording_state(&self) -> RecordingState {
137        self.recording_state.clone()
138    }
139
140    #[cfg(any(test, feature = "testing-transport"))]
141    pub fn router_rtp_capabilities(&self) -> MediaCapabilities {
142        self.topology.router().rtp_capabilities().clone()
143    }
144
145    pub fn transport_user_entries(&self) -> impl Iterator<Item = (&UserId, ConnectionId)> {
146        self.users
147            .iter()
148            .map(|(user_id, user)| (user_id, user.connection_id))
149    }
150
151    /// Returns the transport key for an exact committed router placement.
152    ///
153    /// Use [`Self::committed_transport_user_key`] when placement may be stale.
154    ///
155    /// # Panics
156    ///
157    /// Panics when `user_id` and `connection_id` have no committed router
158    /// placement.
159    pub fn transport_user_key(
160        &self,
161        user_id: &UserId,
162        connection_id: ConnectionId,
163    ) -> TransportSessionKey {
164        if let Some(user) = self.user_for_connection(user_id, connection_id) {
165            return self
166                .topology
167                .transport_user_key(Arc::clone(&user.user_id), connection_id);
168        }
169        self.topology
170            .transport_user_key(user_id.clone(), connection_id)
171    }
172
173    /// Returns `None` unless the exact router placement remains committed.
174    pub fn committed_transport_user_key(
175        &self,
176        user_id: &UserId,
177        connection_id: ConnectionId,
178    ) -> Option<TransportSessionKey> {
179        if let Some(user) = self.user_for_connection(user_id, connection_id) {
180            return self
181                .topology
182                .committed_transport_user_key(Arc::clone(&user.user_id), connection_id);
183        }
184        self.topology
185            .committed_transport_user_key(user_id.clone(), connection_id)
186    }
187
188    pub fn placement_usage_snapshot(&self) -> PlacementSnapshot {
189        self.topology.router().placement_snapshot()
190    }
191
192    pub fn assigned_primary_media_worker_id(&self) -> Option<MediaWorkerId> {
193        self.topology.router().primary_worker()
194    }
195
196    pub(in crate::engine::room) fn committed_consumer_routes(
197        &self,
198    ) -> impl Iterator<Item = ConsumerRouteView<'_>> {
199        self.topology.committed_consumer_routes().filter(|route| {
200            self.user_connection_id(&route.key.receiver)
201                == Some(route.route.consumer_session_key().connection_id())
202        })
203    }
204
205    pub fn user_connection_id(&self, user_id: &UserId) -> Option<ConnectionId> {
206        self.users.get(user_id).map(|user| user.connection_id)
207    }
208
209    pub fn user_count(&self) -> usize {
210        self.users.len()
211    }
212
213    pub fn user_snapshots_except(&self, excluded_user_id: &UserId) -> Vec<PeerSnapshot> {
214        self.users
215            .iter()
216            .filter(|(user_id, _session)| *user_id != excluded_user_id)
217            .map(|(user_id, user)| PeerSnapshot {
218                user_id: user_id.clone(),
219                info: user.project_info(),
220            })
221            .collect()
222    }
223
224    pub fn user_info_snapshot(&self, user_id: &UserId) -> Option<(UserId, UserInfo)> {
225        let user = self.users.get(user_id)?;
226        Some((user_id.clone(), user.project_info()))
227    }
228
229    pub(in crate::engine::room) fn remote_track_snapshot_for_user(
230        &mut self,
231        user_id: &UserId,
232        requires_negotiation: bool,
233    ) -> VersionedRemoteTrackSnapshot {
234        let revision = self.next_track_snapshot_revision;
235        self.next_track_snapshot_revision = revision.saturating_add(1);
236        let connection_id = self.user_connection_id(user_id);
237        let snapshot = RemoteTrackSnapshot {
238            tracks: self
239                .topology
240                .committed_consumer_routes_for_user(user_id)
241                .filter(|route| {
242                    connection_id == Some(route.route.consumer_session_key().connection_id())
243                })
244                .map(|route| {
245                    let source = &route.source.descriptor;
246                    RemoteTrackProjection {
247                        consumer_mid: route.mid.to_owned(),
248                        user_id: source.owner().user_id().clone(),
249                        stream_id: source.stream_id().clone(),
250                        producer_active: route.source.active,
251                    }
252                })
253                .collect(),
254            requires_negotiation,
255        };
256        VersionedRemoteTrackSnapshot { snapshot, revision }
257    }
258
259    pub(in crate::engine::room) fn remote_track_snapshots_for_users(
260        &mut self,
261        user_ids: BTreeSet<UserId>,
262        requires_negotiation: bool,
263    ) -> Vec<(OutboundSender, VersionedRemoteTrackSnapshot)> {
264        user_ids
265            .into_iter()
266            .filter_map(|user_id| {
267                Some((
268                    self.users.get(&user_id)?.sender.clone(),
269                    self.remote_track_snapshot_for_user(&user_id, requires_negotiation),
270                ))
271            })
272            .collect()
273    }
274
275    pub fn user_stats_counts(&self) -> (u64, BTreeMap<UserStreamId, u64>) {
276        (
277            u64::try_from(self.users.len()).unwrap_or(u64::MAX),
278            self.topology.active_stream_user_counts(),
279        )
280    }
281
282    pub fn media_counts(&self) -> RoomMediaCounts {
283        self.topology.media_counts()
284    }
285
286    pub fn is_empty(&self) -> bool {
287        self.users.is_empty()
288    }
289}
o_sfu_core/engine/room/state/shared.rs

o_sfu_core/engine/room/state/
shared.rs
