o_sfu_core/
lib.rs

1//! Room state, routing and media transport orchestration.
2//!
3//! `o-sfu-core` bridges the server runtime, the pure `o-sfu-router` state machine
4//! and the `str0m`-backed media transport. It keeps room admission, user media
5//! intent and RTC worker details behind [`SfuCore`](prelude::SfuCore) and
6//! [`MediaSession`](prelude::MediaSession).
7//!
8//! # Public Surface
9//!
10//! - [`prelude`] contains caller-facing configuration, media intent,
11//!   [`SfuCore`](prelude::SfuCore) and [`MediaSession`](prelude::MediaSession).
12//! - [`server`] contains runtime construction, room integration, transport,
13//!   diagnostics and metrics.
14//! - Fundamental identifiers and [`Bitrate`] remain at the crate root.
15//!
16//! # Architecture
17//!
18//! ```text
19//! server runtime
20//!   -> SfuCore::admit_user
21//!   -> MediaSession
22//!   -> room state and source policy
23//!   -> MediaTransport
24//!   -> RTC workers
25//! ```
26//!
27//! [`MediaTransport`](server::transport::MediaTransport) starts the worker threads
28//! and binds their UDP sockets. Room operations release state locks before awaiting
29//! transport work. Source policy maps layout intent, active-speaker observations
30//! and receiver bandwidth to route activity and packet gates. Worker-local packet
31//! loops then demultiplex UDP and forward RTP through those gates. The private
32//! `rtc::codec` boundary contains capability projection plus codec-specific packet
33//! inspection and rewrite, so source policy does not branch on payload details.
34//!
35//! # Server Construction
36//!
37//! The server builds one [`MediaTransport`](server::transport::MediaTransport)
38//! from owner configuration and shared process services.
39//!
40//! ```no_run
41//! use std::{
42//!     net::{IpAddr, Ipv4Addr},
43//!     sync::Arc,
44//! };
45//!
46//! use o_sfu_core::{
47//!     prelude::{
48//!         Bitrate, CodecPreferences, MediaCodecFlags, RtcPortRange, RtcUdpIoBackend,
49//!         SessionBitrateLimits, VideoBitrateLimits,
50//!     },
51//!     server::{
52//!         metrics::RuntimeMetrics,
53//!         packet_sinks::RoomPacketSinkRegistry,
54//!         transport::{MediaTransport, MediaTransportConfig, MediaTransportDeps},
55//!     },
56//! };
57//!
58//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
59//! let config = MediaTransportConfig {
60//!     worker_count: 1,
61//!     announced_ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
62//!     bitrate_limits: SessionBitrateLimits::new(
63//!         Bitrate::from_mbps(3),
64//!         Bitrate::from_mbps(3),
65//!     ),
66//!     video_bitrate_limits: VideoBitrateLimits::default(),
67//!     rtc_port_range: RtcPortRange::try_new(40_000, 40_099)?,
68//!     rtc_udp_io_backend: RtcUdpIoBackend::Tokio,
69//!     rtc_tcp_config: None,
70//!     codec_flags: MediaCodecFlags::default(),
71//!     codec_preferences: CodecPreferences::default(),
72//!     media_quality_interval: None,
73//! };
74//! let deps = MediaTransportDeps {
75//!     packet_sink_registry: Arc::new(RoomPacketSinkRegistry::default()),
76//!     metrics: Arc::new(RuntimeMetrics::default()),
77//! };
78//!
79//! let transport = MediaTransport::build(config, deps)?;
80//!
81//! # let _transport = transport;
82//! # Ok(())
83//! # }
84//! ```
85//!
86//! [`MediaTransport::build`](server::transport::MediaTransport::build) returns
87//! after every worker runtime has bound its UDP socket. Session-local RTC state
88//! remains lazy.
89//!
90//! # Session Negotiation
91//!
92//! Negotiation is serialized through `&mut MediaSession`. A publish received
93//! while an offer is pending is queued. Applying the answer returns a follow-up
94//! offer when that intent needs another SDP round.
95//!
96//! ```no_run
97//! use o_sfu_core::prelude::{
98//!     MediaSession, NegotiationOffer, SessionError, SourcePublishIntent,
99//! };
100//!
101//! # async fn exchange(_: NegotiationOffer) -> String { String::new() }
102//! async fn publish_source(
103//!     mut session: MediaSession,
104//!     intent: SourcePublishIntent,
105//! ) -> Result<(), SessionError> {
106//!     let Some(initial_offer) = session.establish().await? else {
107//!         return Ok(());
108//!     };
109//!     let initial_answer = exchange(initial_offer).await;
110//!
111//!     // Publish before answering so this intent queues behind the in-flight SDP round.
112//!     let _queued_without_offer = session.publish(intent).await?;
113//!
114//!     let Some(follow_up_offer) = session.answer(&initial_answer).await? else {
115//!         return Ok(());
116//!     };
117//!
118//!     let follow_up_answer = exchange(follow_up_offer).await;
119//!
120//!     let _next_offer = session.answer(&follow_up_answer).await?;
121//!     Ok(())
122//! }
123//! ```
124
125use std::fmt::{self, Display, Formatter};
126
127pub use o_sfu_router::{ConnectionId, MediaWorkerId};
128
129mod engine;
130mod options;
131pub mod prelude;
132pub mod server;
133mod sfu;
134
135pub(crate) use options::{
136    AudioCodecPreference, CodecPreferences, MediaCodecFlags, RoomMediaLimits, RoomWorkerPolicy,
137    RtcPortRange, RtcTcpConfig, RtcUdpIoBackend, RuntimeFeatureFlags, SessionBitrateLimits,
138    VideoAdaptationTuning, VideoBitrateLimits, VideoCodecPreference,
139};
140
141/// Media bitrate stored as bits per second (not bytes per second).
142#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
143pub struct Bitrate(u64);
144
145impl Bitrate {
146    #[must_use]
147    pub const fn from_bps(value: u64) -> Self {
148        Self(value)
149    }
150
151    #[must_use]
152    pub const fn from_kbps(value: u64) -> Self {
153        Self(value.saturating_mul(1_000))
154    }
155
156    #[must_use]
157    pub const fn from_mbps(value: u64) -> Self {
158        Self(value.saturating_mul(1_000_000))
159    }
160
161    #[must_use]
162    pub const fn zero() -> Self {
163        Self(0)
164    }
165
166    #[must_use]
167    pub const fn as_bps(self) -> u64 {
168        self.0
169    }
170
171    #[must_use]
172    pub const fn saturating_add(self, other: Self) -> Self {
173        Self(self.0.saturating_add(other.0))
174    }
175
176    #[must_use]
177    pub const fn saturating_sub(self, other: Self) -> Self {
178        Self(self.0.saturating_sub(other.0))
179    }
180
181    #[must_use]
182    pub const fn divided_by(self, divisor: u64) -> Self {
183        match self.0.checked_div(divisor) {
184            Some(value) => Self(value),
185            None => Self::zero(),
186        }
187    }
188}
189
190/// Process-local generation tag for one room lifecycle.
191///
192/// The tag separates one runtime allocation from its application room identity
193/// in transport keys and telemetry.
194#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
195pub struct RoomInstanceId(u64);
196
197impl RoomInstanceId {
198    /// Allocates the next room generation tag.
199    ///
200    /// The counter saturates, so repeated calls return `u64::MAX` after
201    /// exhaustion.
202    #[must_use]
203    pub fn allocate(next_room_instance_id: &mut u64) -> Self {
204        let room_instance_id = Self(*next_room_instance_id);
205        *next_room_instance_id = next_room_instance_id.saturating_add(1);
206        room_instance_id
207    }
208
209    #[must_use]
210    pub const fn from_raw(raw: u64) -> Self {
211        Self(raw)
212    }
213
214    #[must_use]
215    pub const fn as_u64(self) -> u64 {
216        self.0
217    }
218}
219
220impl Display for RoomInstanceId {
221    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
222        self.0.fmt(formatter)
223    }
224}
o_sfu_core/lib.rs

o_sfu_core/
lib.rs
