From e675531bb0993d212ebbd9ef0532df534e05a0fc Mon Sep 17 00:00:00 2001 From: Blax Software Date: Wed, 22 Jul 2026 10:53:50 +0200 Subject: [PATCH] test(sfu): in-process media-forwarding integration test (rust/tests/forwarding.rs) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the coverage gap the README flagged as "Next": two real str0m clients connect to the sidecar over loopback UDP (full ICE/DTLS/SRTP + the data-channel renegotiation the browser does), peer A sends Opus RTP, and we assert peer B RECEIVES the forwarded media. Pure `cargo test`, ~0.8s, no browser — proves the SFU data plane (accept offer → forward without decode) end to end. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 2 +- rust/tests/forwarding.rs | 217 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 218 insertions(+), 1 deletion(-) create mode 100644 rust/tests/forwarding.rs diff --git a/README.md b/README.md index 0828d5b..209c4c8 100644 --- a/README.md +++ b/README.md @@ -219,7 +219,7 @@ Membership lives in a swappable `RoomStore`. The default `ArrayRoomStore` keeps ## Status -**Working package.** Browser-to-browser (mesh) calls, typed rooms (public / private / presence / open-presence) with authorization + presence, full per-participant **call recording**, and **backend-hosted rooms on the Rust SFU sidecar** (server-terminated ICE/DTLS/SRTP, forwarding without decode, native Ogg/Opus recording, mute/kick/stats/usage events, plug-and-play binary management) — all tested, including a real PHP↔Rust round-trip that spawns the sidecar from PHPUnit. Next: the duplex AI bridge inside the SFU data plane (today `OpenAiRealtimeBridge` covers turn-based AI participants) and browser E2E automation. +**Working package.** Browser-to-browser (mesh) calls, typed rooms (public / private / presence / open-presence) with authorization + presence, full per-participant **call recording**, and **backend-hosted rooms on the Rust SFU sidecar** (server-terminated ICE/DTLS/SRTP, forwarding without decode, native Ogg/Opus recording, mute/kick/stats/usage events, plug-and-play binary management) — all tested, including a real PHP↔Rust round-trip that spawns the sidecar from PHPUnit and an **in-process media-forwarding test** (`rust/tests/forwarding.rs`: two real str0m clients over loopback UDP, one sends Opus, the other receives the forwarded media — full ICE/DTLS/SRTP + data-channel renegotiation, no browser). Next: the duplex AI bridge inside the SFU data plane (today `OpenAiRealtimeBridge` covers turn-based AI participants). ## Architecture diff --git a/rust/tests/forwarding.rs b/rust/tests/forwarding.rs new file mode 100644 index 0000000..ba89d23 --- /dev/null +++ b/rust/tests/forwarding.rs @@ -0,0 +1,217 @@ +//! Media-forwarding integration test — the coverage the README flagged as "Next". +//! +//! Boots the real sidecar and connects TWO real str0m clients to it over loopback +//! UDP (full ICE/DTLS/SRTP, and the data-channel renegotiation the browser does). +//! Peer A sends Opus RTP; we assert peer B RECEIVES the forwarded media. No +//! browser — the "clients" are str0m `Rtc` instances driven over real sockets, so +//! this proves the SFU data plane (accept offer → forward without decode) end to +//! end and runs in plain `cargo test`. + +use std::io::{BufRead, BufReader, Write}; +use std::net::{IpAddr, UdpSocket}; +use std::os::unix::net::UnixStream; +use std::path::PathBuf; +use std::thread; +use std::time::{Duration, Instant}; + +use str0m::change::{SdpAnswer, SdpOffer, SdpPendingOffer}; +use str0m::channel::ChannelId; +use str0m::media::{Direction, MediaKind, MediaTime, Mid}; +use str0m::net::{Protocol, Receive}; +use str0m::{Candidate, Event, Input, Output, Rtc}; + +use blax_webrtc_sfu::server::{self, Config}; + +fn start_sidecar(socket: PathBuf) -> thread::JoinHandle<()> { + let config = Config { + socket, + udp_ip: IpAddr::from([127, 0, 0, 1]), + udp_port: 0, + public_ip: Some(IpAddr::from([127, 0, 0, 1])), + }; + thread::spawn(move || server::run(config).expect("sidecar to run")) +} + +fn control_connect(socket: &PathBuf) -> UnixStream { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Ok(s) = UnixStream::connect(socket) { + return s; + } + assert!(Instant::now() < deadline, "control socket never came up"); + thread::sleep(Duration::from_millis(20)); + } +} + +/// add_peer over the control socket (what PHP does); returns the answer SDP. +fn add_peer( + reader: &mut impl BufRead, + writer: &mut impl Write, + id: u64, + room: &str, + peer: &str, + offer: &str, +) -> String { + let req = serde_json::json!({"id": id, "cmd": "add_peer", "room": room, "peer": peer, "offer": offer}); + writeln!(writer, "{req}").unwrap(); + writer.flush().unwrap(); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + let v: serde_json::Value = serde_json::from_str(&line).unwrap(); + assert_eq!(v["ok"], true, "add_peer failed: {v}"); + v["answer"].as_str().expect("answer sdp").to_string() +} + +struct Peer { + rtc: Rtc, + socket: UdpSocket, + audio_mid: Option, + cid: Option, + connected: bool, + media_recv: usize, +} + +impl Peer { + /// A str0m client offering a data channel (+ an audio track when `with_audio`). + fn new(with_audio: bool) -> (Peer, String, SdpPendingOffer) { + let socket = UdpSocket::bind("127.0.0.1:0").unwrap(); + socket.set_read_timeout(Some(Duration::from_millis(3))).unwrap(); + let addr = socket.local_addr().unwrap(); + + let mut rtc = Rtc::builder().build(); + // No OS candidate gathering in str0m: add our host candidate so the offer + // carries it (the SFU also learns us from inbound STUN). + rtc.add_local_candidate(Candidate::host(addr, "udp").unwrap()); + + let mut change = rtc.sdp_api(); + let audio_mid = + with_audio.then(|| change.add_media(MediaKind::Audio, Direction::SendRecv, None, None)); + change.add_channel("data".to_string()); + let (offer, pending) = change.apply().expect("offer"); + + ( + Peer { rtc, socket, audio_mid, cid: None, connected: false, media_recv: 0 }, + offer.to_sdp_string(), + pending, + ) + } + + fn accept_answer(&mut self, pending: SdpPendingOffer, answer_sdp: &str) { + let answer = SdpAnswer::from_sdp_string(answer_sdp).expect("parse answer"); + self.rtc.sdp_api().accept_answer(pending, answer).expect("accept answer"); + } + + /// Drain all pending output (send Transmits, handle events); return next timeout. + fn poll(&mut self) -> Instant { + loop { + if !self.rtc.is_alive() { + return Instant::now() + Duration::from_millis(50); + } + match self.rtc.poll_output().expect("poll_output") { + Output::Transmit(t) => { + let _ = self.socket.send_to(&t.contents, t.destination); + } + Output::Timeout(t) => return t, + Output::Event(e) => self.on_event(e), + } + } + } + + fn on_event(&mut self, e: Event) { + match e { + Event::Connected => self.connected = true, + Event::ChannelOpen(cid, _) => self.cid = Some(cid), + Event::ChannelData(cd) => self.on_channel_data(&cd.data), + Event::MediaData(_) => self.media_recv += 1, + _ => {} + } + } + + /// Data-channel renegotiation: the SFU offers a peer's track over the DC as + /// others join; accept it and answer on the same channel (mirrors the browser). + fn on_channel_data(&mut self, data: &[u8]) { + let Ok(offer) = serde_json::from_slice::(data) else { return }; + let Ok(answer) = self.rtc.sdp_api().accept_offer(offer) else { return }; + if let Some(cid) = self.cid { + if let Some(mut ch) = self.rtc.channel(cid) { + let json = serde_json::to_string(&answer).unwrap(); + let _ = ch.write(false, json.as_bytes()); + } + } + } + + fn recv(&mut self) { + let mut buf = vec![0u8; 2000]; + loop { + match self.socket.recv_from(&mut buf) { + Ok((n, source)) => { + let dest = self.socket.local_addr().unwrap(); + if let Ok(contents) = (&buf[..n]).try_into() { + let _ = self.rtc.handle_input(Input::Receive( + Instant::now(), + Receive { proto: Protocol::Udp, source, destination: dest, contents }, + )); + } + } + Err(_) => break, // WouldBlock / read timeout + } + } + let _ = self.rtc.handle_input(Input::Timeout(Instant::now())); + } + + /// Write one Opus packet on the audio track (payload is opaque — the SFU + /// forwards without decoding, so B receives these bytes verbatim). + fn write_audio(&mut self, seq: u64) { + let Some(mid) = self.audio_mid else { return }; + let Some(w) = self.rtc.writer(mid) else { return }; + let pt = w.payload_params().find(|p| p.spec().codec.is_audio()).map(|p| p.pt()); + if let Some(pt) = pt { + let _ = w.write(pt, Instant::now(), MediaTime::from_micros(seq * 20_000), vec![0xA5u8; 80]); + } + } +} + +#[test] +fn two_clients_forward_audio() { + let socket = std::env::temp_dir().join(format!("blax-sfu-fwd-{}.sock", std::process::id())); + let _ = std::fs::remove_file(&socket); + let _sidecar = start_sidecar(socket.clone()); + + let ctrl = control_connect(&socket); + let mut reader = BufReader::new(ctrl.try_clone().unwrap()); + let mut writer = ctrl; + + // A sends audio; B (data channel only) receives A's track via renegotiation. + let (mut a, a_offer, a_pending) = Peer::new(true); + let a_answer = add_peer(&mut reader, &mut writer, 1, "room", "alice", &a_offer); + a.accept_answer(a_pending, &a_answer); + + let (mut b, b_offer, b_pending) = Peer::new(false); + let b_answer = add_peer(&mut reader, &mut writer, 2, "room", "bob", &b_offer); + b.accept_answer(b_pending, &b_answer); + + let deadline = Instant::now() + Duration::from_secs(20); + let mut seq = 0u64; + while Instant::now() < deadline { + a.poll(); + a.recv(); + b.poll(); + b.recv(); + if a.connected { + a.write_audio(seq); + seq += 1; + } + if b.media_recv >= 5 { + break; + } + thread::sleep(Duration::from_millis(3)); + } + + assert!(a.connected, "peer A never reached Connected"); + assert!(b.connected, "peer B never reached Connected"); + assert!( + b.media_recv >= 5, + "peer B did not receive forwarded media (received {} packets)", + b.media_recv + ); +}