o_sfu_core/engine/room/
instance.rs

1use std::{fmt, sync::Arc};
2
3use tokio::sync::{Mutex as AsyncMutex, RwLock};
4
5use super::{definition::RoomDefinition, factory::RoomInit, state::RoomState};
6use crate::{
7    RoomWorkerPolicy,
8    engine::{
9        AvailableFeatures, ConnectionId, MediaWorkerId, PeerSnapshot, RecordingState,
10        RoomInstanceId, UserId,
11        media_transport::{MediaTransport, TransportSessionKey},
12        metrics::RuntimeMetrics,
13    },
14};
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
17pub enum RoomJoinError {
18    #[error("room is full")]
19    RoomFull,
20    #[error("router state error")]
21    RouterState,
22}
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
25pub enum RoomManagerJoinError {
26    #[error("room not found")]
27    MissingRoom,
28    #[error("room is full")]
29    RoomFull,
30    #[error("router state error")]
31    RouterState,
32}
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
35pub enum RoomManagerServeError {
36    #[error("conflicting room reservation")]
37    ConflictingReservation,
38}
39
40#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
41pub struct RoomMediaCounts {
42    pub publications: usize,
43    pub subscriptions: usize,
44}
45
46#[derive(Clone, Copy)]
47pub(crate) struct RoomUserOperation<'a> {
48    pub room: &'a Room,
49    pub user_id: &'a UserId,
50    pub connection_id: ConnectionId,
51    pub media_transport: &'a MediaTransport,
52}
53
54pub struct Room {
55    pub(super) definition: RoomDefinition,
56    pub(super) metrics: Arc<RuntimeMetrics>,
57    /// Serializes publication commits and activity changes with source-policy
58    /// turns so their transport effects cannot overtake each other.
59    pub(super) source_policy_turn: AsyncMutex<()>,
60    pub(super) state: RwLock<RoomState>,
61}
62
63impl Room {
64    pub(super) fn new(init: RoomInit) -> Self {
65        let RoomInit {
66            runtime_context,
67            runtime_policy,
68            issuer,
69            key,
70            config,
71            metrics,
72        } = init;
73        let definition =
74            RoomDefinition::new(&runtime_context, &runtime_policy, issuer, key, config);
75        Self {
76            definition,
77            metrics,
78            source_policy_turn: AsyncMutex::new(()),
79            state: RwLock::new(RoomState::new(
80                &runtime_context,
81                runtime_policy.admission_policy,
82                runtime_policy.media_limits,
83                runtime_policy.video_adaptation_tuning,
84                runtime_policy.router_rtp_capabilities,
85            )),
86        }
87    }
88
89    pub(crate) fn user_operation<'a>(
90        &'a self,
91        user_id: &'a UserId,
92        connection_id: ConnectionId,
93        media_transport: &'a MediaTransport,
94    ) -> RoomUserOperation<'a> {
95        RoomUserOperation {
96            room: self,
97            user_id,
98            connection_id,
99            media_transport,
100        }
101    }
102
103    #[must_use]
104    pub fn uuid(&self) -> &str {
105        self.definition.uuid()
106    }
107
108    #[must_use]
109    pub fn issuer(&self) -> &str {
110        self.definition.issuer()
111    }
112
113    #[must_use]
114    pub fn key(&self) -> &str {
115        self.definition.key()
116    }
117
118    #[must_use]
119    pub(crate) fn available_features(&self) -> AvailableFeatures {
120        self.definition.available_features()
121    }
122
123    pub async fn recording_state(&self) -> RecordingState {
124        self.state.read().await.recording_state()
125    }
126
127    pub(crate) async fn user_snapshots_except(
128        &self,
129        excluded_user_id: &UserId,
130    ) -> Vec<PeerSnapshot> {
131        self.state
132            .read()
133            .await
134            .user_snapshots_except(excluded_user_id)
135    }
136
137    #[must_use]
138    pub fn web_rtc_enabled(&self) -> bool {
139        self.definition.web_rtc_enabled()
140    }
141
142    /// Returns the transport key for an exact committed router placement.
143    ///
144    /// # Panics
145    ///
146    /// Panics when `user_id` and `connection_id` have no committed router
147    /// placement.
148    #[must_use]
149    pub async fn transport_user_key(
150        &self,
151        user_id: &UserId,
152        connection_id: ConnectionId,
153    ) -> TransportSessionKey {
154        self.state
155            .read()
156            .await
157            .transport_user_key(user_id, connection_id)
158    }
159
160    pub fn room_worker_policy(&self) -> RoomWorkerPolicy {
161        self.definition.room_worker_policy()
162    }
163
164    #[must_use]
165    pub(crate) fn instance_id(&self) -> RoomInstanceId {
166        self.definition.instance_id()
167    }
168}
169
170impl fmt::Debug for Room {
171    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
172        let media_worker_id = self
173            .state
174            .try_read()
175            .ok()
176            .and_then(|state| state.assigned_primary_media_worker_id())
177            .map(MediaWorkerId::as_usize);
178        formatter
179            .debug_struct("Room")
180            .field("instance_id", &self.definition.instance_id())
181            .field("media_worker_id", &media_worker_id)
182            .field("uuid", &self.definition.uuid())
183            .field("issuer", &self.definition.issuer())
184            .field("web_rtc_enabled", &self.definition.web_rtc_enabled())
185            .finish_non_exhaustive()
186    }
187}
o_sfu_core/engine/room/instance.rs

o_sfu_core/engine/room/
instance.rs
