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::new(40_000, 40_099),
68//!     rtc_udp_io_backend: RtcUdpIoBackend::Tokio,
69//!     codec_flags: MediaCodecFlags::default(),
70//!     codec_preferences: CodecPreferences::default(),
71//!     media_quality_interval: None,
72//! };
73//! let deps = MediaTransportDeps {
74//!     packet_sink_registry: Arc::new(RoomPacketSinkRegistry::default()),
75//!     metrics: Arc::new(RuntimeMetrics::default()),
76//! };
77//!
78//! let transport = MediaTransport::build(config, deps)?;
79//!
80//! # let _transport = transport;
81//! # Ok(())
82//! # }
83//! ```
84//!
85//! [`MediaTransport::build`](server::transport::MediaTransport::build) returns
86//! after every worker runtime has bound its UDP socket. Session-local RTC state
87//! remains lazy.
88//!
89//! # Session Negotiation
90//!
91//! Negotiation is serialized through `&mut MediaSession`. A publish received
92//! while an offer is pending is queued. Applying the answer returns a follow-up
93//! offer when that intent needs another SDP round.
94//!
95//! ```no_run
96//! use o_sfu_core::prelude::{
97//!     MediaSession, NegotiationOffer, SessionError, SourcePublishIntent,
98//! };
99//!
100//! # async fn exchange(_: NegotiationOffer) -> String { String::new() }
101//! async fn publish_source(
102//!     mut session: MediaSession,
103//!     intent: SourcePublishIntent,
104//! ) -> Result<(), SessionError> {
105//!     let Some(initial_offer) = session.establish().await? else {
106//!         return Ok(());
107//!     };
108//!     let initial_answer = exchange(initial_offer).await;
109//!
110//!     // Publish before answering so this intent queues behind the in-flight SDP round.
111//!     let _queued_without_offer = session.publish(intent).await?;
112//!
113//!     let Some(follow_up_offer) = session.answer(&initial_answer).await? else {
114//!         return Ok(());
115//!     };
116//!
117//!     let follow_up_answer = exchange(follow_up_offer).await;
118//!
119//!     let _next_offer = session.answer(&follow_up_answer).await?;
120//!     Ok(())
121//! }
122//! ```
123
124use std::fmt::{self, Display, Formatter};
125
126pub use o_sfu_router::{ConnectionId, MediaWorkerId};
127
128mod engine;
129mod options;
130pub mod prelude;
131pub mod server;
132mod sfu;
133
134pub(crate) use options::{
135    AudioCodecPreference, CodecPreferences, MediaCodecFlags, RoomMediaLimits, RoomWorkerPolicy,
136    RtcPortRange, RtcUdpIoBackend, RuntimeFeatureFlags, SessionBitrateLimits,
137    VideoAdaptationTuning, VideoBitrateLimits, VideoCodecPreference,
138};
139
140/// Media bitrate stored as bits per second (not bytes per second).
141#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
142pub struct Bitrate(u64);
143
144impl Bitrate {
145    #[must_use]
146    pub const fn from_bps(value: u64) -> Self {
147        Self(value)
148    }
149
150    #[must_use]
151    pub const fn from_kbps(value: u64) -> Self {
152        Self(value.saturating_mul(1_000))
153    }
154
155    #[must_use]
156    pub const fn from_mbps(value: u64) -> Self {
157        Self(value.saturating_mul(1_000_000))
158    }
159
160    #[must_use]
161    pub const fn zero() -> Self {
162        Self(0)
163    }
164
165    #[must_use]
166    pub const fn as_bps(self) -> u64 {
167        self.0
168    }
169
170    #[must_use]
171    pub const fn saturating_add(self, other: Self) -> Self {
172        Self(self.0.saturating_add(other.0))
173    }
174
175    #[must_use]
176    pub const fn saturating_sub(self, other: Self) -> Self {
177        Self(self.0.saturating_sub(other.0))
178    }
179
180    #[must_use]
181    pub const fn divided_by(self, divisor: u64) -> Self {
182        match self.0.checked_div(divisor) {
183            Some(value) => Self(value),
184            None => Self::zero(),
185        }
186    }
187}
188
189/// Process-local generation tag for one room lifecycle.
190///
191/// The tag separates one runtime allocation from its application room identity
192/// in transport keys and telemetry.
193#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
194pub struct RoomInstanceId(u64);
195
196impl RoomInstanceId {
197    /// Allocates the next room generation tag.
198    ///
199    /// The counter saturates, so repeated calls return `u64::MAX` after
200    /// exhaustion.
201    #[must_use]
202    pub fn allocate(next_room_instance_id: &mut u64) -> Self {
203        let room_instance_id = Self(*next_room_instance_id);
204        *next_room_instance_id = next_room_instance_id.saturating_add(1);
205        room_instance_id
206    }
207
208    #[must_use]
209    pub const fn from_raw(raw: u64) -> Self {
210        Self(raw)
211    }
212
213    #[must_use]
214    pub const fn as_u64(self) -> u64 {
215        self.0
216    }
217}
218
219impl Display for RoomInstanceId {
220    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
221        self.0.fmt(formatter)
222    }
223}
o_sfu_core/lib.rs

o_sfu_core/
lib.rs
