74d0cb42eb
cargo fmt --check (CI gate per AGENTS.md) flagged import-ordering + line-wrapping drift across slice-2's committed code (rutster-tap, rutster-tap-echo, rutster binary) that pre-dates slice-3. This worktree branched from pivot-and-ui-design (which branched from main before slice-1-review-fixes' fmt cleanup commit landed), so the drift is real + would break CI on the slice-3 branch. Purely cosmetic: alphabetical use ordering, line wrapping under 100 cols. No semantic changes. Mirrors the same fmt cleanup that slice-1-review-fixes applied.
258 lines
12 KiB
Rust
258 lines
12 KiB
Rust
//! # 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::rtc_session::{IDLE_TIMEOUT, RtcSession};
|
|
|
|
/// 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. */ }
|
|
}
|
|
}
|