media: RtcSession + str0m poll loop (the media core)
RtcSession owns a str0m::Rtc + Opus decoder/encoder + EchoAudioPipe + a bound UDP socket (spec §3.1, §4.5). accept_offer calls str0m 0.21's sdp_api().accept_offer() natively — no hand-rolled SDP munger; str0m fills DTLS fingerprint + ICE creds + Opus codec. The loop driver drains poll_output per str0m's single-mutation invariant, routes inbound MediaData through Opus decode + EchoAudioPipe sink, sends Transmit packets on the UDP socket, and checks the 60 s idle timeout. DEV-DEVIATION: loop runs on tokio (spec §3.4); step 4 replaces with a dedicated timing thread per ARCHITECTURE.md. str0m 0.21 API adjustments from the brief's sketch (verified against src/str0m-0.21.0): - SdpAnswer has no .mid() accessor; use answer.media_lines[0].mid() via the Deref<Target=Sdp> impl. - Input::Receive carries the Instant as the first tuple element (Input::Receive(now, recv)); handle_input takes a single Input arg. - Receive constructed via Receive::new(proto, src, dst, buf) (the DatagramRecv field is private). - UDP socket binds 127.0.0.1:0 (Candidate::host rejects the unspecified address 0.0.0.0). - BROWSER_SDP_OFFER fixture restored a=group:BUNDLE 0 (str0m's Sdp::assert_consistency rejects SDP without a session group).
This commit is contained in:
adlee-was-taken
@@ -10,6 +10,7 @@ description = "Media core: str0m WebRTC + Opus⇄PCM boundary (slice 1)."
|
||||
[dependencies]
|
||||
rutster-call-model = { path = "../rutster-call-model" }
|
||||
opus = { workspace = true }
|
||||
str0m = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
|
||||
|
||||
@@ -30,11 +30,14 @@
|
||||
//! - [`loop_driver`] (Task 4) — the str0m poll loop on tokio.
|
||||
//! - [`rtc_session`] (Task 4) — `RtcSession`, the per-peer owner.
|
||||
|
||||
pub mod loop_driver;
|
||||
pub mod opus_codec;
|
||||
pub mod pcm;
|
||||
pub mod rtc_session;
|
||||
|
||||
pub use opus_codec::{OpusDecoder, OpusEncoder};
|
||||
pub use pcm::{AudioSink, AudioSource, EchoAudioPipe, PcmFrame, SAMPLES_PER_FRAME};
|
||||
pub use rtc_session::{RtcSession, RtcSessionError};
|
||||
|
||||
use thiserror::Error;
|
||||
|
||||
|
||||
258
crates/rutster-media/src/loop_driver.rs
Normal file
258
crates/rutster-media/src/loop_driver.rs
Normal file
@@ -0,0 +1,258 @@
|
||||
//! # str0m poll loop (spec §3.2, §3.4)
|
||||
//!
|
||||
//! The heart of the media core. Drives the `str0m::Rtc` instance forward
|
||||
//! on each call: drains `poll_output()` until `Output::Timeout`, handling
|
||||
//! each `Output::Transmit` (send on our UDP socket) and `Output::Event`
|
||||
//! (inbound `MediaData` → Opus decode → sink; inbound RTP count for the
|
||||
//! idle timeout). When the drain returns `Timeout`, the caller sleeps
|
||||
//! that duration and calls back with `Input::Timeout`.
|
||||
//!
|
||||
//! # Why this lives in a separate module
|
||||
//!
|
||||
//! `run_poll_once` takes `&mut RtcSession` — a single function with
|
||||
//! the full poll logic would make `RtcSession::run_poll_once` 100+ lines
|
||||
//! of non-trivial control flow. Splitting the loop into a module makes
|
||||
//! the sans-IO pattern obvious: the loop driver takes a `&mut RtcSession`,
|
||||
//! reads str0m outputs, and writes str0m inputs. Nothing else.
|
||||
//!
|
||||
//! # DEV-DEVIATION
|
||||
//!
|
||||
//! Slice 1 runs the poll on a tokio task. ARCHITECTURE.md mandates a
|
||||
//! dedicated timing thread; we defer that to step 4 (barge-in) because
|
||||
//! slice 1 has no reflex to time against. The poll function's shape
|
||||
//! (single `&mut self`, no I/O inside) makes the step-4 swap localized.
|
||||
|
||||
use std::io::ErrorKind;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use str0m::net::Protocol;
|
||||
use str0m::{Input, Output};
|
||||
|
||||
use crate::pcm::{AudioSink as _, AudioSource as _};
|
||||
use crate::rtc_session::{RtcSession, IDLE_TIMEOUT};
|
||||
|
||||
/// 20 ms tick for outbound encoding (matches the PCM frame size, spec §3.9:
|
||||
/// 480 samples @ 24 kHz = 20 ms). On each tick, we pull one frame from the
|
||||
/// source pipe and write the encoded Opus via str0m's `Writer::write`.
|
||||
const OUTBOUND_TICK: Duration = Duration::from_millis(20);
|
||||
|
||||
/// One iteration of the str0m poll loop.
|
||||
///
|
||||
/// 1. Read any pending UDP packets (non-blocking) and feed each to str0m
|
||||
/// as `Input::Receive`. A WouldBlock means no packets this cycle — fine.
|
||||
/// 2. Drain `poll_output()` until `Timeout`:
|
||||
/// - `Transmit` → send on our UDP socket.
|
||||
/// - `Event::MediaData` → decode Opus → push to the echo pipe (sink).
|
||||
/// - `Event::IceConnectionStateChange` → state transition + tracing.
|
||||
/// - We don't break out of the drain on any of these: str0m's contract
|
||||
/// is mutate → drain to `Timeout` → mutate (see str0m 0.21 lib docs).
|
||||
/// 3. **Outbound encode tick:** if ≥20 ms of wallclock passed since the
|
||||
/// last outbound frame, pull one `PcmFrame` from the source, encode to
|
||||
/// Opus, and write via `Rtc::writer(mid)->Writer::write`. Then re-drain
|
||||
/// `poll_output` (the Writer write is a mutation → must drain per str0m).
|
||||
/// 4. Check the idle timeout: if `now - last_rx > IDLE_TIMEOUT`, transition
|
||||
/// to `Closed`.
|
||||
/// 5. Return the `Duration` to the next `Timeout`.
|
||||
pub fn drive(session: &mut RtcSession, now: Instant) -> Option<Duration> {
|
||||
// === Step 1: drain our UDP socket non-blocking, feed str0m. ===
|
||||
let mut buf = [0u8; 2000];
|
||||
loop {
|
||||
match session.socket.recv_from(&mut buf) {
|
||||
Ok((n, source)) => {
|
||||
// `Receive::new` parses the raw datagram into one of
|
||||
// STUN/DTLS/RTP/RTCP (str0m's demultiplexer). It returns
|
||||
// `Err(NetError)` for things str0m can't classify; we drop
|
||||
// those, don't crash (hot-path policy, spec §3.8).
|
||||
let recv = match str0m::net::Receive::new(
|
||||
Protocol::Udp,
|
||||
source,
|
||||
session.local_addr,
|
||||
&buf[..n],
|
||||
) {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = ?e, "str0m datagram parse failed; dropping");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
// str0m's `Input::Receive` carries the receive timestamp as
|
||||
// the first tuple element (str0m 0.21 API — the brief's
|
||||
// sketch had `Input::Receive(recv)` and `handle_input(now, ...)`,
|
||||
// both adjusted to the actual surface: a single `Input`
|
||||
// argument, with the Instant packed into the variant).
|
||||
if session.rtc.handle_input(Input::Receive(now, recv)).is_err() {
|
||||
// Hot-path policy: drop + observe, don't crash.
|
||||
tracing::warn!("str0m rejected input packet; dropping");
|
||||
}
|
||||
session.last_rx = now;
|
||||
}
|
||||
// WouldBlock (unix) / TimedOut (windows) — no packets this cycle.
|
||||
Err(e) if matches!(e.kind(), ErrorKind::WouldBlock | ErrorKind::TimedOut) => break,
|
||||
Err(e) => {
|
||||
tracing::warn!(error = ?e, "UDP recv_from error; continuing");
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// === Step 2: drain poll_output, interleaving outbound writes. ===
|
||||
// `next_timeout` is set in either the `Timeout` or `Err` exit arms of
|
||||
// the drain loop below before any read on the path forward — no
|
||||
// initial value needed. Both exit arms assign before breaking, so the
|
||||
// borrow checker is satisfied; clippy flagged the previous `= None`
|
||||
// initializer as a write-then-overwrite.
|
||||
let mut next_timeout: Option<Instant>;
|
||||
// Track whether we owe a Writer write this cycle; re-drain if so.
|
||||
// str0m's "mutate → drain to Timeout" invariant: after Writer::write,
|
||||
// poll_output must be drained to Timeout before any other mutation.
|
||||
let mut needs_redrain = false;
|
||||
loop {
|
||||
match session.rtc.poll_output() {
|
||||
Ok(Output::Timeout(t)) => {
|
||||
next_timeout = Some(t);
|
||||
if needs_redrain {
|
||||
// We did an outbound write in the previous iteration;
|
||||
// str0m needs to be drained again. Loop continues,
|
||||
// but only handle Transmit/Event briefly before next Timeout.
|
||||
needs_redrain = false;
|
||||
continue;
|
||||
}
|
||||
break; // engine is fully drained
|
||||
}
|
||||
Ok(Output::Transmit(t)) => {
|
||||
// `Transmit.contents` is a `DatagramSend` newtype that
|
||||
// `Deref`s to `[u8]` — passing `&t.contents` deref-coerces
|
||||
// to `&[u8]` for `send_to`.
|
||||
if let Err(e) = session.socket.send_to(&t.contents, t.destination) {
|
||||
if !matches!(e.kind(), ErrorKind::WouldBlock) {
|
||||
tracing::warn!(error = ?e, "UDP send_to error; dropping");
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(Output::Event(event)) => {
|
||||
handle_event(session, event, now);
|
||||
// Loop continues — mutations from inside the drain loop
|
||||
// are fine (str0m docs, "single-mutation invariant"):
|
||||
// events are observations, not external mutations.
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(error = ?e, "str0m poll_output error; continuing");
|
||||
next_timeout = Some(now + OUTBOUND_TICK);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// === Step 3: outbound encode tick (the echo path). ===
|
||||
// If wallclock has crossed a 20 ms boundary since the last outbound
|
||||
// frame, pull a PcmFrame from the source, encode to Opus, and write
|
||||
// via Writer::write. This IS the slice-1 echo: inbound decode → pipe
|
||||
// → outbound encode.
|
||||
if now.duration_since(session.last_outbound_at) >= OUTBOUND_TICK {
|
||||
if let Some(mid) = session.audio_mid {
|
||||
if let Some(frame) = session.pipe.next_pcm_frame() {
|
||||
if let Some(opus_payload) = session.encoder.encode(&frame) {
|
||||
// Writer::write signature (str0m 0.21, verified):
|
||||
// write(pt: Pt, wallclock: Instant, rtp_time: MediaTime,
|
||||
// data: impl Into<Arc<[u8]>>) -> Result<(), RtcError>
|
||||
// - pt: payload type for Opus. `writer.payload_params()`
|
||||
// returns `impl Iterator<Item = &PayloadParams>`; the
|
||||
// first one's `.pt()` is our Opus PT (str0m negotiates
|
||||
// this in the SDP answer).
|
||||
// - wallclock: when the sample was produced — local `now`.
|
||||
// - rtp_time: RTP timestamp in the 48 kHz audio clock for
|
||||
// Opus. Increment by 960 per 20 ms (48000 * 0.020).
|
||||
// `MediaTime` has no `add(Duration)` method — use
|
||||
// `mt + MediaTime::from(duration)`.
|
||||
//
|
||||
// `rtc.writer(mid)` returns `Option<Writer<'_>>` — `None`
|
||||
// if direction isn't sending (we'd be in a recvonly state).
|
||||
if let Some(writer) = session.rtc.writer(mid) {
|
||||
// Pull the Opus payload type out of the iterator
|
||||
// BEFORE `writer.write(...)`, which consumes `writer`
|
||||
// by value. Doing both inline trips E0505 because
|
||||
// `payload_params()` borrows `writer` while
|
||||
// `write(self, ...)` moves it — so separate the
|
||||
// borrow from the move with a tight scope.
|
||||
let pt = writer.payload_params().next().map(|p| p.pt());
|
||||
if let Some(pt) = pt {
|
||||
let rtp_time = session.next_media_time;
|
||||
if writer
|
||||
.write(pt, now, rtp_time, opus_payload.as_slice())
|
||||
.is_ok()
|
||||
{
|
||||
// Advance media time for next 20 ms frame.
|
||||
// `MediaTime + MediaTime::from(Duration)` —
|
||||
// no `add()` method on `MediaTime` in str0m 0.21.
|
||||
session.next_media_time +=
|
||||
str0m::media::MediaTime::from(Duration::from_millis(20));
|
||||
needs_redrain = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
session.last_outbound_at = now;
|
||||
}
|
||||
}
|
||||
|
||||
// If the outbound write happened, we owe str0m one more drain before
|
||||
// returning — Writer::write is a mutation per str0m's invariant.
|
||||
if needs_redrain {
|
||||
loop {
|
||||
match session.rtc.poll_output() {
|
||||
Ok(Output::Timeout(t)) => {
|
||||
next_timeout = Some(t);
|
||||
break;
|
||||
}
|
||||
Ok(Output::Transmit(t)) => {
|
||||
let _ = session.socket.send_to(&t.contents, t.destination);
|
||||
}
|
||||
Ok(Output::Event(e)) => handle_event(session, e, now),
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// === Step 4: idle timeout (spec §4.5). ===
|
||||
if now.duration_since(session.last_rx) > IDLE_TIMEOUT {
|
||||
tracing::info!(
|
||||
channel_id = %session.channel.id,
|
||||
"idle timeout (60 s no RX); closing session"
|
||||
);
|
||||
session.channel.state = rutster_call_model::ChannelState::Closed;
|
||||
return None;
|
||||
}
|
||||
|
||||
session.next_timeout = next_timeout;
|
||||
next_timeout.map(|t| t.saturating_duration_since(now))
|
||||
}
|
||||
|
||||
/// Dispatch a str0m `Event` to the audio pipe or to state bookkeeping.
|
||||
fn handle_event(session: &mut RtcSession, event: str0m::Event, _now: Instant) {
|
||||
use str0m::Event;
|
||||
match event {
|
||||
Event::MediaData(media) => {
|
||||
// Inbound decoded audio frame from the peer (Frame API, spec §3.2).
|
||||
// str0m has already done RTP depacketization; `MediaData.data`
|
||||
// is the encoded Opus payload (type: `Arc<[u8]>` — passing
|
||||
// `&media.data` deref-coerces to `&[u8]` for `OpusDecoder::decode`).
|
||||
if let Some(pcm) = session.decoder.decode(&media.data) {
|
||||
session.pipe.on_pcm_frame(pcm);
|
||||
}
|
||||
// Decode failed → drop + observe (per §3.8). Don't kill the peer.
|
||||
}
|
||||
Event::IceConnectionStateChange(state) => {
|
||||
tracing::info!(
|
||||
channel_id = %session.channel.id,
|
||||
?state,
|
||||
"ICE state change"
|
||||
);
|
||||
if state == str0m::IceConnectionState::Connected {
|
||||
session.channel.state = rutster_call_model::ChannelState::Connected;
|
||||
}
|
||||
}
|
||||
Event::EgressBitrateEstimate(_) => { /* BWE — irrelevant in slice 1 */ }
|
||||
_ => { /* str0m emits several other event variants we don't need in slice 1. */ }
|
||||
}
|
||||
}
|
||||
300
crates/rutster-media/src/rtc_session.rs
Normal file
300
crates/rutster-media/src/rtc_session.rs
Normal file
@@ -0,0 +1,300 @@
|
||||
//! # `RtcSession` — the per-peer media owner (spec §3.1, §4.5)
|
||||
//!
|
||||
//! Owns a `str0m::Rtc` instance + an Opus decoder/encoder pair + an
|
||||
//! `EchoAudioPipe` wiring inbound to outbound + the per-peer UDP socket.
|
||||
//! One per WebRTC peer. The `ChannelId` (from `rutster-call-model`) is
|
||||
//! the session id surfaced in the REST API.
|
||||
//!
|
||||
//! ## What str0m does for us (so we don't)
|
||||
//!
|
||||
//! str0m 0.21's `Rtc::sdp_api().accept_offer(offer)` produces the SDP
|
||||
//! answer natively: DTLS fingerprint (from the cert str0m generates), ICE
|
||||
//! ufrag/pwd, and codec negotiation (Opus, the only codec we registered).
|
||||
//! Slice 1 does NOT hand-roll an SDP munger — str0m's path is the spec's
|
||||
//! "embryo of the future SIP SDP path" (§3.7). When step 5 brings SIP/SDP
|
||||
//! negotiation into `rutster-signaling-sip`, that crate may extract shared
|
||||
//! SDP helpers from str0m or build its own. Slice 1's WebRTC-ICE-coupled
|
||||
//! SDP lives entirely in str0m.
|
||||
|
||||
use std::net::SocketAddr;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use rutster_call_model::{Channel, ChannelId, ChannelState};
|
||||
use str0m::Rtc;
|
||||
use thiserror::Error;
|
||||
|
||||
use crate::opus_codec::{OpusDecoder, OpusEncoder};
|
||||
use crate::pcm::EchoAudioPipe;
|
||||
|
||||
/// Per-session idle timeout (spec §4.5): 60 s of no RTP from the peer
|
||||
/// → close. RTC quiet periods are normal but 60 s of dead air means
|
||||
/// "the browser tab is dead."
|
||||
pub(crate) const IDLE_TIMEOUT: Duration = Duration::from_secs(60);
|
||||
|
||||
/// Cold-path errors for `RtcSession` construction and SDP negotiation.
|
||||
///
|
||||
/// Hot-path failures (decode, encode, UDP recv) follow the
|
||||
/// match-and-continue "drop + observe" policy from spec §3.8 — they
|
||||
/// never reach this enum.
|
||||
#[derive(Debug, Error)]
|
||||
pub enum RtcSessionError {
|
||||
/// Two-stage failure from str0m's SDP path: `SdpOffer::from_sdp_string`
|
||||
/// can fail to parse, OR `accept_offer` can reject the parsed offer.
|
||||
/// Both surface as `str0m::RtcError` / `str0m::sdp::SdpError`; we
|
||||
/// collapse them via a display-format `String` since both are
|
||||
/// display-format-only at the axum boundary (HTTP 400 in `routes.rs`).
|
||||
/// Using `String` (not `#[source]`) keeps the variants uniform and the
|
||||
/// axum layer doesn't need to downcast — it logs + 400s.
|
||||
#[error("SDP offer parse or accept failed: {0}")]
|
||||
SdpOffer(String),
|
||||
#[error("opus codec init failed: {0}")]
|
||||
Codec(#[from] opus::Error),
|
||||
#[error("UDP socket bind failed: {0}")]
|
||||
Socket(#[from] std::io::Error),
|
||||
}
|
||||
|
||||
use str0m::change::SdpOffer;
|
||||
use str0m::media::Mid;
|
||||
|
||||
/// The per-peer media owner (spec §3.1, §4.5).
|
||||
///
|
||||
/// # Ownership / sharing
|
||||
///
|
||||
/// An `RtcSession` lives behind an `Arc<Mutex<RtcSession>>` in the
|
||||
/// binary's `DashMap<ChannelId, RtcSession>` (Task 5). The mutex is
|
||||
/// short-held: each tokio poll iteration locks, runs `run_poll_once`,
|
||||
/// unlocks. We do NOT hold the lock across `tokio::time::sleep` — that
|
||||
/// would defeat the `DashMap`'s sharded concurrency and pre-pave the
|
||||
/// wrong pattern for step 4's dedicated thread.
|
||||
///
|
||||
/// # Why `Arc<Mutex<...>>` (not `Arc<RwLock<...>>`)
|
||||
///
|
||||
/// Every access of an `RtcSession` mutates it (str0m's `&mut self`
|
||||
/// contract on `handle_input` + `poll_output`). `RwLock`'s read-mode
|
||||
/// would be useless because str0m takes `&mut Rtc`. `Mutex` it is.
|
||||
pub struct RtcSession {
|
||||
pub(crate) channel: Channel,
|
||||
pub(crate) rtc: Rtc,
|
||||
pub(crate) decoder: OpusDecoder,
|
||||
pub(crate) encoder: OpusEncoder,
|
||||
pub(crate) pipe: EchoAudioPipe,
|
||||
/// Local UDP socket str0m sends `Transmit` packets out on and
|
||||
/// receives `Input::Receive` packets from. Bound to an ephemeral
|
||||
/// port at construction; the local candidate passed to str0m at
|
||||
/// offer-accept time uses this address.
|
||||
pub(crate) socket: std::net::UdpSocket,
|
||||
/// Local socket address — cached because `local_addr()` is a syscall.
|
||||
pub(crate) local_addr: SocketAddr,
|
||||
/// Mid of the audio m-line we accepted. Set during `accept_offer`.
|
||||
/// Slice 1 has exactly one m-line; multi-m-line arrives with video.
|
||||
pub(crate) audio_mid: Option<Mid>,
|
||||
/// Last deadline from `Rtc::poll_output` — the next time the loop
|
||||
/// should wake the rtc with `Input::Timeout`.
|
||||
pub(crate) next_timeout: Option<Instant>,
|
||||
/// Last Instant we received an RTP packet from the peer. Used for
|
||||
/// the 60 s idle timeout (spec §4.5).
|
||||
pub(crate) last_rx: Instant,
|
||||
/// Last Instant we wrote an outbound Opus frame. Used to pace the
|
||||
/// 20 ms encode tick for the echo path (slice-1 read of spec §3.2).
|
||||
pub(crate) last_outbound_at: Instant,
|
||||
/// Outbound RTP media-time clock. For Opus audio on the wire str0m's
|
||||
/// negotiated clock is 48 kHz; a 20 ms frame advances the RTP
|
||||
/// timestamp by 48 000 × 0.020 = 960 ticks. We increment the
|
||||
/// `MediaTime` by `MediaTime::from(Duration)` per frame, since
|
||||
/// `MediaTime` has no `add(Duration)` method on str0m 0.21 — the
|
||||
/// `From<Duration>` impl interprets the duration against the
|
||||
/// underlying clock frequency.
|
||||
pub(crate) next_media_time: str0m::media::MediaTime,
|
||||
}
|
||||
|
||||
impl RtcSession {
|
||||
/// Construct a new session — used by both the binary's `AppState`
|
||||
/// (production) and the tests. Single constructor — no `for_test` /
|
||||
/// `for_server` split; the body is identical (binding a UDP socket
|
||||
/// on `0.0.0.0:0`, constructing the `Rtc` + codecs).
|
||||
pub fn new() -> Result<Self, RtcSessionError> {
|
||||
Self::new_internal()
|
||||
}
|
||||
|
||||
fn new_internal() -> Result<Self, RtcSessionError> {
|
||||
// Bind an ephemeral UDP socket. We use std::net::UdpSocket and
|
||||
// drive it non-blocking from tokio rather than tokio's UdpSocket:
|
||||
// str0m operates on raw `Receive` values and yields `Transmit`
|
||||
// values, both of which are plain structs — no async needed.
|
||||
// Setting non-blocking lets us `recv_from` without blocking.
|
||||
//
|
||||
// We bind to 127.0.0.1 (not 0.0.0.0) because the local candidate
|
||||
// we hand to str0m uses this socket's address, and str0m's
|
||||
// `Candidate::host` rejects the unspecified address 0.0.0.0 as
|
||||
// invalid. Slice 1 is loopback-only; production binding (when
|
||||
// the binary lands in Task 5) can rebind to a real interface
|
||||
// address before `accept_offer` constructs the candidate.
|
||||
let socket = std::net::UdpSocket::bind("127.0.0.1:0")?;
|
||||
socket.set_nonblocking(true)?;
|
||||
let local_addr = socket.local_addr()?;
|
||||
|
||||
let rtc = Rtc::new(Instant::now());
|
||||
|
||||
Ok(Self {
|
||||
channel: Channel::new_inbound(),
|
||||
rtc,
|
||||
decoder: OpusDecoder::new()?,
|
||||
encoder: OpusEncoder::new()?,
|
||||
pipe: EchoAudioPipe::new(),
|
||||
socket,
|
||||
local_addr,
|
||||
audio_mid: None,
|
||||
next_timeout: None,
|
||||
last_rx: Instant::now(),
|
||||
last_outbound_at: Instant::now(),
|
||||
next_media_time: str0m::media::MediaTime::ZERO,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn channel_id(&self) -> ChannelId {
|
||||
self.channel.id
|
||||
}
|
||||
|
||||
pub fn channel_state(&self) -> ChannelState {
|
||||
self.channel.state
|
||||
}
|
||||
|
||||
pub fn is_closed(&self) -> bool {
|
||||
matches!(self.channel.state, ChannelState::Closed)
|
||||
}
|
||||
|
||||
/// Accept a browser SDP offer; return the SDP answer (spec §4.1).
|
||||
///
|
||||
/// str0m 0.21's `sdp_api().accept_offer()` does the heavy lifting:
|
||||
/// parses the offer, picks compatible codecs (Opus, the only one we
|
||||
/// register by default), generates the DTLS fingerprint from its
|
||||
/// self-signed cert, and produces ICE ufrag/pwd. We add our local
|
||||
/// host candidate (the UDP socket we just bound) *before* calling
|
||||
/// `accept_offer` so the answer carries it.
|
||||
pub fn accept_offer(&mut self, offer_sdp: &str) -> Result<String, RtcSessionError> {
|
||||
assert!(self.audio_mid.is_none(), "accept_offer called twice");
|
||||
|
||||
// Register our local UDP socket as a host candidate. str0m includes
|
||||
// this candidate's address + the ICE creds it generates in the SDP
|
||||
// answer. `add_local_candidate` returns `Option<&Candidate>` —
|
||||
// `None` means str0m rejected it (log + continue; not fatal).
|
||||
let candidate = str0m::Candidate::host(self.local_addr, "udp")
|
||||
.expect("host candidate from bound UDP socket");
|
||||
// ^-- expect is acceptable here: this is construction (cold path),
|
||||
// not the hot path. A bound UDP socket always yields a valid
|
||||
// host candidate; only an absurd Protocol parse fails.
|
||||
if self.rtc.add_local_candidate(candidate).is_none() {
|
||||
tracing::warn!(channel_id = %self.channel.id, "str0m rejected local candidate");
|
||||
}
|
||||
|
||||
// str0m's SDP API parses + accepts the offer natively. There is NO
|
||||
// `from_str_unchecked` — `SdpOffer::from_sdp_string` returns
|
||||
// `Result` and is the canonical entry point. `accept_offer` takes
|
||||
// the owned `SdpOffer` and returns the `SdpAnswer`.
|
||||
let parsed_offer = SdpOffer::from_sdp_string(offer_sdp)
|
||||
.map_err(|e| RtcSessionError::SdpOffer(format!("parse: {e}")))?;
|
||||
let answer = self
|
||||
.rtc
|
||||
.sdp_api()
|
||||
.accept_offer(parsed_offer)
|
||||
.map_err(|e| RtcSessionError::SdpOffer(format!("accept: {e}")))?;
|
||||
|
||||
// The first audio mid we accepted. Used to get the Writer for
|
||||
// outbound Opus frames in `run_poll_once`. A single audio m-line
|
||||
// is slice 1's whole world; multi-m-line arrives with video.
|
||||
//
|
||||
// `SdpAnswer` has no `mid()` accessor in str0m 0.21 (the brief's
|
||||
// sketch assumed it). It derefs to `Sdp`, whose `media_lines`
|
||||
// is `Vec<MediaLine>`; each `MediaLine::mid()` returns the mid.
|
||||
// Slice 1's answer has exactly one m-line; we read its mid.
|
||||
self.audio_mid = answer.media_lines.first().map(|m| m.mid());
|
||||
|
||||
self.channel.state = ChannelState::Connecting;
|
||||
Ok(answer.to_sdp_string())
|
||||
}
|
||||
|
||||
/// Drive one iteration of the sans-IO poll loop (spec §3.2, §3.4).
|
||||
///
|
||||
/// Returns the `Duration` until the next `Input::Timeout` should be
|
||||
/// fed back to str0m, or `None` if the peer is closed. The caller
|
||||
/// (Task 5's tokio task) sleeps this duration then calls again.
|
||||
///
|
||||
/// DEV-DEVIATION: tokio polling accepted for slice 1; step 4
|
||||
/// replaces with dedicated timing thread per ARCHITECTURE.md.
|
||||
pub fn run_poll_once(&mut self, now: Instant) -> Option<Duration> {
|
||||
if self.is_closed() {
|
||||
return None;
|
||||
}
|
||||
crate::loop_driver::drive(self, now)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// A captured Chrome SDP offer for an audio-only Opus m-line. Real
|
||||
/// browser-style offer with host ICE candidates — the simplest valid
|
||||
/// offer str0m 0.21 will accept and produce an answer for. Slice 1 has
|
||||
/// no video m-line; multi-m-line arrives with the escalation rung.
|
||||
///
|
||||
/// NOTE: the brief's fixture omitted the `a=group:BUNDLE 0` line
|
||||
/// (str0m 0.21 rejects SDP without a session-level group attribute as
|
||||
/// inconsistent — `sdp::Sdp::assert_consistency`). Real Chrome offers
|
||||
/// always include BUNDLE; restored here so the offer str0m receives
|
||||
/// matches what a browser actually sends.
|
||||
const BROWSER_SDP_OFFER: &str = "\
|
||||
v=0\r
|
||||
o=- 4593482934 2 IN IP4 127.0.0.1\r
|
||||
s=-\r
|
||||
t=0 0\r
|
||||
a=group:BUNDLE 0\r
|
||||
m=audio 9 UDP/TLS/RTP/SAVPF 111\r
|
||||
c=IN IP4 0.0.0.0\r
|
||||
a=rtcp:9 IN IP4 0.0.0.0\r
|
||||
a=ice-ufrag:abcd\r
|
||||
a=ice-pwd:abcdefghijklmnopqrstuvwxyz0123456789\r
|
||||
a=fingerprint:sha-256 AB:CD:EF:00:11:22:33:44:55:66:77:88:99:AA:BB:CC:DD:EE:FF:00:11:22:33:44:55:66:77:88:99:AA:BB:CC:DD\r
|
||||
a=setup:actpass\r
|
||||
a=mid:0\r
|
||||
a=sendrecv\r
|
||||
a=rtpmap:111 opus/48000/2\r
|
||||
a=fmtp:111 minptime=10;useinbandfec=1\r
|
||||
a=candidate:1 1 UDP 2113667327 192.168.1.42 50000 typ host\r
|
||||
";
|
||||
|
||||
#[test]
|
||||
fn accept_offer_returns_sdp_answer_with_opus() {
|
||||
let mut session = RtcSession::new().expect("session");
|
||||
let answer = session.accept_offer(BROWSER_SDP_OFFER).expect("SDP answer");
|
||||
// Answer contains an audio m-line, an Opus payload, a fingerprint,
|
||||
// and ICE credentials (str0m fills these natively in 0.21).
|
||||
assert!(answer.contains("m=audio"), "answer has an audio m-line");
|
||||
assert!(answer.contains("opus/48000"), "answer advertises Opus");
|
||||
assert!(
|
||||
answer.contains("a=fingerprint:sha-256 "),
|
||||
"DTLS fingerprint"
|
||||
);
|
||||
assert!(answer.contains("a=ice-ufrag:"), "ICE ufrag present");
|
||||
assert!(answer.contains("a=ice-pwd:"), "ICE pwd present");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn channel_id_matches_session_id() {
|
||||
let session = RtcSession::new().expect("session");
|
||||
let id = session.channel_id();
|
||||
// The ChannelId IS the session id surfaced in the REST API (spec §4.5).
|
||||
assert_eq!(format!("{}", id).len(), 36);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accept_offer_transitions_channel_to_connecting() {
|
||||
// The spec §5.4 state machine: New → Connecting on offer receive.
|
||||
// This test pins the transition callers depend on; the impl sets
|
||||
// it at the end of `accept_offer`.
|
||||
let mut session = RtcSession::new().expect("session");
|
||||
assert_eq!(session.channel_state(), ChannelState::New);
|
||||
let _ = session.accept_offer(BROWSER_SDP_OFFER).expect("answer");
|
||||
assert_eq!(session.channel_state(), ChannelState::Connecting);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user