diff --git a/tests/transport_loopback.rs b/tests/transport_loopback.rs new file mode 100644 index 0000000..94d2b15 --- /dev/null +++ b/tests/transport_loopback.rs @@ -0,0 +1,144 @@ +//! End-to-end loopback test for the redesigned audio transport. +//! +//! Spins up two real iroh endpoints on localhost (relay disabled, addresses +//! exchanged directly) and drives the actual production path: reactive +//! `connect_peer`, sequenced `broadcast`, `receive_datagrams`, and the +//! per-peer `JitterBuffer` decode. No microphone, speakers, or GUI required. + +use std::sync::Arc; +use std::time::Duration; + +use bytes::Bytes; +use iroh::address_lookup::memory::MemoryLookup; +use iroh::endpoint::presets; +use iroh::protocol::Router; +use iroh::{Endpoint, RelayMode}; +use opus::{Application, Channels}; + +use peerspeak::codec::AudioEncoder; +use peerspeak::codec::opus_impl::OpusEncoder; +use peerspeak::core::jitter::{FRAME_SAMPLES, JitterBuffer}; +use peerspeak::network::NetworkTransport; +use peerspeak::network::iroh_impl::IrohTransport; + +const AUDIO_ALPN: &[u8] = b"peerspeak-audio"; + +struct Node { + endpoint: Endpoint, + transport: Arc, + _router: Router, + lookup: MemoryLookup, +} + +async fn spawn_node() -> Node { + let lookup = MemoryLookup::new(); + let endpoint = Endpoint::builder(presets::Minimal) + .secret_key(iroh::SecretKey::generate()) + // Direct-only: two endpoints on the same host reach each other via the + // addresses we seed into each other's lookup, with no external relay. + .relay_mode(RelayMode::Disabled) + .address_lookup(lookup.clone()) + .bind() + .await + .expect("bind endpoint"); + + let (transport, audio_proto) = IrohTransport::new(endpoint.clone()); + let router = Router::builder(endpoint.clone()) + .accept(AUDIO_ALPN, audio_proto) + .spawn(); + + Node { + endpoint, + transport: Arc::new(transport), + _router: router, + lookup, + } +} + +/// One real, decodable Opus packet for a 20ms mono frame, prefixed with the +/// 4-byte little-endian sequence header the transport/jitter buffer expect. +fn packet(enc: &mut OpusEncoder, seq: u32) -> Bytes { + let pcm: Vec = (0..FRAME_SAMPLES) + .map(|i| if i % 2 == 0 { 2000 } else { -2000 }) + .collect(); + let encoded = enc.encode(&pcm).unwrap(); + let mut buf = Vec::with_capacity(4 + encoded.len()); + buf.extend_from_slice(&seq.to_le_bytes()); + buf.extend_from_slice(&encoded); + Bytes::from(buf) +} + +#[tokio::test] +async fn loopback_sequenced_audio_reaches_peer_and_decodes() { + let a = spawn_node().await; + let b = spawn_node().await; + + // Seed each side with the other's full address so direct dialing works. + a.lookup.add_endpoint_info(b.endpoint.addr()); + b.lookup.add_endpoint_info(a.endpoint.addr()); + + let a_id = a.endpoint.id(); + let b_id = b.endpoint.id(); + + // Subscribe to incoming datagrams on B before any are sent. + let mut b_rx = b.transport.receive_datagrams().await.expect("subscribe B"); + + // Reactive connection setup, exactly as core does on peer-join. Calling on + // both sides is fine: the lower EndpointId dials, the higher accepts, and a + // single shared connection forms. + a.transport.connect_peer(b_id).await; + b.transport.connect_peer(a_id).await; + + // Let the dial + accept registration settle. + tokio::time::sleep(Duration::from_millis(500)).await; + + // A sends 50 sequenced frames. + const N: u32 = 50; + let mut enc = OpusEncoder::new(48000, Channels::Mono, Application::Voip).unwrap(); + for seq in 0..N { + a.transport.broadcast(packet(&mut enc, seq)); + tokio::time::sleep(Duration::from_millis(5)).await; + } + + // Collect what B receives and feed it through a real jitter buffer, + // popping a frame per arrival to mirror the mixer's steady 20ms cadence + // (so the buffer stays shallow rather than overflowing its cap). + let mut jitter = JitterBuffer::new().unwrap(); + let mut received = 0u32; + let mut decoded_frames = 0u32; + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + // Loop ends when the channel closes or the deadline is hit (pattern stops matching). + while let Ok(Some((from, bytes))) = tokio::time::timeout_at(deadline, b_rx.recv()).await { + assert_eq!(from, a_id, "datagram should be attributed to sender A"); + assert!(bytes.len() >= 4, "packet carries a sequence header"); + let seq = u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]); + jitter.insert(seq, bytes[4..].to_vec()); + received += 1; + + if let Some(frame) = jitter.pop_frame() { + assert_eq!(frame.len(), FRAME_SAMPLES, "decoded frame is one 20ms frame"); + decoded_frames += 1; + } + if received >= N { + break; + } + } + + // Drain whatever remains buffered behind the playout delay. + while let Some(frame) = jitter.pop_frame() { + assert_eq!(frame.len(), FRAME_SAMPLES); + decoded_frames += 1; + } + + // On localhost essentially nothing should be lost over a real QUIC datagram path. + assert!( + received >= N - 2, + "expected to receive ~{N} datagrams, got {received}" + ); + // And nearly all received packets should decode to PCM (a few absorbed by + // the initial priming delay). + assert!( + decoded_frames >= N - 5, + "expected to decode ~{N} frames, got {decoded_frames} (received {received})" + ); +}