diff --git a/Cargo.toml b/Cargo.toml index b7c2204..a7931de 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -44,6 +44,11 @@ crossbeam = "0.8" parking_lot = "0.12" typed-builder = "0.23.2" +# mDNS advertising for the server role (src/server/discovery.rs) — was +# already a dev-dependency for the client-side examples; the server role +# needs it as a real dependency to advertise `_sendspin-server._tcp.local.`. +mdns-sd = "0.20" + [target.'cfg(target_os = "linux")'.dependencies] libc = "0.2" @@ -62,7 +67,6 @@ cpal = { version = "0.18", features = ["realtime"] } tokio-test = "0.4" env_logger = "0.11" clap = { version = "4.6", features = ["derive"] } -mdns-sd = "0.20" [profile.release] opt-level = 3 diff --git a/examples/play_wav.rs b/examples/play_wav.rs new file mode 100644 index 0000000..7399ba5 --- /dev/null +++ b/examples/play_wav.rs @@ -0,0 +1,401 @@ +// ABOUTME: Example server that streams a WAV file to connected Sendspin devices +// ABOUTME: Handles both dial-in clients and clients discovered/dialed over mDNS +// +// Usage: +// cargo run --example play_wav -- path/to/clip.wav +// cargo run --example play_wav -- --dial ws://192.168.1.42:8928/sendspin clip.wav +// cargo run --example play_wav -- --bind 0.0.0.0:8927 --no-advertise clip.wav +// +// Streams the file to every connected client as one synchronized group, and +// prints the client/state, client/command, and client/goodbye messages it +// receives. Handles both connection directions: clients that dial in +// (accepted via ServerListener, discoverable through this tool's mDNS +// advertisement) and clients that only run their own embedded server +// (discovered and dialed via ClientManager). `--dial` targets specific +// clients and disables discovery unless `--also-discover` is given. +// +// Only plain PCM WAV (fmt tag 1) is supported. Re-encode anything else first, +// e.g.: ffmpeg -i in.mp3 -ar 48000 -ac 2 -acodec pcm_s16le out.wav + +use clap::Parser; +use sendspin::protocol::messages::Message; +use sendspin::server::{dial_client, Advertisement, ClientEvent, ClientManager, Group}; +use sendspin::{DefaultClock, ServerConnection, ServerListener}; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; + +#[derive(Parser, Debug)] +#[command(name = "play_wav")] +struct Args { + /// WAV file to stream (plain PCM, any sample rate/channels/bit depth) + file: PathBuf, + + /// Address to bind the server on, for clients that dial in + #[arg(long, default_value = "0.0.0.0:8927")] + bind: String, + + /// HTTP path clients connect to (fixed by the spec for real deployments) + #[arg(long, default_value = "/sendspin")] + path: String, + + /// Server identifier advertised to clients and over mDNS + #[arg(long, default_value = "sendspin-rs-play-wav")] + server_id: String, + + /// Human-readable server name + #[arg(long, default_value = "Sendspin Rust Test Server")] + name: String, + + /// Advertise `_sendspin-server._tcp.local.` so clients that dial in can + /// find this server + #[arg(long, default_value_t = true)] + advertise: bool, + + /// Disable mDNS advertising (e.g. to avoid a device picking up this test + /// server instead of your production one on the same network) + #[arg(long)] + no_advertise: bool, + + /// Disable mDNS client discovery. Discovery (of `_sendspin._tcp.local.` + /// clients — devices that only run their own embedded server, e.g. Home + /// Assistant Voice PE) is on by default only when no --dial is given; + /// see --also-discover to keep it on alongside --dial. + #[arg(long)] + no_discover: bool, + + /// Keep mDNS discovery on even when --dial is given. By default, giving + /// --dial turns discovery off so you can isolate a test to just the + /// device(s) you named — otherwise every other Sendspin device on the + /// network joins the group too. + #[arg(long)] + also_discover: bool, + + /// Dial a specific client URL directly (e.g. + /// ws://192.168.1.42:8928/sendspin). Repeatable — pass it more than once + /// to test a specific, known set of devices. + #[arg(long = "dial")] + dial_urls: Vec, + + /// After the first client connects, wait this long for additional + /// clients before starting playback (0 to start immediately) + #[arg(long, default_value_t = 5)] + wait_secs: u64, + + /// Milliseconds of audio per pushed chunk + #[arg(long, default_value_t = 100)] + chunk_ms: u64, + + /// How far ahead of "now" each audio chunk's timestamp is scheduled, in + /// milliseconds — must comfortably exceed real delivery jitter or + /// clients will receive chunks whose intended playback time already + /// passed. Passed through to Group::with_send_ahead_us. + #[arg(long, default_value_t = 250)] + send_ahead_ms: u64, +} + +struct WavInfo { + channels: u8, + sample_rate: u32, + bit_depth: u8, + data: Vec, +} + +/// Minimal RIFF/WAVE chunk walker — just enough to find `fmt ` and `data` +/// for plain PCM. Not a general-purpose WAV reader (no WAVE_FORMAT_EXTENSIBLE, +/// no metadata chunks surfaced) — this is a manual test tool, not a decoder +/// this crate needs to ship. +fn read_wav(path: &PathBuf) -> Result { + let bytes = std::fs::read(path).map_err(|e| format!("reading {}: {e}", path.display()))?; + if bytes.len() < 12 || &bytes[0..4] != b"RIFF" || &bytes[8..12] != b"WAVE" { + return Err(format!("{} is not a RIFF/WAVE file", path.display())); + } + + let mut pos = 12; + let mut fmt: Option<(u16, u8, u32, u8)> = None; // (format_tag, channels, sample_rate, bit_depth) + let mut data: Option> = None; + + while pos + 8 <= bytes.len() { + let chunk_id = &bytes[pos..pos + 4]; + let chunk_size = u32::from_le_bytes(bytes[pos + 4..pos + 8].try_into().unwrap()) as usize; + let body_start = pos + 8; + let body_end = (body_start + chunk_size).min(bytes.len()); + let body = &bytes[body_start..body_end]; + + match chunk_id { + b"fmt " if body.len() >= 16 => { + let format_tag = u16::from_le_bytes(body[0..2].try_into().unwrap()); + let channels = u16::from_le_bytes(body[2..4].try_into().unwrap()) as u8; + let sample_rate = u32::from_le_bytes(body[4..8].try_into().unwrap()); + let bit_depth = u16::from_le_bytes(body[14..16].try_into().unwrap()) as u8; + fmt = Some((format_tag, channels, sample_rate, bit_depth)); + } + b"data" => data = Some(body.to_vec()), + _ => {} + } + + // Chunks are padded to even length. + pos = body_start + chunk_size + (chunk_size % 2); + } + + let (format_tag, channels, sample_rate, bit_depth) = fmt.ok_or("no fmt chunk found")?; + if format_tag != 1 { + return Err(format!( + "unsupported WAV format tag {format_tag} (only plain PCM, tag 1, is supported) — \ + re-encode with: ffmpeg -i {} -acodec pcm_s16le out.wav", + path.display() + )); + } + let data = data.ok_or("no data chunk found")?; + + Ok(WavInfo { + channels, + sample_rate, + bit_depth, + data, + }) +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + env_logger::init(); + let args = Args::parse(); + let advertise = args.advertise && !args.no_advertise; + let discover = !args.no_discover && (args.dial_urls.is_empty() || args.also_discover); + + let wav = read_wav(&args.file).map_err(|e| format!("failed to read WAV: {e}"))?; + println!( + "loaded {}: {} Hz, {} ch, {}-bit, {:.1}s", + args.file.display(), + wav.sample_rate, + wav.channels, + wav.bit_depth, + wav.data.len() as f64 + / (wav.sample_rate as f64 * wav.channels as f64 * (wav.bit_depth as f64 / 8.0)) + ); + + let listener = ServerListener::bind(&args.bind, &args.server_id, &args.name) + .await? + .path(&args.path); + let port = listener.local_addr()?.port(); + println!( + "listening on {} (path {}), for clients that dial in", + listener.local_addr()?, + args.path + ); + + let _advertisement = if advertise { + println!( + "advertising _sendspin-server._tcp.local. as {:?} on port {port}", + args.server_id + ); + Some(Advertisement::new( + &args.server_id, + &args.name, + port, + &args.path, + )?) + } else { + None + }; + + let group = Arc::new( + Group::new(Arc::new(DefaultClock::default())) + .with_send_ahead_us(args.send_ahead_ms as i64 * 1000), + ); + + spawn_accept_loop(listener, Arc::clone(&group)); + // Kept alive for main()'s whole lifetime — dropping it would stop + // discovery and abort every reconnect loop it's supervising. + let _manager = if discover { + Some(spawn_manager_loop( + Arc::clone(&group), + args.server_id.clone(), + args.name.clone(), + )?) + } else { + None + }; + for url in &args.dial_urls { + dial_one(&group, url, &args.server_id, &args.name).await; + } + + println!("waiting for at least one client (inbound accept, mDNS discovery, or --dial)..."); + wait_for_first_member(&group).await; + + if args.wait_secs > 0 { + println!( + "waiting up to {}s for additional clients (connect more devices now for a multi-room test)...", + args.wait_secs + ); + tokio::time::sleep(Duration::from_secs(args.wait_secs)).await; + } + + let member_count = group.len(); + println!("starting playback to {member_count} client(s)"); + + let chunk_bytes = (wav.sample_rate as u64 * args.chunk_ms / 1000) as usize + * wav.channels as usize + * (wav.bit_depth as usize / 8); + if chunk_bytes == 0 { + return Err("computed chunk size is zero — check --chunk-ms and the WAV format".into()); + } + + group + .start_stream(sendspin::protocol::messages::StreamPlayerConfig { + codec: "pcm".to_string(), + sample_rate: wav.sample_rate, + channels: wav.channels, + bit_depth: wav.bit_depth, + codec_header: None, + }) + .await; + + // push_audio enqueues without blocking and the Group anchors its own + // timeline, so playback timing no longer depends on the exact push cadence + // — a plain per-chunk sleep is enough to pace roughly real-time and keep + // each member's send queue shallow. Nothing here holds a lock on the group, + // so clients arriving mid-stream (via the accept or discovery loops) are + // added concurrently and join the playback in progress. + let total_chunks = wav.data.len().div_ceil(chunk_bytes); + for (i, chunk) in wav.data.chunks(chunk_bytes).enumerate() { + let timestamp_us = group.push_audio(chunk); + if i % 10 == 0 || i + 1 == total_chunks { + println!( + "chunk {}/{total_chunks} (timestamp {timestamp_us}us)", + i + 1 + ); + } + tokio::time::sleep(Duration::from_millis(args.chunk_ms)).await; + } + + println!("done sending audio, ending stream"); + group.end_stream().await; + + // Give the last chunks time to actually finish playing before hanging up. + tokio::time::sleep(Duration::from_secs(2)).await; + println!("done — Ctrl+C to exit (connections stay open so you can inspect further)"); + tokio::signal::ctrl_c().await.ok(); + Ok(()) +} + +async fn wait_for_first_member(group: &Arc) { + loop { + if !group.is_empty() { + return; + } + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +/// Accept clients that dial in to us. +fn spawn_accept_loop(listener: ServerListener, group: Arc) { + tokio::spawn(async move { + loop { + match listener.accept().await { + Ok((conn, addr)) => { + println!( + "[{addr}] client connected (inbound): id={} name={:?} roles={:?}", + conn.client_id(), + conn.hello().name, + conn.active_roles() + ); + add_and_drain(&group, conn).await; + } + Err(e) => { + println!("accept error: {e}"); + tokio::time::sleep(Duration::from_millis(500)).await; + } + } + } + }); +} + +/// Discover clients that only run their own embedded server, dial them, and +/// keep them connected via ClientManager (which retries with backoff on +/// disconnect and re-dials if a device reappears at a new address). +fn spawn_manager_loop( + group: Arc, + server_id: String, + name: String, +) -> Result> { + let (manager, mut events) = + ClientManager::start(server_id, name, Arc::new(DefaultClock::default()))?; + println!("discovering Sendspin clients via mDNS (_sendspin._tcp.local.)..."); + tokio::spawn(async move { + while let Some(event) = events.recv().await { + match event { + ClientEvent::Connected { + client_id, + fullname, + active_roles, + sender, + } => { + println!( + "[{client_id}] connected (dialed via {fullname}): roles={active_roles:?}" + ); + if let Err(e) = group.add_member(client_id, sender).await { + println!("failed to add member to group: {e}"); + } + } + ClientEvent::Message { client_id, message } => match *message { + Message::ClientGoodbye(g) => { + println!("[{client_id}] client/goodbye: {:?}", g.reason); + } + other => println!("[{client_id}] <- {other:?}"), + }, + ClientEvent::Disconnected { client_id } => { + println!("[{client_id}] connection closed, will retry in the background"); + group.remove_member(&client_id); + } + } + } + }); + Ok(manager) +} + +/// Dial one specific URL given via `--dial`, once, at startup. +async fn dial_one(group: &Arc, url: &str, server_id: &str, name: &str) { + dial_and_add(group, url, server_id, name).await; +} + +async fn dial_and_add(group: &Arc, url: &str, server_id: &str, name: &str) { + match dial_client(url, server_id, name, Arc::new(DefaultClock::default())).await { + Ok(conn) => { + println!( + "[{url}] client connected (dialed): id={} name={:?} roles={:?}", + conn.client_id(), + conn.hello().name, + conn.active_roles() + ); + add_and_drain(group, conn).await; + } + Err(e) => println!("[{url}] dial failed: {e}"), + } +} + +async fn add_and_drain(group: &Arc, conn: ServerConnection) { + let client_id = conn.client_id().to_string(); + let sender = conn.sender(); + spawn_message_drain(client_id.clone(), conn); + if let Err(e) = group.add_member(client_id, sender).await { + println!("failed to add member to group: {e}"); + } +} + +/// Print every client/state, client/command, and client/goodbye a connection +/// sends for as long as it lasts — this is what you want on screen when a +/// real device does something unexpected. +fn spawn_message_drain(client_id: String, mut conn: ServerConnection) { + tokio::spawn(async move { + while let Some(msg) = conn.recv_message().await { + match &msg { + Message::ClientGoodbye(g) => { + println!("[{client_id}] client/goodbye: {:?}", g.reason); + } + other => println!("[{client_id}] <- {other:?}"), + } + } + println!("[{client_id}] connection closed"); + }); +} diff --git a/src/lib.rs b/src/lib.rs index 6fbd271..8cb71db 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -14,6 +14,10 @@ pub mod audio; /// Protocol implementation for WebSocket communication pub mod protocol; +/// Server-role implementation: accept inbound player connections, sync +/// clocks, stream audio. Prototype scope — see the module docs for what's +/// deferred. +pub mod server; /// Clock synchronization utilities pub mod sync; @@ -24,6 +28,7 @@ pub use protocol::client::{Connection, ConnectionGuard, Controller, ProtocolClie pub use protocol::client_builder::ProtocolClientBuilder; pub use protocol::listener::ProtocolListener; pub use protocol::messages::ServerHello; +pub use server::{ServerConnection, ServerListener}; pub use sync::raw_clock::{Clock, DefaultClock}; /// Result type for sendspin operations diff --git a/src/server/binary.rs b/src/server/binary.rs new file mode 100644 index 0000000..a17d636 --- /dev/null +++ b/src/server/binary.rs @@ -0,0 +1,44 @@ +// ABOUTME: Binary audio frame encoding for the server role (the encode-side +// ABOUTME: mirror of protocol::client::AudioChunk::from_bytes, which only parses). + +use crate::protocol::client::binary_types; + +/// Build a binary WebSocket frame carrying one player audio chunk: +/// `[type_id: u8][timestamp: i64 big-endian, µs][payload]`. +/// +/// A client never sends audio, so [`crate::protocol::client::AudioChunk`] only +/// ever parses this layout — the server role needs the builder that side never +/// required. `timestamp_us` is the intended playback time in the server's +/// clock domain; the receiving client converts it to its own clock domain +/// using the offset/drift it tracks from `server/time` replies. +pub fn encode_audio_frame(timestamp_us: i64, payload: &[u8]) -> Vec { + let mut frame = Vec::with_capacity(9 + payload.len()); + frame.push(binary_types::PLAYER_AUDIO); + frame.extend_from_slice(×tamp_us.to_be_bytes()); + frame.extend_from_slice(payload); + frame +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::protocol::client::AudioChunk; + + #[test] + fn round_trips_through_the_client_side_parser() { + let payload = [1u8, 2, 3, 4, 5, 255, 0]; + let frame = encode_audio_frame(123_456_789, &payload); + let chunk = AudioChunk::from_bytes(&frame).expect("client parser must accept our frame"); + assert_eq!(chunk.timestamp, 123_456_789); + assert_eq!(&*chunk.data, &payload[..]); + } + + #[test] + fn negative_timestamps_round_trip() { + // Clock domains are implementation-defined epochs (see sync::raw_clock::Clock), + // so a timestamp can legitimately be negative relative to process start. + let frame = encode_audio_frame(-42, &[9, 9]); + let chunk = AudioChunk::from_bytes(&frame).unwrap(); + assert_eq!(chunk.timestamp, -42); + } +} diff --git a/src/server/connection.rs b/src/server/connection.rs new file mode 100644 index 0000000..89db147 --- /dev/null +++ b/src/server/connection.rs @@ -0,0 +1,512 @@ +// ABOUTME: Per-client connection actor for the server role: drives the +// ABOUTME: server-side handshake, time-sync echo, and message dispatch. + +use crate::error::Error; +use crate::protocol::messages::{ + ClientHello, ConnectionReason, Message, PlayerCommand, ServerCommand, ServerHello, ServerTime, + StreamClear, StreamEnd, StreamPlayerConfig, StreamStart, +}; +use crate::server::binary::encode_audio_frame; +use crate::sync::raw_clock::Clock; +use futures_util::{ + stream::{SplitSink, SplitStream}, + SinkExt, StreamExt, +}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}; +use tokio_tungstenite::{ + tungstenite::{Bytes, Message as WsMessage}, + WebSocketStream, +}; + +/// The only role this server negotiates in v1. See the crate-level server +/// docs for the list of roles deferred for a later contribution +/// (color/visualizer/artwork/controller/metadata). +const PLAYER_ROLE: &str = "player@v1"; + +/// Maximum audio frames a single connection may have queued but not yet +/// written before [`ServerSender::enqueue_audio`] starts dropping frames. This +/// bounds memory for a slow or stalled member so it can't back up the whole +/// process — its own audio suffers, nobody else's does. +const MAX_QUEUED_AUDIO_FRAMES: usize = 32; + +/// Outcome of a non-blocking [`ServerSender::enqueue_audio`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AudioEnqueue { + /// The frame was queued for the writer task. + Sent, + /// The connection's audio backlog was at capacity; the frame was dropped. + Evicted, +} + +enum WriteCommand { + Send { + msg: WsMessage, + ack: tokio::sync::oneshot::Sender>, + }, + /// `server/time` reply: `server_transmitted` is stamped from the clock + /// immediately before the frame reaches the wire, not when this command + /// was enqueued — queueing delay would otherwise leak into the client's + /// clock filter as measurement error (this is why it's its own variant + /// rather than a pre-built `Send`). + TimeReply { + client_transmitted: i64, + server_received: i64, + ack: tokio::sync::oneshot::Sender>, + }, + Close { + ack: tokio::sync::oneshot::Sender>, + }, + /// Fire-and-forget audio frame — no ack, so broadcasting to a group never + /// blocks on any member's socket write. `queued` is decremented once the + /// frame leaves the sink, so the sender can bound the backlog. + Audio { + frame: Bytes, + queued: Arc, + }, +} + +async fn writer_task( + mut sink: SplitSink, WsMessage>, + mut rx: UnboundedReceiver, + clock: Arc, +) where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + while let Some(cmd) = rx.recv().await { + match cmd { + WriteCommand::Send { msg, ack } => { + let result = sink + .send(msg) + .await + .map_err(|e| Error::WebSocket(e.to_string())); + let failed = result.is_err(); + // Ignore SendError: the caller may have dropped its receiver. + let _ = ack.send(result); + if failed { + break; + } + } + WriteCommand::TimeReply { + client_transmitted, + server_received, + ack, + } => { + let reply = Message::ServerTime(ServerTime { + client_transmitted, + server_received, + server_transmitted: clock.now_micros(), + }); + let result = match serde_json::to_string(&reply) { + Ok(json) => sink + .send(WsMessage::Text(json.into())) + .await + .map_err(|e| Error::WebSocket(e.to_string())), + Err(e) => Err(Error::Protocol(e.to_string())), + }; + let failed = result.is_err(); + let _ = ack.send(result); + if failed { + break; + } + } + WriteCommand::Close { ack } => { + let result = sink + .close() + .await + .map_err(|e| Error::WebSocket(e.to_string())); + let _ = ack.send(result); + break; + } + WriteCommand::Audio { frame, queued } => { + let result = sink.send(WsMessage::Binary(frame)).await; + queued.fetch_sub(1, Ordering::Relaxed); + if result.is_err() { + break; + } + } + } + } + log::debug!("Server connection writer task exiting"); +} + +/// Sender half of a server-role connection. Cheap to clone; all clones share +/// the same underlying connection and audio backlog counter. +#[derive(Debug, Clone)] +pub struct ServerSender { + tx: UnboundedSender, + audio_queued: Arc, +} + +impl ServerSender { + /// Enqueue one pre-encoded audio frame without waiting for it to reach the + /// wire — a group broadcast calls this on every member, so it must never + /// block on any one member's socket. `frame` is a [`Bytes`], so fanning the + /// same frame out to N members is N cheap refcount clones, not N copies. + /// + /// Returns [`AudioEnqueue::Evicted`] if this connection's audio backlog is + /// already full (a slow/stalled member), dropping the frame rather than + /// growing memory without bound. `Err` means the writer task is gone (the + /// member is dead) and the caller should stop broadcasting to it. + pub fn enqueue_audio(&self, frame: Bytes) -> Result { + if self.audio_queued.load(Ordering::Relaxed) >= MAX_QUEUED_AUDIO_FRAMES { + return Ok(AudioEnqueue::Evicted); + } + self.audio_queued.fetch_add(1, Ordering::Relaxed); + match self.tx.send(WriteCommand::Audio { + frame, + queued: Arc::clone(&self.audio_queued), + }) { + Ok(()) => Ok(AudioEnqueue::Sent), + Err(_) => { + self.audio_queued.fetch_sub(1, Ordering::Relaxed); + Err(Error::WebSocket("connection closed".to_string())) + } + } + } + + async fn send_message(&self, msg: Message) -> Result<(), Error> { + let json = serde_json::to_string(&msg).map_err(|e| Error::Protocol(e.to_string()))?; + log::debug!("Sending message: {}", json); + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); + self.tx + .send(WriteCommand::Send { + msg: WsMessage::Text(json.into()), + ack: ack_tx, + }) + .map_err(|_| Error::WebSocket("connection closed".to_string()))?; + ack_rx + .await + .map_err(|_| Error::WebSocket("connection closed".to_string()))? + } + + /// Announce the start of a player audio stream. Send this once before + /// the first [`Self::send_audio_chunk`]. + pub async fn send_stream_start(&self, player: StreamPlayerConfig) -> Result<(), Error> { + self.send_message(Message::StreamStart(StreamStart { + player: Some(player), + artwork: None, + visualizer: None, + })) + .await + } + + /// Push one player audio chunk. `timestamp_us` is the intended playback + /// time in this server's clock domain (see [`crate::sync::raw_clock::Clock`]); + /// the client converts it to its own domain using the offset/drift it + /// tracks from `server/time` replies. + pub async fn send_audio_chunk(&self, timestamp_us: i64, payload: &[u8]) -> Result<(), Error> { + let frame = encode_audio_frame(timestamp_us, payload); + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); + self.tx + .send(WriteCommand::Send { + msg: WsMessage::Binary(frame.into()), + ack: ack_tx, + }) + .map_err(|_| Error::WebSocket("connection closed".to_string()))?; + ack_rx + .await + .map_err(|_| Error::WebSocket("connection closed".to_string()))? + } + + /// End the player audio stream. + pub async fn send_stream_end(&self) -> Result<(), Error> { + self.send_message(Message::StreamEnd(StreamEnd { + roles: Some(vec!["player".to_string()]), + })) + .await + } + + /// Ask the client to discard any buffered-but-unplayed audio (e.g. after + /// a seek), without ending the stream. + pub async fn send_stream_clear(&self) -> Result<(), Error> { + self.send_message(Message::StreamClear(StreamClear { + roles: Some(vec!["player".to_string()]), + })) + .await + } + + /// Send a player command (volume, mute, static delay) to the client. + pub async fn send_player_command(&self, command: PlayerCommand) -> Result<(), Error> { + self.send_message(Message::ServerCommand(ServerCommand { + player: Some(command), + })) + .await + } +} + +/// Aborts background tasks on drop. Hold this alive for the lifetime of the +/// connection — mirrors [`crate::protocol::client::ConnectionGuard`]. +pub struct ServerConnectionGuard { + sender: ServerSender, + router_handle: Option>, + writer_handle: Option>, +} + +impl ServerConnectionGuard { + /// Close the connection. Unlike the client role, the server has no + /// `goodbye` message of its own to send — it just closes the socket + /// (optionally after the caller has already sent `stream/end`). + pub async fn disconnect(mut self) -> Result<(), Error> { + let (ack_tx, ack_rx) = tokio::sync::oneshot::channel(); + let close_result = self + .sender + .tx + .send(WriteCommand::Close { ack: ack_tx }) + .map_err(|_| Error::WebSocket("connection closed".to_string())); + let result = match close_result { + Ok(()) => ack_rx + .await + .map_err(|_| Error::WebSocket("connection closed".to_string()))?, + Err(e) => Err(e), + }; + if let Some(h) = self.writer_handle.take() { + let _ = h.await; + } + if let Some(h) = self.router_handle.take() { + h.abort(); + } + result + } +} + +impl Drop for ServerConnectionGuard { + fn drop(&mut self) { + if let Some(h) = self.router_handle.take() { + h.abort(); + } + if let Some(h) = self.writer_handle.take() { + h.abort(); + } + } +} + +/// A single accepted client, past the handshake. Returned by +/// [`crate::server::ServerListener::accept`]. +pub struct ServerConnection { + /// The client's `client/hello` payload — identity, declared capabilities, + /// device info. Kept in full so callers can read `player@v1_support` + /// (supported formats, buffer capacity) before starting a stream. + hello: ClientHello, + /// Roles this server granted this client (currently always `["player@v1"]` + /// if the client declared support for it, else empty). + active_roles: Vec, + /// `client/state`, `client/command`, and `client/goodbye` messages, + /// forwarded as received. `client/time` is consumed internally (time-sync + /// echo) and never forwarded here — same convention as + /// [`crate::protocol::client::Connection::messages`]. + messages: UnboundedReceiver, + sender: ServerSender, + guard: ServerConnectionGuard, +} + +impl ServerConnection { + /// The client's `client/hello` payload. + pub fn hello(&self) -> &ClientHello { + &self.hello + } + + /// Convenience accessor for `hello().client_id`. + pub fn client_id(&self) -> &str { + &self.hello.client_id + } + + /// Roles granted to this client. + pub fn active_roles(&self) -> &[String] { + &self.active_roles + } + + /// A cheap-to-clone sender for pushing stream control and audio messages + /// to this client, usable independently of `&mut self`. + pub fn sender(&self) -> ServerSender { + self.sender.clone() + } + + /// Receive the next `client/state`, `client/command`, or `client/goodbye` + /// message. Returns `None` once the connection has closed. + pub async fn recv_message(&mut self) -> Option { + self.messages.recv().await + } + + /// Close the connection. + pub async fn disconnect(self) -> Result<(), Error> { + self.guard.disconnect().await + } + + /// Drive the server-side handshake and message loop over an + /// already-handshaked WebSocket stream. Shared by + /// [`crate::server::ServerListener::accept`] and tests. + pub(crate) async fn drive( + ws_stream: WebSocketStream, + server_id: &str, + server_name: &str, + connection_reason: ConnectionReason, + clock: Arc, + ) -> Result + where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + { + let (mut write, mut read) = ws_stream.split(); + + log::debug!("Waiting for client/hello..."); + let hello = loop { + let Some(result) = read.next().await else { + return Err(Error::Connection( + "connection closed before client/hello".to_string(), + )); + }; + match result { + Ok(WsMessage::Text(text)) => { + let msg: Message = serde_json::from_str(&text).map_err(|e| { + log::warn!("Failed to parse client message: {} (payload: {})", e, text); + Error::Protocol(e.to_string()) + })?; + match msg { + Message::ClientHello(hello) => { + if hello.version != 1 { + return Err(Error::Protocol(format!( + "unsupported protocol version {} (only 1 is supported)", + hello.version + ))); + } + break hello; + } + other => { + return Err(Error::Protocol(format!( + "expected client/hello, got {:?}", + other + ))) + } + } + } + Ok(WsMessage::Ping(_)) | Ok(WsMessage::Pong(_)) => continue, + Ok(WsMessage::Close(_)) => { + return Err(Error::Connection("client closed connection".to_string())) + } + Ok(_) => continue, + Err(e) => return Err(Error::WebSocket(e.to_string())), + } + }; + log::debug!("Received client/hello: {:?}", hello); + + let active_roles: Vec = if hello.supported_roles.iter().any(|r| r == PLAYER_ROLE) { + vec![PLAYER_ROLE.to_string()] + } else { + Vec::new() + }; + + let server_hello = Message::ServerHello(ServerHello { + server_id: server_id.to_string(), + name: server_name.to_string(), + version: 1, + active_roles: active_roles.clone(), + connection_reason, + }); + let json = + serde_json::to_string(&server_hello).map_err(|e| Error::Protocol(e.to_string()))?; + write + .send(WsMessage::Text(json.into())) + .await + .map_err(|e| Error::WebSocket(e.to_string()))?; + + let (out_tx, out_rx) = unbounded_channel::(); + let (message_tx, message_rx) = unbounded_channel(); + + let writer_handle = tokio::spawn(writer_task(write, out_rx, Arc::clone(&clock))); + + let out_tx_router = out_tx.clone(); + let router_handle = tokio::spawn(async move { + Self::message_router(read, message_tx, out_tx_router, clock).await; + }); + + let audio_queued = Arc::new(AtomicUsize::new(0)); + Ok(Self { + hello, + active_roles, + messages: message_rx, + sender: ServerSender { + tx: out_tx.clone(), + audio_queued: Arc::clone(&audio_queued), + }, + guard: ServerConnectionGuard { + sender: ServerSender { + tx: out_tx, + audio_queued, + }, + router_handle: Some(router_handle), + writer_handle: Some(writer_handle), + }, + }) + } + + async fn message_router( + mut read: SplitStream>, + message_tx: UnboundedSender, + out_tx: UnboundedSender, + clock: Arc, + ) where + S: AsyncRead + AsyncWrite + Unpin + Send + 'static, + { + let mut message_closed = false; + + while let Some(msg) = read.next().await { + match msg { + Ok(WsMessage::Text(text)) => { + // Capture receive time before deserialization so + // `server_received` is as close to true arrival as possible. + let server_received = clock.now_micros(); + match serde_json::from_str::(&text) { + Ok(Message::ClientTime(t)) => { + let (ack_tx, _ack_rx) = tokio::sync::oneshot::channel(); + if out_tx + .send(WriteCommand::TimeReply { + client_transmitted: t.client_transmitted, + server_received, + ack: ack_tx, + }) + .is_err() + { + break; + } + } + Ok(Message::ClientHello(_)) => { + log::warn!("Ignoring unexpected client/hello after handshake"); + } + Ok(msg) => { + log::debug!("Received message: {:?}", msg); + if !message_closed && message_tx.send(msg).is_err() { + log::error!( + "Message receiver dropped — messages will be discarded" + ); + message_closed = true; + } + } + Err(e) => { + log::warn!("Failed to parse message: {} (payload: {})", e, text); + } + } + } + Ok(WsMessage::Binary(_)) => { + // A client never sends binary frames in the current + // protocol (audio/artwork/visualizer are server->client + // only); log and ignore rather than erroring the + // connection over a forward-compatible future frame. + log::warn!("Ignoring unexpected binary frame from client"); + } + Ok(WsMessage::Ping(_)) | Ok(WsMessage::Pong(_)) => {} + Ok(WsMessage::Close(_)) => { + log::info!("Client closed connection"); + break; + } + Err(e) => { + log::warn!("WebSocket error: {}", e); + break; + } + _ => {} + } + } + log::debug!("Message router: WebSocket stream ended"); + } +} diff --git a/src/server/dial.rs b/src/server/dial.rs new file mode 100644 index 0000000..95cc6df --- /dev/null +++ b/src/server/dial.rs @@ -0,0 +1,41 @@ +// ABOUTME: Server-initiated connections to Sendspin clients that only run +// ABOUTME: their own embedded WebSocket server (never dialing out +// ABOUTME: themselves) — discovered via discovery::ClientBrowser, dialed here. + +use crate::error::Error; +use crate::protocol::messages::ConnectionReason; +use crate::server::connection::ServerConnection; +use crate::sync::raw_clock::Clock; +use std::sync::Arc; +use tokio_tungstenite::connect_async; + +/// Dial a Sendspin client's own WebSocket server (e.g. a URL discovered via +/// [`crate::server::ClientBrowser`]) and drive the server-role handshake over +/// the resulting connection. +/// +/// The protocol-level roles are identical regardless of which side initiated +/// the TCP connection — the client still sends `client/hello` first, this +/// still replies `server/hello` — so this is otherwise exactly +/// [`crate::server::ServerListener::accept`]'s handshake, just dialed instead +/// of accepted. `sendspin-rs`'s own `protocol::listener::ProtocolListener` is +/// the client-role mirror of this for the reverse case (a client accepting a +/// server that dials in). +pub async fn dial_client( + url: &str, + server_id: &str, + server_name: &str, + clock: Arc, +) -> Result { + let (ws, _response) = connect_async(url) + .await + .map_err(|e| Error::Connection(format!("dial to {url} failed: {e}")))?; + // The server dialed out to stream to this client, so announce Playback. + ServerConnection::drive( + ws, + server_id, + server_name, + ConnectionReason::Playback, + clock, + ) + .await +} diff --git a/src/server/discovery.rs b/src/server/discovery.rs new file mode 100644 index 0000000..92b42b0 --- /dev/null +++ b/src/server/discovery.rs @@ -0,0 +1,188 @@ +// ABOUTME: mDNS for the server role: advertising this server +// ABOUTME: (`_sendspin-server._tcp.local.`) for clients that dial in, and +// ABOUTME: discovering clients that only run their own embedded server +// ABOUTME: (`_sendspin._tcp.local.`) and need to be dialed instead. + +use crate::error::Error; +use mdns_sd::{Receiver, ResolvedService, ScopedIp, ServiceDaemon, ServiceEvent, ServiceInfo}; +use std::net::IpAddr; + +/// Service type clients browse for to discover a Sendspin server (matches +/// aiosendspin's `server/server.py`; distinct from `_sendspin._tcp.local.`, +/// which is what a *client* advertises for server-initiated connections — +/// see `examples/server_initiated_metadata.rs`). +const SERVER_SERVICE_TYPE: &str = "_sendspin-server._tcp.local."; + +/// Service type that clients which only run their own embedded server (e.g. +/// ESPHome's `sendspin:` component) advertise for themselves, to be discovered +/// and dialed by a server. +const CLIENT_SERVICE_TYPE: &str = "_sendspin._tcp.local."; + +/// A live mDNS advertisement. Unregisters and shuts down its background +/// daemon on drop — hold this alive for as long as the server should stay +/// discoverable. +pub struct Advertisement { + daemon: ServiceDaemon, + fullname: String, +} + +impl Advertisement { + /// Advertise a running server. `server_id` should be the same stable + /// identifier passed to [`crate::server::ServerListener::bind`] — it + /// becomes both the mDNS instance name and the advertised hostname. + /// `path` is the HTTP path clients should connect to (the spec fixes + /// this to `/sendspin` for real deployments). + pub fn new(server_id: &str, name: &str, port: u16, path: &str) -> Result { + let daemon = ServiceDaemon::new() + .map_err(|e| Error::Connection(format!("mDNS daemon start failed: {e}")))?; + let service = ServiceInfo::new( + SERVER_SERVICE_TYPE, + server_id, + &format!("{server_id}.local."), + "", + port, + &[("path", path), ("name", name)][..], + ) + .map_err(|e| Error::Connection(format!("mDNS service build failed: {e}")))? + .enable_addr_auto(); + let fullname = service.get_fullname().to_string(); + daemon + .register(service) + .map_err(|e| Error::Connection(format!("mDNS register failed: {e}")))?; + Ok(Self { daemon, fullname }) + } +} + +impl Drop for Advertisement { + fn drop(&mut self) { + if let Err(e) = self.daemon.unregister(&self.fullname) { + log::warn!("mDNS unregister failed: {e}"); + } + if let Err(e) = self.daemon.shutdown() { + log::warn!("mDNS daemon shutdown failed: {e}"); + } + } +} + +/// Discovers Sendspin clients that advertise themselves over mDNS instead of +/// dialing out (see [`CLIENT_SERVICE_TYPE`]'s docs for why real hardware +/// routinely needs this). +pub struct ClientBrowser { + daemon: ServiceDaemon, + receiver: Receiver, +} + +impl ClientBrowser { + /// Start browsing. Keep this alive for as long as discovery should keep + /// running — dropping it shuts down its background daemon. + pub fn new() -> Result { + let daemon = ServiceDaemon::new() + .map_err(|e| Error::Connection(format!("mDNS daemon start failed: {e}")))?; + let receiver = daemon + .browse(CLIENT_SERVICE_TYPE) + .map_err(|e| Error::Connection(format!("mDNS browse failed: {e}")))?; + Ok(Self { daemon, receiver }) + } + + /// Wait for the next discovery event: a client resolving at a usable URL, + /// or a previously-advertised client being removed. Skips events that + /// aren't a fully-resolved, usable service or a removal. Returns `None` + /// once the daemon shuts down. Callers loop on this to keep discovering. + pub async fn next_event(&self) -> Option { + while let Ok(event) = self.receiver.recv_async().await { + match event { + ServiceEvent::ServiceResolved(info) => { + if let Some(url) = resolve_client_url(&info) { + return Some(Discovered::Found { + fullname: info.fullname.clone(), + url, + }); + } + } + ServiceEvent::ServiceRemoved(_service_type, fullname) => { + return Some(Discovered::Removed { fullname }); + } + _ => {} + } + } + None + } + + /// Wait for the next resolved client, returning its mDNS instance full name + /// (stable identity across address changes) and WebSocket URL (ready to + /// hand to [`crate::server::dial_client`]). Removal events are skipped. + pub async fn next_client(&self) -> Option<(String, String)> { + loop { + match self.next_event().await? { + Discovered::Found { fullname, url } => return Some((fullname, url)), + Discovered::Removed { .. } => continue, + } + } + } + + /// Like [`Self::next_client`], but returns only the WebSocket URL. + pub async fn next_client_url(&self) -> Option { + self.next_client().await.map(|(_fullname, url)| url) + } +} + +/// An event from [`ClientBrowser::next_event`]. +#[derive(Debug, Clone)] +pub enum Discovered { + /// A client resolved at a usable WebSocket URL. + Found { + /// mDNS instance full name — stable identity across address changes. + fullname: String, + /// WebSocket URL, ready to hand to [`crate::server::dial_client`]. + url: String, + }, + /// A previously-advertised client's service was removed from mDNS. + Removed { + /// mDNS instance full name of the service that went away. + fullname: String, + }, +} + +impl Drop for ClientBrowser { + fn drop(&mut self) { + if let Err(e) = self.daemon.shutdown() { + log::warn!("mDNS daemon shutdown failed: {e}"); + } + } +} + +/// Build a WebSocket URL from a resolved client service, or `None` if it has +/// no usable address or well-formed `path` TXT property. Addresses are sorted +/// before selection: the resolver returns them unordered, and [`ClientBrowser`] +/// callers treat an address change as "device moved, reconnect", so a stable +/// choice avoids reconnect thrashing on multi-homed hosts. +fn resolve_client_url(info: &ResolvedService) -> Option { + // Accept both IPv4 and IPv6; drop loopback/link-local/unspecified so we + // never dial an address that isn't routable to the device. + let mut addrs: Vec = info + .get_addresses() + .iter() + .map(ScopedIp::to_ip_addr) + .filter(|a| !a.is_loopback() && !is_link_local(a) && !a.is_unspecified()) + .collect(); + addrs.sort(); + let addr = addrs.into_iter().next()?; + let path = info.get_property_val_str("path")?; + if !path.starts_with('/') { + return None; + } + // IPv6 literals must be bracketed in a URL authority. + let host = match addr { + IpAddr::V4(v4) => v4.to_string(), + IpAddr::V6(v6) => format!("[{v6}]"), + }; + Some(format!("ws://{host}:{port}{path}", port = info.get_port())) +} + +/// Whether an address is link-local (IPv4 169.254.0.0/16 or IPv6 fe80::/10). +fn is_link_local(ip: &IpAddr) -> bool { + match ip { + IpAddr::V4(v4) => v4.is_link_local(), + IpAddr::V6(v6) => (v6.segments()[0] & 0xffc0) == 0xfe80, + } +} diff --git a/src/server/group.rs b/src/server/group.rs new file mode 100644 index 0000000..7241ead --- /dev/null +++ b/src/server/group.rs @@ -0,0 +1,274 @@ +// ABOUTME: Multi-client synchronized playback group +// ABOUTME: Every member receives identical audio bytes with identical timestamps, so each client's clock-sync offset alone yields sample-accurate multi-room sync + +use crate::error::Error; +use crate::protocol::messages::{PlayerCommand, StreamPlayerConfig}; +use crate::server::binary::encode_audio_frame; +use crate::server::connection::{AudioEnqueue, ServerSender}; +use crate::sync::raw_clock::Clock; +use futures_util::future::join_all; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use tokio_tungstenite::tungstenite::Bytes; + +/// How far ahead of "now" (in this group's clock domain) the audio timeline is +/// anchored, by default. Must comfortably exceed a client's own startup +/// buffering latency, or it'll receive chunks whose intended playback time has +/// already passed. v1 uses one fixed lead time for the whole group rather than +/// negotiating per-client (see [`Group::with_send_ahead_us`]); per-client lead +/// negotiation is deferred along with per-client format negotiation. +pub const DEFAULT_SEND_AHEAD_US: i64 = 250_000; + +/// A synchronized playback group. +/// +/// The server-side synchronization trick is simple and doesn't require +/// knowing any client's individual clock: every member is sent the *same* +/// audio bytes tagged with the *same* `server/time`-domain timestamp. Each +/// client independently converts that timestamp into its own clock domain +/// (via the offset/drift it tracks from `client/time`/`server/time` +/// exchanges) and schedules local playback there — so two members with +/// converged clock-sync play the same chunk at the same wall-clock instant +/// without the server ever comparing their clocks to each other. +/// +/// v1 scope: one shared PCM format for the whole group — no per-client +/// transcoding, so a member that can't take the group's format is a v1 +/// limitation, not silently-wrong audio. No late-join catch-up (a client +/// added mid-stream just gets `stream/start` and audio from that point +/// forward) and no historical buffer replay. +pub struct Group { + clock: Arc, + send_ahead_us: i64, + state: Mutex, +} + +/// Mutable group state behind a single lock, so `Group`'s methods take `&self` +/// and a slow member can never block membership changes or another member. +struct State { + members: HashMap, + stream_config: Option, + /// Timestamp (this group's clock domain) to stamp on the next pushed chunk. + /// `None` until the first push after a start/clear (re)anchors the timeline. + next_ts_us: Option, + /// Carry for the sub-microsecond part of a chunk's duration (numerator over + /// the sample rate), so advancing the timeline doesn't accumulate drift. + residue: i64, +} + +impl State { + fn reset_timeline(&mut self) { + self.next_ts_us = None; + self.residue = 0; + } +} + +impl Group { + /// Create an empty group using `clock` as the shared timestamp domain — + /// pass the same clock the [`crate::server::ServerListener`] that + /// accepted these connections was built with, so timestamps here are in + /// the same domain as the `server/time` replies members already trust. + pub fn new(clock: Arc) -> Self { + Self { + clock, + send_ahead_us: DEFAULT_SEND_AHEAD_US, + state: Mutex::new(State { + members: HashMap::new(), + stream_config: None, + next_ts_us: None, + residue: 0, + }), + } + } + + /// Override the default send-ahead lead time. + pub fn with_send_ahead_us(mut self, send_ahead_us: i64) -> Self { + self.send_ahead_us = send_ahead_us; + self + } + + /// Client IDs of every current member. + pub fn member_ids(&self) -> Vec { + self.state.lock().unwrap().members.keys().cloned().collect() + } + + /// Number of current members. + pub fn len(&self) -> usize { + self.state.lock().unwrap().members.len() + } + + /// Whether the group has no members. + pub fn is_empty(&self) -> bool { + self.state.lock().unwrap().members.is_empty() + } + + /// Add a member. If a stream is already active for this group, starts + /// it for the new member too (matching the group's already-negotiated + /// format) — but does not replay any audio already delivered to + /// existing members (no late-join catch-up in v1, see the type docs). + pub async fn add_member( + &self, + client_id: impl Into, + sender: ServerSender, + ) -> Result<(), Error> { + let config = self.state.lock().unwrap().stream_config.clone(); + if let Some(cfg) = config { + sender.send_stream_start(cfg).await?; + } + self.state + .lock() + .unwrap() + .members + .insert(client_id.into(), sender); + Ok(()) + } + + /// Remove a member, if present. The caller is responsible for actually + /// disconnecting it (e.g. via [`crate::server::ServerConnection::disconnect`]) — + /// this only stops future broadcasts from reaching it. + pub fn remove_member(&self, client_id: &str) -> Option { + self.state.lock().unwrap().members.remove(client_id) + } + + /// Start (or restart, e.g. after a format change) the shared stream for + /// every current member, and re-anchor the audio timeline. + pub async fn start_stream(&self, config: StreamPlayerConfig) { + { + let mut state = self.state.lock().unwrap(); + state.stream_config = Some(config.clone()); + state.reset_timeline(); + } + self.broadcast_control(|sender| { + let config = config.clone(); + async move { sender.send_stream_start(config).await } + }) + .await; + } + + /// Push one PCM chunk to every member, stamped with a single shared + /// timestamp so every member schedules it at the same instant. Returns + /// that timestamp. + /// + /// The timestamp comes from an anchored timeline rather than + /// `now + lead` per call, so pushing faster or slower than real time + /// doesn't shift playback: the first push after a start/clear anchors at + /// `now + send_ahead_us`, and each push advances the timeline by the + /// chunk's own duration (derived from the stream format). If pushes fall + /// behind — the timeline would schedule a chunk too close to now — it + /// re-anchors forward. Enqueue is non-blocking, so one slow member never + /// delays the others; a member whose connection has died is pruned. + pub fn push_audio(&self, pcm: &[u8]) -> i64 { + let mut state = self.state.lock().unwrap(); + let now = self.clock.now_micros(); + + // Anchor on the first push, or re-anchor if the timeline has fallen too + // close to (or behind) now — otherwise chunks would be scheduled in the + // past. Half the lead is the low-water mark, giving hysteresis so steady + // real-time pacing doesn't re-anchor every push. + let ts = match state.next_ts_us { + Some(t) if t >= now + self.send_ahead_us / 2 => t, + _ => now + self.send_ahead_us, + }; + + // Advance the timeline by this chunk's exact duration, carrying the + // fractional-microsecond remainder so it doesn't drift. + let advanced = match &state.stream_config { + Some(cfg) => { + let bytes_per_sample = (cfg.channels as usize) * (cfg.bit_depth as usize / 8); + if bytes_per_sample > 0 && cfg.sample_rate > 0 { + let samples = (pcm.len() / bytes_per_sample) as i64; + let total = samples * 1_000_000 + state.residue; + let rate = cfg.sample_rate as i64; + state.residue = total % rate; + Some(ts + total / rate) + } else { + Some(ts) + } + } + None => Some(ts), + }; + state.next_ts_us = advanced; + + // Encode once; fan the same frame out to every member as cheap refcount + // clones. Prune members whose connection has died. + let frame: Bytes = encode_audio_frame(ts, pcm).into(); + let mut dead = Vec::new(); + for (id, sender) in state.members.iter() { + match sender.enqueue_audio(frame.clone()) { + Ok(AudioEnqueue::Sent) => {} + Ok(AudioEnqueue::Evicted) => { + log::trace!("group member {id} audio backlog full, dropping chunk") + } + Err(_) => dead.push(id.clone()), + } + } + for id in dead { + log::warn!("dropping dead group member {id}"); + state.members.remove(&id); + } + ts + } + + /// Broadcast a player command (volume, mute, static delay) to every + /// member. + pub async fn send_player_command(&self, command: PlayerCommand) { + self.broadcast_control(|sender| { + let command = command.clone(); + async move { sender.send_player_command(command).await } + }) + .await; + } + + /// End the shared stream for every current member and reset the timeline. + pub async fn end_stream(&self) { + { + let mut state = self.state.lock().unwrap(); + state.stream_config = None; + state.reset_timeline(); + } + self.broadcast_control(|sender| async move { sender.send_stream_end().await }) + .await; + } + + /// Run an awaiting control send `f` concurrently against every member — a + /// slow member never blocks delivery to the others — then drop any member + /// `f` failed against (its writer task is gone; the failure is permanent). + async fn broadcast_control(&self, f: F) + where + F: Fn(ServerSender) -> Fut, + Fut: std::future::Future>, + { + let members: Vec<(String, ServerSender)> = { + let state = self.state.lock().unwrap(); + state + .members + .iter() + .map(|(id, sender)| (id.clone(), sender.clone())) + .collect() + }; + let results = join_all(members.into_iter().map(|(id, sender)| { + let fut = f(sender); + async move { (id, fut.await) } + })) + .await; + let mut state = self.state.lock().unwrap(); + for (id, result) in results { + if let Err(e) = result { + log::warn!("dropping group member {id}: {e}"); + state.members.remove(&id); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::sync::raw_clock::DefaultClock; + + #[test] + fn new_group_is_empty() { + let group = Group::new(Arc::new(DefaultClock::default())); + assert!(group.is_empty()); + assert_eq!(group.len(), 0); + assert_eq!(group.member_ids().len(), 0); + } +} diff --git a/src/server/listener.rs b/src/server/listener.rs new file mode 100644 index 0000000..45d0083 --- /dev/null +++ b/src/server/listener.rs @@ -0,0 +1,195 @@ +// ABOUTME: Inbound WebSocket acceptor that drives the Sendspin protocol-server +// ABOUTME: state machine (client/hello -> server/hello) on every peer that connects. + +use crate::error::Error; +use crate::protocol::messages::ConnectionReason; +use crate::server::connection::ServerConnection; +use crate::sync::raw_clock::{Clock, DefaultClock}; +use std::net::SocketAddr; +use std::sync::Arc; +use tokio::io::{AsyncRead, AsyncWrite}; +use tokio::net::{lookup_host, TcpListener, TcpSocket, TcpStream, ToSocketAddrs}; +use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response}; +use tokio_tungstenite::tungstenite::http; +use tokio_tungstenite::{accept_async, accept_hdr_async, WebSocketStream}; + +/// Accept inbound WebSocket peers and drive each one through the +/// protocol-**server** state machine: read `client/hello`, reply +/// `server/hello`, then hand back a [`ServerConnection`] for pushing +/// stream/audio messages and receiving state/command/goodbye. +/// +/// This is the counterpart to [`crate::protocol::listener::ProtocolListener`], +/// which accepts inbound connections but drives the protocol-**client** role +/// on them (used when a server dials out to a client that runs its own tiny +/// WS listener — a reversed-topology case). `ServerListener` is what a +/// Sendspin server itself binds to accept the usual inbound player +/// connections. +/// +/// [`Self::accept`] drives the full handshake before returning, so it serves +/// one inbound connection at a time — a slow handshake blocks the next +/// `accept()`. `tokio::spawn` a task per accepted connection if you need to +/// keep accepting while driving existing ones. +pub struct ServerListener { + tcp: TcpListener, + server_id: String, + server_name: String, + path: Option, + clock: Arc, +} + +impl std::fmt::Debug for ServerListener { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ServerListener") + .field("local_addr", &self.tcp.local_addr().ok()) + .field("server_id", &self.server_id) + .field("path", &self.path) + .finish() + } +} + +impl ServerListener { + /// Bind a listener. `server_id` should be stable across restarts (it's + /// how a client recognizes "the same server" across reconnects); + /// `server_name` is human-readable and shown to users. + pub async fn bind( + addr: impl ToSocketAddrs, + server_id: impl Into, + server_name: impl Into, + ) -> Result { + // Bind with SO_REUSEADDR so a port freed by a just-torn-down server can + // be reused immediately — otherwise recreating a group on the same port + // can race the old socket's close and fail with EADDRINUSE. + let sockaddr: SocketAddr = lookup_host(addr) + .await + .map_err(|e| Error::Connection(format!("resolve failed: {e}")))? + .next() + .ok_or_else(|| Error::Connection("no address to bind".to_string()))?; + let socket = if sockaddr.is_ipv4() { + TcpSocket::new_v4() + } else { + TcpSocket::new_v6() + } + .map_err(|e| Error::Connection(format!("socket create failed: {e}")))?; + socket + .set_reuseaddr(true) + .map_err(|e| Error::Connection(format!("set_reuseaddr failed: {e}")))?; + socket + .bind(sockaddr) + .map_err(|e| Error::Connection(format!("bind failed: {e}")))?; + let tcp = socket + .listen(128) + .map_err(|e| Error::Connection(format!("listen failed: {e}")))?; + Ok(Self { + tcp, + server_id: server_id.into(), + server_name: server_name.into(), + path: None, + clock: Arc::new(DefaultClock::default()), + }) + } + + /// Restrict accepted connections to a specific HTTP path (the Sendspin + /// spec fixes this to `/sendspin` for real deployments). Mismatches are + /// rejected with HTTP 404 during the WebSocket handshake; the listener + /// stays bound. Defaults to accepting any path. + pub fn path(mut self, path: impl Into) -> Self { + let path = path.into(); + self.path = Some(if path.starts_with('/') { + path + } else { + format!("/{path}") + }); + self + } + + /// Use a custom clock instead of [`DefaultClock`] — mainly for tests that + /// need deterministic or synchronized-with-a-peer timestamps. + pub fn clock(mut self, clock: Arc) -> Self { + self.clock = clock; + self + } + + /// Accept the next inbound connection, returning the driven + /// [`ServerConnection`] and the peer's address. + /// + /// Per-peer failures surface as [`Error`] without affecting the + /// listener; callers typically call `accept()` in a loop. + /// + /// Not cancel-safe: dropping the returned future mid-handshake tears + /// down that connection. A peer that connects but stalls the handshake + /// will block this future indefinitely, so wrap it in a timeout if + /// untrusted peers can reach the socket. + pub async fn accept(&self) -> Result<(ServerConnection, SocketAddr), Error> { + let (tcp_stream, peer_addr) = self + .tcp + .accept() + .await + .map_err(|e| Error::Connection(format!("TCP accept failed: {e}")))?; + log::debug!("Accepted TCP connection from {}", peer_addr); + + match self.handshake_and_drive(tcp_stream).await { + Ok(conn) => Ok((conn, peer_addr)), + Err(e) => { + log::warn!("Inbound handshake from {peer_addr} failed: {e}"); + Err(e) + } + } + } + + async fn handshake_and_drive(&self, tcp_stream: TcpStream) -> Result { + let ws = self.handshake_ws(tcp_stream).await?; + // Inbound: the client initiated the connection, so the server is + // simply present/available — announce Discovery rather than Playback. + ServerConnection::drive( + ws, + &self.server_id, + &self.server_name, + ConnectionReason::Discovery, + Arc::clone(&self.clock), + ) + .await + } + + // `ErrorResponse` is large by Clippy's standard but mandated by + // tungstenite's `Callback` trait — same tradeoff ProtocolListener makes. + #[allow(clippy::result_large_err)] + async fn handshake_ws(&self, stream: S) -> Result, Error> + where + S: AsyncRead + AsyncWrite + Unpin, + { + match &self.path { + Some(expected_path) => { + let expected = expected_path.clone(); + let callback = move |request: &Request, response: Response| { + if request.uri().path() == expected { + Ok(response) + } else { + log::debug!( + "Rejecting inbound connection: path {:?} != expected {:?}", + request.uri().path(), + expected + ); + Err(http::Response::builder() + .status(http::StatusCode::NOT_FOUND) + .body(None) + .expect("static 404 response is well-formed")) + as Result + } + }; + accept_hdr_async(stream, callback) + .await + .map_err(|e| Error::WebSocket(format!("WebSocket handshake failed: {e}"))) + } + None => accept_async(stream) + .await + .map_err(|e| Error::WebSocket(format!("WebSocket handshake failed: {e}"))), + } + } + + /// Local bound address. + pub fn local_addr(&self) -> Result { + self.tcp + .local_addr() + .map_err(|e| Error::Connection(format!("local_addr failed: {e}"))) + } +} diff --git a/src/server/manager.rs b/src/server/manager.rs new file mode 100644 index 0000000..e97795a --- /dev/null +++ b/src/server/manager.rs @@ -0,0 +1,327 @@ +// ABOUTME: Continuous discovery + reconnect-with-backoff supervision for clients +// ABOUTME: that only run their own embedded server (the supervised form of dial_client) + +use crate::protocol::messages::Message; +use crate::server::connection::{ServerConnection, ServerSender}; +use crate::server::dial::dial_client; +use crate::server::discovery::{ClientBrowser, Discovered}; +use crate::sync::raw_clock::Clock; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender}; +use tokio::sync::watch; +use tokio::task::JoinHandle; +use tokio::time::Instant; + +/// Initial reconnect backoff, and the value backoff resets to after a stable +/// session or an address change. +const MIN_BACKOFF: Duration = Duration::from_secs(1); + +/// Reconnect backoff ceiling — matches aiosendspin's `MAX_RECONNECT_BACKOFF_S`. +const MAX_BACKOFF: Duration = Duration::from_secs(300); + +/// A connection must last at least this long before backoff resets to the +/// minimum (matches aiosendspin's `STABLE_SERVER_INITIATED_SESSION_S`), so a +/// crash-looping device isn't hammered at 1-second intervals. +const STABLE_SESSION: Duration = Duration::from_secs(10); + +/// Control signal sent to a client supervisor task via a watch channel. +#[derive(Clone, Debug)] +enum Directive { + /// (Re)dial the client at this URL. A new URL for an already-connected + /// client makes the supervisor drop the current connection and redial. + Dial(String), + /// Stop supervising — the device's mDNS advertisement was removed. + Stop, +} + +/// Events for a client discovered and managed by [`ClientManager`]. The +/// manager owns each connection's message loop internally so it can detect +/// disconnection and drive reconnects — callers get a [`ServerSender`] for +/// control instead of the raw [`ServerConnection`]. +#[derive(Debug)] +pub enum ClientEvent { + /// A client connected — the first time, or after a reconnect. `client_id` + /// is stable across reconnects (it comes from the client's own + /// `client/hello`), so callers can use it as the group-membership key. + Connected { + /// The connected client's identifier, from its `client/hello`. + client_id: String, + /// The mDNS instance fullname this connection was dialed from — stable + /// discovery identity, lets callers map a connection back to the + /// discovered service (the `client_id` may be an opaque MAC that does + /// not match the advertised name). + fullname: String, + /// Roles this server granted the client. + active_roles: Vec, + /// Sender for pushing stream/audio/command messages to this client. + sender: ServerSender, + }, + /// A `client/state`, `client/command`, or `client/goodbye` message from + /// a connected client (`client/time` is consumed internally and never + /// forwarded, same convention as [`ServerConnection::recv_message`]). + Message { + /// Which client sent this message. + client_id: String, + /// The message itself. + message: Box, + }, + /// The client disconnected. If it's still discoverable, a reconnect + /// attempt is already running in the background — this just tells the + /// caller to stop treating `client_id` as a live group member for now. + Disconnected { + /// The client that disconnected. + client_id: String, + }, +} + +struct ManagedClient { + handle: JoinHandle<()>, + directive_tx: watch::Sender, + url: String, +} + +/// Discovers Sendspin clients that only run their own embedded server and keeps +/// each one connected: dials on discovery, retries with capped exponential +/// backoff on failure or disconnect, re-dials promptly if a device reappears at +/// a new address, and stops supervising once a device's mDNS advertisement is +/// removed. +pub struct ClientManager { + tasks: Arc>>, + browse_handle: JoinHandle<()>, +} + +impl ClientManager { + /// Start discovering and managing every Sendspin client this process can + /// see on the network. Returns immediately; events arrive on the + /// returned receiver as they happen. Drop the returned `ClientManager` + /// to stop discovery and every reconnect loop it's running. + /// + /// This is unfiltered: on a LAN where other servers already serve some of + /// these clients, you will compete with them for those devices. Use + /// [`Self::start_filtered`] to scope discovery to a known set of devices. + pub fn start( + server_id: impl Into, + server_name: impl Into, + clock: Arc, + ) -> Result<(Self, UnboundedReceiver), crate::error::Error> { + Self::start_filtered(server_id, server_name, clock, |_fullname| true) + } + + /// Like [`Self::start`], but only discovers and manages clients whose + /// mDNS instance full name (e.g. `my-device._sendspin._tcp.local.`) + /// satisfies `allow`. + pub fn start_filtered( + server_id: impl Into, + server_name: impl Into, + clock: Arc, + allow: impl Fn(&str) -> bool + Send + 'static, + ) -> Result<(Self, UnboundedReceiver), crate::error::Error> { + let server_id = server_id.into(); + let server_name = server_name.into(); + let (event_tx, event_rx) = unbounded_channel(); + let tasks: Arc>> = + Arc::new(Mutex::new(HashMap::new())); + + let browser = ClientBrowser::new()?; + let tasks_for_browse = Arc::clone(&tasks); + let browse_handle = tokio::spawn(async move { + while let Some(event) = browser.next_event().await { + match event { + Discovered::Found { fullname, url } => { + if !allow(&fullname) { + continue; + } + let mut tasks = tasks_for_browse.lock().unwrap(); + match tasks.get_mut(&fullname) { + // Same device, same address — already supervised. + Some(existing) if existing.url == url => {} + // Same device at a new address: redirect the running + // supervisor (it closes the current connection, emits + // Disconnected, and redials). Respawn only if the + // supervisor has already exited. + Some(existing) => { + log::info!( + "[{fullname}] address changed ({} -> {url}), reconnecting", + existing.url + ); + existing.url = url.clone(); + if existing + .directive_tx + .send(Directive::Dial(url.clone())) + .is_err() + { + *existing = spawn_supervisor( + fullname.clone(), + url, + server_id.clone(), + server_name.clone(), + Arc::clone(&clock), + event_tx.clone(), + ); + } + } + None => { + let managed = spawn_supervisor( + fullname.clone(), + url, + server_id.clone(), + server_name.clone(), + Arc::clone(&clock), + event_tx.clone(), + ); + tasks.insert(fullname, managed); + } + } + } + // The device's advertisement went away: stop supervising it + // (gracefully, so a live connection emits Disconnected) + // instead of redialing a gone device forever. + Discovered::Removed { fullname } => { + if let Some(managed) = tasks_for_browse.lock().unwrap().remove(&fullname) { + log::info!("[{fullname}] mDNS service removed, stopping supervision"); + let _ = managed.directive_tx.send(Directive::Stop); + } + } + } + } + }); + + Ok(( + Self { + tasks, + browse_handle, + }, + event_rx, + )) + } +} + +impl Drop for ClientManager { + fn drop(&mut self) { + self.browse_handle.abort(); + for (_, managed) in self.tasks.lock().unwrap().drain() { + managed.handle.abort(); + } + } +} + +fn spawn_supervisor( + fullname: String, + url: String, + server_id: String, + server_name: String, + clock: Arc, + event_tx: UnboundedSender, +) -> ManagedClient { + let (directive_tx, directive_rx) = watch::channel(Directive::Dial(url.clone())); + let handle = tokio::spawn(supervise( + fullname, + directive_rx, + server_id, + server_name, + clock, + event_tx, + )); + ManagedClient { + handle, + directive_tx, + url, + } +} + +/// Keep one discovered client connected until told to stop. Dials the current +/// directive URL, reports Connected/Message/Disconnected, and reconnects with +/// capped backoff. A `Directive::Dial` with a new URL closes the current +/// connection and redials it immediately; `Directive::Stop` ends the task. +async fn supervise( + fullname: String, + mut directive_rx: watch::Receiver, + server_id: String, + server_name: String, + clock: Arc, + event_tx: UnboundedSender, +) { + let mut backoff = MIN_BACKOFF; + loop { + let url = match directive_rx.borrow_and_update().clone() { + Directive::Dial(url) => url, + Directive::Stop => return, + }; + + match dial_client(&url, &server_id, &server_name, Arc::clone(&clock)).await { + Ok(conn) => { + // A Stop that arrived during the dial: don't announce a + // connection we're about to tear down. + if matches!(*directive_rx.borrow(), Directive::Stop) { + return; + } + let started = Instant::now(); + let client_id = conn.client_id().to_string(); + log::info!("[{fullname}] connected as {client_id} ({url})"); + let _ = event_tx.send(ClientEvent::Connected { + client_id: client_id.clone(), + fullname: fullname.clone(), + active_roles: conn.active_roles().to_vec(), + sender: conn.sender(), + }); + + // Drain until the client disconnects, or a directive redirects + // us (dropping `conn` here closes that connection). + let redirected = tokio::select! { + _ = drain_until_disconnected(conn, &client_id, &event_tx) => false, + _ = directive_rx.changed() => true, + }; + + let _ = event_tx.send(ClientEvent::Disconnected { + client_id: client_id.clone(), + }); + + if matches!(*directive_rx.borrow(), Directive::Stop) { + return; + } + if redirected { + log::info!("[{fullname}] address changed, reconnecting immediately"); + backoff = MIN_BACKOFF; + continue; + } + log::info!("[{fullname}] disconnected, will retry"); + if started.elapsed() >= STABLE_SESSION { + backoff = MIN_BACKOFF; + } + } + Err(e) => { + log::warn!("[{fullname}] dial to {url} failed: {e}"); + } + } + + // Wait out the backoff, but wake early if a directive arrives. + tokio::select! { + _ = tokio::time::sleep(backoff) => {} + _ = directive_rx.changed() => {} + } + if matches!(*directive_rx.borrow(), Directive::Stop) { + return; + } + backoff = (backoff * 2).min(MAX_BACKOFF); + } +} + +async fn drain_until_disconnected( + mut conn: ServerConnection, + client_id: &str, + event_tx: &UnboundedSender, +) { + while let Some(message) = conn.recv_message().await { + if event_tx + .send(ClientEvent::Message { + client_id: client_id.to_string(), + message: Box::new(message), + }) + .is_err() + { + return; // receiver dropped; nothing left to report to + } + } +} diff --git a/src/server/mod.rs b/src/server/mod.rs new file mode 100644 index 0000000..740067b --- /dev/null +++ b/src/server/mod.rs @@ -0,0 +1,30 @@ +// ABOUTME: Server-role implementation of the Sendspin protocol +// ABOUTME: Accepts or dials player clients, syncs clocks, streams audio to synchronized multi-client groups + +// Handles both connection directions: clients that dial in +// (`ServerListener::accept`) and clients that only run their own embedded +// server and must be discovered over mDNS and dialed (`ClientBrowser` + +// `dial_client`, or the supervised `ClientManager`). +// +// Not yet supported: per-client codec transcoding (one PCM format per group), +// the non-player roles (color, visualizer, artwork, controller, metadata), +// external player registration, and late-join history replay — a client that +// joins mid-stream receives the current stream and all subsequent audio, +// synchronized with existing members, but nothing buffered from before it +// joined. + +mod binary; +mod connection; +mod dial; +mod discovery; +mod group; +mod listener; +mod manager; + +pub use binary::encode_audio_frame; +pub use connection::{ServerConnection, ServerConnectionGuard, ServerSender}; +pub use dial::dial_client; +pub use discovery::{Advertisement, ClientBrowser, Discovered}; +pub use group::{Group, DEFAULT_SEND_AHEAD_US}; +pub use listener::ServerListener; +pub use manager::{ClientEvent, ClientManager}; diff --git a/tests/client_manager.rs b/tests/client_manager.rs new file mode 100644 index 0000000..c07b3f5 --- /dev/null +++ b/tests/client_manager.rs @@ -0,0 +1,206 @@ +// ABOUTME: Integration tests for ClientManager +// ABOUTME: Discovery/dial, retry after disconnect, and re-dial on address change +// +// These exercise real mDNS multicast, so they are wrapped in +// common::retry_flaky to absorb environmental timing noise on a shared LAN. + +mod common; + +use futures_util::{SinkExt, StreamExt}; +use mdns_sd::ServiceInfo; +use sendspin::protocol::messages::{ClientHello, Message}; +use sendspin::server::{ClientEvent, ClientManager}; +use sendspin::DefaultClock; +use std::sync::Arc; +use std::time::Duration; +use tokio::net::TcpListener; +use tokio::time::timeout; +use tokio_tungstenite::tungstenite::Message as WsMessage; + +fn test_hello(client_id: &str) -> ClientHello { + ClientHello { + client_id: client_id.to_string(), + name: "Fake Embedded Client".to_string(), + version: 1, + supported_roles: vec!["player@v1".to_string()], + device_info: None, + player_v1_support: None, + artwork_v1_support: None, + visualizer_v1_support: None, + } +} + +async fn free_port() -> u16 { + TcpListener::bind("127.0.0.1:0") + .await + .unwrap() + .local_addr() + .unwrap() + .port() +} + +fn advertise(fullname_instance: &str, port: u16) -> mdns_sd::ServiceDaemon { + let daemon = mdns_sd::ServiceDaemon::new().expect("mdns daemon"); + let service = ServiceInfo::new( + "_sendspin._tcp.local.", + fullname_instance, + &format!("{fullname_instance}.local."), + "", + port, + &[("path", "/sendspin")][..], + ) + .unwrap() + .enable_addr_auto(); + daemon.register(service).expect("register"); + daemon +} + +/// Accepts exactly one connection, completes the handshake, then closes — +/// standing in for a real device's connection dropping. +async fn accept_one_and_close(port: u16, client_id: &str) { + let tcp = TcpListener::bind(("127.0.0.1", port)).await.unwrap(); + let (stream, _) = tcp.accept().await.unwrap(); + let ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + let (mut write, mut read) = ws.split(); + let hello = serde_json::to_string(&Message::ClientHello(test_hello(client_id))).unwrap(); + write.send(WsMessage::Text(hello.into())).await.unwrap(); + read.next().await.expect("no server/hello").unwrap(); + write.close().await.ok(); +} + +/// Waits for a `Connected` event whose `client_id` matches, ignoring any +/// interleaved `Disconnected`/`Message` events from other test noise on a +/// shared LAN (see tests/dial_discovery.rs for why that matters here). +async fn next_connected( + rx: &mut tokio::sync::mpsc::UnboundedReceiver, + expected_client_id: &str, +) { + loop { + match timeout(Duration::from_secs(30), rx.recv()) + .await + .expect("timed out waiting for Connected") + .expect("event channel closed") + { + ClientEvent::Connected { client_id, .. } if client_id == expected_client_id => return, + _ => continue, + } + } +} + +/// Same idea as [`next_connected`], for `Disconnected` — the shared LAN can +/// interleave events from other real devices, so this skips anything that +/// isn't the one we're waiting for instead of assuming the next event is it. +async fn next_disconnected( + rx: &mut tokio::sync::mpsc::UnboundedReceiver, + expected_client_id: &str, +) { + loop { + match timeout(Duration::from_secs(30), rx.recv()) + .await + .expect("timed out waiting for Disconnected") + .expect("event channel closed") + { + ClientEvent::Disconnected { client_id } if client_id == expected_client_id => return, + _ => continue, + } + } +} + +#[tokio::test] +async fn reconnects_after_the_client_drops() { + if !common::net_tests_enabled() { + eprintln!("skipping: set SENDSPIN_NET_TESTS=1 to run mDNS multicast tests"); + return; + } + common::retry_flaky(3, reconnects_after_the_client_drops_impl).await; +} + +async fn reconnects_after_the_client_drops_impl() { + let port = free_port().await; + // Includes the port so a retried attempt (after a previous one + // panicked, leaving its bare mdns_sd::ServiceDaemon's advertisement + // registered with no Drop impl to clean it up) never collides with a + // still-lingering advertisement from an earlier attempt. + let instance_name = format!("manager-test-reconnect-{port}"); + let daemon = advertise(&instance_name, port); + + // Filtered to just this test's own fake client: the dev network this + // was written against has real Sendspin devices on it too, and an + // unfiltered manager would also try to dial (and hold reconnect loops + // against) those — noisy for the test and, worse, actively interferes + // with hardware you might be using for something else at the time. + let (manager, mut events) = ClientManager::start_filtered( + "test-server", + "Test Server", + Arc::new(DefaultClock::default()), + move |fullname| fullname.starts_with(&instance_name), + ) + .expect("start"); + + accept_one_and_close(port, "reconnect-client").await; + next_connected(&mut events, "reconnect-client").await; + + // First connection closing should surface as Disconnected... + next_disconnected(&mut events, "reconnect-client").await; + + // ...and the manager should dial again on its own (backoff starts at + // 1s) without anything external triggering it. + accept_one_and_close(port, "reconnect-client").await; + next_connected(&mut events, "reconnect-client").await; + + drop(manager); + let _ = daemon.shutdown(); +} + +#[tokio::test] +async fn redials_promptly_when_the_same_device_reappears_at_a_new_address() { + if !common::net_tests_enabled() { + eprintln!("skipping: set SENDSPIN_NET_TESTS=1 to run mDNS multicast tests"); + return; + } + common::retry_flaky( + 3, + redials_promptly_when_the_same_device_reappears_at_a_new_address_impl, + ) + .await; +} + +async fn redials_promptly_when_the_same_device_reappears_at_a_new_address_impl() { + let _ = env_logger::builder().is_test(false).try_init(); + let old_port = free_port().await; + let new_port = free_port().await; + // Includes old_port (fresh per attempt) so a retried attempt never + // collides with a still-lingering advertisement from an earlier one — + // stable across the old_port -> new_port move within *this* attempt, + // since that's the specific thing being tested. + let instance_name = format!("manager-test-address-change-{old_port}"); + let daemon = advertise(&instance_name, old_port); + + let (manager, mut events) = ClientManager::start_filtered( + "test-server", + "Test Server", + Arc::new(DefaultClock::default()), + { + let instance_name = instance_name.clone(); + move |fullname| fullname.starts_with(&instance_name) + }, + ) + .expect("start"); + + accept_one_and_close(old_port, "movable-client").await; + next_connected(&mut events, "movable-client").await; + next_disconnected(&mut events, "movable-client").await; + + // Re-advertise the *same* instance name at a new port before the old + // task's backoff would naturally retry — this proves reconnection + // happens via the new resolution, not by coincidentally retrying an + // address that still happens to work. + let _ = daemon.shutdown(); + let daemon2 = advertise(&instance_name, new_port); + + accept_one_and_close(new_port, "movable-client").await; + next_connected(&mut events, "movable-client").await; + + drop(manager); + let _ = daemon2.shutdown(); +} diff --git a/tests/common/mod.rs b/tests/common/mod.rs new file mode 100644 index 0000000..275044e --- /dev/null +++ b/tests/common/mod.rs @@ -0,0 +1,66 @@ +// ABOUTME: Shared test helper for retrying flaky mDNS-multicast integration tests + +/// Re-run an async test body up to `attempts` times, succeeding as soon as one +/// attempt doesn't panic and only failing (re-raising the last attempt's panic) +/// if every attempt does. +/// +/// Intended for tests that depend on real mDNS multicast timing, where +/// occasional packet loss or scheduling delay on a shared LAN is environmental +/// noise rather than a real failure. A test that fails the same way every time +/// still fails after `attempts` tries. +/// +/// `test_fn` must be a zero-argument async function so each retry gets a fresh +/// attempt; `tokio::spawn` isolates a panicking attempt and yields a `JoinError` +/// to detect it. +/// Whether real-network tests (which use live mDNS multicast) are enabled, via +/// the `SENDSPIN_NET_TESTS` environment variable. Off by default so an ordinary +/// `cargo test` run doesn't depend on multicast being available. +#[allow(dead_code)] +pub fn net_tests_enabled() -> bool { + std::env::var_os("SENDSPIN_NET_TESTS").is_some() +} + +// Each integration test file that does `mod common;` compiles its own copy, and +// no single file uses both helpers — hence the allow. +#[allow(dead_code)] +pub async fn retry_flaky(attempts: u32, test_fn: F) +where + F: Fn() -> Fut, + Fut: std::future::Future + Send + 'static, +{ + for attempt in 1..=attempts { + match tokio::spawn(test_fn()).await { + Ok(()) => return, + Err(join_err) if attempt < attempts => { + eprintln!( + "flaky test attempt {attempt}/{attempts} failed ({join_err}), retrying..." + ); + } + Err(join_err) => std::panic::resume_unwind(join_err.into_panic()), + } + } +} + +/// Synchronous counterpart to [`retry_flaky`], for plain `#[test]` functions +/// that don't need a tokio runtime (e.g. tests using `mdns_sd`'s blocking +/// `recv_timeout` directly rather than `ClientBrowser`'s async API). +#[allow(dead_code)] +pub fn retry_flaky_sync(attempts: u32, test_fn: F) +where + F: Fn(), +{ + for attempt in 1..=attempts { + match std::panic::catch_unwind(std::panic::AssertUnwindSafe(&test_fn)) { + Ok(()) => return, + Err(payload) if attempt < attempts => { + let msg = payload + .downcast_ref::<&str>() + .map(|s| s.to_string()) + .or_else(|| payload.downcast_ref::().cloned()) + .unwrap_or_else(|| "".to_string()); + eprintln!("flaky test attempt {attempt}/{attempts} failed ({msg}), retrying..."); + } + Err(payload) => std::panic::resume_unwind(payload), + } + } +} diff --git a/tests/dial_discovery.rs b/tests/dial_discovery.rs new file mode 100644 index 0000000..fa161b1 --- /dev/null +++ b/tests/dial_discovery.rs @@ -0,0 +1,138 @@ +// ABOUTME: Integration test for the server-initiated connection path — +// ABOUTME: discovering a client that advertises itself over mDNS +// ABOUTME: (like ESPHome's sendspin: component) and dialing in to it. + +mod common; + +use futures_util::{SinkExt, StreamExt}; +use mdns_sd::ServiceInfo; +use sendspin::protocol::messages::Message; +use sendspin::server::{dial_client, ClientBrowser}; +use sendspin::DefaultClock; +use std::sync::Arc; +use std::time::Duration; +use tokio::net::TcpListener; +use tokio::time::timeout; +use tokio_tungstenite::tungstenite::Message as WsMessage; + +/// Stands in for a real embedded client (e.g. ESPHome's sendspin: component) +/// that only ever runs its own WebSocket server and advertises itself over +/// mDNS for a real Sendspin server to discover and dial in to. +async fn run_fake_embedded_client(port: u16) -> Message { + let tcp = TcpListener::bind(("127.0.0.1", port)).await.unwrap(); + let (stream, _) = tcp.accept().await.unwrap(); + let ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + let (mut write, mut read) = ws.split(); + + let hello = serde_json::to_string(&Message::ClientHello( + sendspin::protocol::messages::ClientHello { + client_id: "fake-embedded-client".to_string(), + name: "Fake Embedded Client".to_string(), + version: 1, + supported_roles: vec!["player@v1".to_string()], + device_info: None, + player_v1_support: None, + artwork_v1_support: None, + visualizer_v1_support: None, + }, + )) + .unwrap(); + write.send(WsMessage::Text(hello.into())).await.unwrap(); + + let msg = read.next().await.expect("no server/hello").unwrap(); + match msg { + WsMessage::Text(text) => { + serde_json::from_str(&text).expect("server/hello must deserialize") + } + other => panic!("expected text server/hello, got {other:?}"), + } +} + +#[tokio::test] +async fn discovers_and_dials_a_self_advertising_client() { + if !common::net_tests_enabled() { + eprintln!("skipping: set SENDSPIN_NET_TESTS=1 to run mDNS multicast tests"); + return; + } + // Real mDNS multicast timing (see tests/common/mod.rs) — retry rather + // than let one transient hiccup fail CI. + common::retry_flaky(3, discovers_and_dials_a_self_advertising_client_impl).await; +} + +async fn discovers_and_dials_a_self_advertising_client_impl() { + // A free ephemeral port, then bind our fake client to it directly (not + // through the OS's "any free port" allocator) so the mDNS advertisement + // below can name it explicitly, matching how a real embedded device + // knows its own listening port. + let port = { + let probe = TcpListener::bind("127.0.0.1:0").await.unwrap(); + probe.local_addr().unwrap().port() + }; + + // Instance name includes the port so a retried attempt (after a + // previous one panicked, leaving its ServiceDaemon's advertisement + // registered — ServiceDaemon has no Drop impl to unregister it for us) + // never collides with a still-lingering advertisement from an earlier + // attempt in the same test run. + let instance_name = format!("fake-embedded-client-{port}"); + let advertise_daemon = mdns_sd::ServiceDaemon::new().expect("mdns daemon"); + let service = ServiceInfo::new( + "_sendspin._tcp.local.", + &instance_name, + &format!("{instance_name}.local."), + "", + port, + &[("path", "/sendspin")][..], + ) + .unwrap() + .enable_addr_auto(); + advertise_daemon.register(service).expect("register"); + + let client_task = tokio::spawn(run_fake_embedded_client(port)); + + // The test host's LAN can (and, per direct observation while writing + // this test, does) have *real* Sendspin devices also advertising + // `_sendspin._tcp.local.` — so this loops past whatever else it sees + // rather than assuming the first resolved client is ours. + let browser = ClientBrowser::new().expect("browser"); + let url = timeout(Duration::from_secs(15), async { + loop { + let url = browser.next_client_url().await?; + if url.contains(&port.to_string()) { + return Some(url); + } + } + }) + .await + .expect("discovery timed out") + .expect("browser channel closed without finding our fake client"); + assert!( + url.contains(&port.to_string()) && url.ends_with("/sendspin"), + "unexpected discovered URL: {url}" + ); + + let conn = dial_client( + &url, + "test-dial-server", + "Test Dial Server", + Arc::new(DefaultClock::default()), + ) + .await + .expect("dial_client failed"); + assert_eq!(conn.client_id(), "fake-embedded-client"); + assert_eq!(conn.active_roles(), ["player@v1".to_string()]); + + let server_hello = timeout(Duration::from_secs(5), client_task) + .await + .expect("fake client task timed out") + .expect("fake client task panicked"); + match server_hello { + Message::ServerHello(hello) => { + assert_eq!(hello.server_id, "test-dial-server"); + assert_eq!(hello.active_roles, vec!["player@v1".to_string()]); + } + other => panic!("expected server/hello, got {other:?}"), + } + + let _ = advertise_daemon.shutdown(); +} diff --git a/tests/discovery.rs b/tests/discovery.rs new file mode 100644 index 0000000..8158bbf --- /dev/null +++ b/tests/discovery.rs @@ -0,0 +1,64 @@ +// ABOUTME: Integration test for the server role's mDNS advertisement — +// ABOUTME: proves a real browser resolves the service with the right port/path. + +mod common; + +use mdns_sd::{ServiceDaemon, ServiceEvent}; +use sendspin::server::Advertisement; +use std::time::Duration; + +#[test] +fn advertised_service_resolves_with_expected_port_and_path() { + if !common::net_tests_enabled() { + eprintln!("skipping: set SENDSPIN_NET_TESTS=1 to run mDNS multicast tests"); + return; + } + // Real mDNS multicast timing (see tests/common/mod.rs) — retry rather + // than let one transient hiccup fail CI. + common::retry_flaky_sync( + 3, + advertised_service_resolves_with_expected_port_and_path_impl, + ); +} + +fn advertised_service_resolves_with_expected_port_and_path_impl() { + let ad = Advertisement::new( + "test-discovery-server", + "Test Discovery Server", + 18999, + "/sendspin", + ) + .expect("advertise"); + + let browser = ServiceDaemon::new().expect("browser daemon"); + let receiver = browser + .browse("_sendspin-server._tcp.local.") + .expect("browse"); + + let deadline = std::time::Instant::now() + Duration::from_secs(10); + let mut resolved = None; + while std::time::Instant::now() < deadline { + if let Ok(ServiceEvent::ServiceResolved(info)) = + receiver.recv_timeout(Duration::from_secs(1)) + { + if info.get_port() == 18999 { + resolved = Some(info); + break; + } + } + } + drop(ad); + let _ = browser.shutdown(); + + let info = resolved.expect("service was never resolved via mDNS within 10s"); + assert_eq!(info.get_port(), 18999); + assert_eq!( + info.get_property_val_str("path"), + Some("/sendspin"), + "path TXT record must match what clients need to connect to" + ); + assert_eq!( + info.get_property_val_str("name"), + Some("Test Discovery Server") + ); +} diff --git a/tests/group_sync.rs b/tests/group_sync.rs new file mode 100644 index 0000000..416d518 --- /dev/null +++ b/tests/group_sync.rs @@ -0,0 +1,319 @@ +// ABOUTME: Integration tests for Group — proves the actual multi-room +// ABOUTME: correctness property: every member receives the identical audio +// ABOUTME: bytes tagged with the identical timestamp for a given push. + +use futures_util::{SinkExt, StreamExt}; +use sendspin::protocol::client::AudioChunk; +use sendspin::protocol::messages::{ + ClientHello, Message, PlayerCommand, PlayerCommandType, StreamPlayerConfig, +}; +use sendspin::server::Group; +use sendspin::ServerListener; +use std::time::Duration; +use tokio::time::timeout; +use tokio_tungstenite::{connect_async, tungstenite::Message as WsMessage}; + +fn test_hello(client_id: &str) -> ClientHello { + ClientHello { + client_id: client_id.to_string(), + name: "Test Player".to_string(), + version: 1, + supported_roles: vec!["player@v1".to_string()], + device_info: None, + player_v1_support: None, + artwork_v1_support: None, + visualizer_v1_support: None, + } +} + +/// Connects a bare peer that plays the client role manually: sends +/// client/hello, discards server/hello, then hands back the read half so the +/// test can assert on whatever the server sends next. +async fn connect_peer( + url: &str, + client_id: &str, +) -> futures_util::stream::SplitStream< + tokio_tungstenite::WebSocketStream>, +> { + let (ws, _) = connect_async(url).await.expect("ws connect"); + let (mut write, mut read) = ws.split(); + let hello = serde_json::to_string(&Message::ClientHello(test_hello(client_id))).unwrap(); + write.send(WsMessage::Text(hello.into())).await.unwrap(); + read.next().await.expect("no server/hello").unwrap(); // discard server/hello + read +} + +#[tokio::test] +async fn two_members_receive_identical_timestamped_audio() { + let listener = ServerListener::bind("127.0.0.1:0", "test-server", "Test Server") + .await + .expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + let url = format!("ws://{addr}"); + + let peer_a = tokio::spawn({ + let url = url.clone(); + async move { connect_peer(&url, "member-a").await } + }); + let peer_b = tokio::spawn({ + let url = url.clone(); + async move { connect_peer(&url, "member-b").await } + }); + + let (conn_a, _) = timeout(Duration::from_secs(5), listener.accept()) + .await + .unwrap() + .unwrap(); + let (conn_b, _) = timeout(Duration::from_secs(5), listener.accept()) + .await + .unwrap() + .unwrap(); + + let group = Group::new(std::sync::Arc::new(sendspin::DefaultClock::default())); + group + .add_member(conn_a.client_id().to_string(), conn_a.sender()) + .await + .unwrap(); + group + .add_member(conn_b.client_id().to_string(), conn_b.sender()) + .await + .unwrap(); + assert_eq!(group.len(), 2); + + group + .start_stream(StreamPlayerConfig { + codec: "pcm".to_string(), + sample_rate: 48000, + channels: 2, + bit_depth: 16, + codec_header: None, + }) + .await; + let sent_timestamp = group.push_audio(&[1, 2, 3, 4, 5, 6, 7, 8]); + + let mut read_a = peer_a.await.unwrap(); + let mut read_b = peer_b.await.unwrap(); + + // stream/start reaches both. + for read in [&mut read_a, &mut read_b] { + let msg = read.next().await.expect("no stream/start").unwrap(); + assert!(matches!(msg, WsMessage::Text(_))); + } + + // Both members get the *same* timestamp and the *same* bytes for this push — + // the actual multi-room correctness property. + let frame_a = match read_a.next().await.expect("no audio frame A").unwrap() { + WsMessage::Binary(b) => b, + other => panic!("expected binary, got {other:?}"), + }; + let frame_b = match read_b.next().await.expect("no audio frame B").unwrap() { + WsMessage::Binary(b) => b, + other => panic!("expected binary, got {other:?}"), + }; + let chunk_a = AudioChunk::from_bytes(&frame_a).unwrap(); + let chunk_b = AudioChunk::from_bytes(&frame_b).unwrap(); + assert_eq!(chunk_a.timestamp, sent_timestamp); + assert_eq!(chunk_b.timestamp, sent_timestamp); + assert_eq!(&*chunk_a.data, &*chunk_b.data); + assert_eq!(&*chunk_a.data, &[1, 2, 3, 4, 5, 6, 7, 8][..]); +} + +/// The scope decision for late joiners (see src/server/mod.rs): no +/// historical replay — a member that joins mid-stream gets stream/start +/// with the current config and then whatever the *next* push_audio() call +/// sends, in sync with everyone already in the group, but nothing from +/// before it joined. This proves that's actually what happens, not just +/// what the doc comment claims. +#[tokio::test] +async fn a_late_joiner_gets_current_stream_start_and_only_subsequent_audio() { + let listener = ServerListener::bind("127.0.0.1:0", "test-server", "Test Server") + .await + .expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + let url = format!("ws://{addr}"); + + let peer_a = tokio::spawn({ + let url = url.clone(); + async move { connect_peer(&url, "early-member").await } + }); + let (conn_a, _) = timeout(Duration::from_secs(5), listener.accept()) + .await + .unwrap() + .unwrap(); + let group = Group::new(std::sync::Arc::new(sendspin::DefaultClock::default())); + group + .add_member(conn_a.client_id().to_string(), conn_a.sender()) + .await + .unwrap(); + + group + .start_stream(StreamPlayerConfig { + codec: "pcm".to_string(), + sample_rate: 48000, + channels: 2, + bit_depth: 16, + codec_header: None, + }) + .await; + // Sent before the late joiner exists — it must never see this. + group.push_audio(&[0xAA; 8]); + + let mut read_a = peer_a.await.unwrap(); + assert!(matches!( + read_a + .next() + .await + .expect("no stream/start for early member") + .unwrap(), + WsMessage::Text(_) + )); + assert!(matches!( + read_a + .next() + .await + .expect("no first chunk for early member") + .unwrap(), + WsMessage::Binary(_) + )); + + // Now the late joiner connects, after a stream is already live. + let peer_b = tokio::spawn({ + let url = url.clone(); + async move { connect_peer(&url, "late-member").await } + }); + let (conn_b, _) = timeout(Duration::from_secs(5), listener.accept()) + .await + .unwrap() + .unwrap(); + group + .add_member(conn_b.client_id().to_string(), conn_b.sender()) + .await + .unwrap(); + let mut read_b = peer_b.await.unwrap(); + + // It gets stream/start with the live config, immediately (not deferred + // until some future push_audio call). + let start_msg = match timeout(Duration::from_secs(5), read_b.next()) + .await + .expect("timed out waiting for late joiner's stream/start") + .unwrap() + .unwrap() + { + WsMessage::Text(t) => t, + other => panic!("expected text stream/start, got {other:?}"), + }; + match serde_json::from_str::(&start_msg).unwrap() { + Message::StreamStart(s) => { + let player = s.player.expect("expected player config"); + assert_eq!(player.sample_rate, 48000); + } + other => panic!("expected stream/start, got {other:?}"), + } + + // The next push reaches both members with the identical timestamp/bytes + // — the late joiner is synchronized with the one that was already there. + let sent_timestamp = group.push_audio(&[0xBB; 8]); + + let frame_a = match read_a.next().await.expect("no second chunk for A").unwrap() { + WsMessage::Binary(b) => b, + other => panic!("expected binary, got {other:?}"), + }; + let frame_b = match read_b + .next() + .await + .expect("no chunk for late joiner") + .unwrap() + { + WsMessage::Binary(b) => b, + other => panic!("expected binary, got {other:?}"), + }; + let chunk_a = AudioChunk::from_bytes(&frame_a).unwrap(); + let chunk_b = AudioChunk::from_bytes(&frame_b).unwrap(); + assert_eq!(chunk_a.timestamp, sent_timestamp); + assert_eq!(chunk_b.timestamp, sent_timestamp); + assert_eq!(&*chunk_b.data, &[0xBB; 8][..]); + assert_eq!( + &*chunk_a.data, &*chunk_b.data, + "late joiner must be in sync with the member that was already there" + ); + + // And it never received the chunk sent before it joined. + assert_ne!(&*chunk_b.data, &[0xAA; 8][..]); +} + +#[tokio::test] +async fn a_dead_member_is_pruned_without_blocking_the_survivor() { + let listener = ServerListener::bind("127.0.0.1:0", "test-server", "Test Server") + .await + .expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + let url = format!("ws://{addr}"); + + let peer_a = tokio::spawn({ + let url = url.clone(); + async move { connect_peer(&url, "member-a").await } + }); + let peer_b = tokio::spawn({ + let url = url.clone(); + async move { connect_peer(&url, "member-b").await } + }); + + let (conn_a, _) = timeout(Duration::from_secs(5), listener.accept()) + .await + .unwrap() + .unwrap(); + let (conn_b, _) = timeout(Duration::from_secs(5), listener.accept()) + .await + .unwrap() + .unwrap(); + + let group = Group::new(std::sync::Arc::new(sendspin::DefaultClock::default())); + group + .add_member(conn_a.client_id().to_string(), conn_a.sender()) + .await + .unwrap(); + group + .add_member(conn_b.client_id().to_string(), conn_b.sender()) + .await + .unwrap(); + + // Kill member A's connection outright (not a clean disconnect) — the + // survivor must still get its command. + conn_a.disconnect().await.unwrap(); + let mut read_a = peer_a.await.unwrap(); + let mut read_b = peer_b.await.unwrap(); + // Drain member A's own close frame so its stream doesn't matter further. + let _ = timeout(Duration::from_secs(2), read_a.next()).await; + + group + .send_player_command(PlayerCommand { + command: PlayerCommandType::Volume, + volume: Some(77), + mute: None, + static_delay_ms: None, + }) + .await; + + assert_eq!( + group.member_ids(), + vec!["member-b".to_string()], + "member-a's failed send must have pruned it from the group" + ); + + let msg = timeout(Duration::from_secs(5), read_b.next()) + .await + .expect("timed out waiting for server/command") + .expect("no message") + .unwrap(); + let text = match msg { + WsMessage::Text(t) => t, + other => panic!("expected text, got {other:?}"), + }; + match serde_json::from_str::(&text).unwrap() { + Message::ServerCommand(cmd) => { + let player = cmd.player.expect("expected player command"); + assert_eq!(player.volume, Some(77)); + } + other => panic!("expected server/command, got {other:?}"), + } +} diff --git a/tests/server_listener.rs b/tests/server_listener.rs new file mode 100644 index 0000000..eb2dcda --- /dev/null +++ b/tests/server_listener.rs @@ -0,0 +1,222 @@ +// ABOUTME: Integration tests for the server role's ServerListener — inbound +// ABOUTME: WebSocket acceptor that drives the protocol-server state machine. + +use futures_util::{SinkExt, StreamExt}; +use sendspin::protocol::messages::{ + ClientHello, ClientState, ClientSyncState, ClientTime, Message, StreamPlayerConfig, +}; +use sendspin::ServerListener; +use std::time::Duration; +use tokio::time::timeout; +use tokio_tungstenite::{connect_async, tungstenite::Message as WsMessage}; + +fn test_hello(client_id: &str) -> ClientHello { + ClientHello { + client_id: client_id.to_string(), + name: "Test Player".to_string(), + version: 1, + supported_roles: vec!["player@v1".to_string()], + device_info: None, + player_v1_support: None, + artwork_v1_support: None, + visualizer_v1_support: None, + } +} + +#[tokio::test] +async fn accept_drives_handshake_and_grants_player_role() { + let listener = ServerListener::bind("127.0.0.1:0", "test-server", "Test Server") + .await + .expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + let url = format!("ws://{addr}"); + + let peer = tokio::spawn(async move { + let (ws, _) = connect_async(&url).await.expect("ws connect"); + let (mut write, mut read) = ws.split(); + + let hello = serde_json::to_string(&Message::ClientHello(test_hello("peer-1"))).unwrap(); + write.send(WsMessage::Text(hello.into())).await.unwrap(); + + let text = match read.next().await.expect("no server/hello").unwrap() { + WsMessage::Text(t) => t, + other => panic!("expected text server/hello, got {other:?}"), + }; + match serde_json::from_str::(&text).expect("server/hello must deserialize") { + Message::ServerHello(hello) => hello, + other => panic!("expected server/hello, got {other:?}"), + } + }); + + let (conn, peer_addr) = timeout(Duration::from_secs(5), listener.accept()) + .await + .expect("accept timed out") + .expect("accept failed"); + assert!(peer_addr.ip().is_loopback()); + assert_eq!(conn.client_id(), "peer-1"); + assert_eq!(conn.active_roles(), ["player@v1".to_string()]); + + let server_hello = peer.await.expect("peer task panicked"); + assert_eq!(server_hello.server_id, "test-server"); + assert_eq!(server_hello.active_roles, vec!["player@v1".to_string()]); +} + +#[tokio::test] +async fn client_without_player_role_gets_no_active_roles() { + let listener = ServerListener::bind("127.0.0.1:0", "test-server", "Test Server") + .await + .expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + let url = format!("ws://{addr}"); + + let peer = tokio::spawn(async move { + let (ws, _) = connect_async(&url).await.expect("ws connect"); + let (mut write, mut read) = ws.split(); + let mut hello = test_hello("peer-2"); + hello.supported_roles = vec!["controller@v1".to_string()]; + let hello = serde_json::to_string(&Message::ClientHello(hello)).unwrap(); + write.send(WsMessage::Text(hello.into())).await.unwrap(); + read.next().await.expect("no server/hello").unwrap() + }); + + let (conn, _) = timeout(Duration::from_secs(5), listener.accept()) + .await + .unwrap() + .unwrap(); + assert!(conn.active_roles().is_empty()); + peer.await.unwrap(); +} + +#[tokio::test] +async fn time_sync_echo_reflects_client_transmitted_and_orders_timestamps() { + let listener = ServerListener::bind("127.0.0.1:0", "test-server", "Test Server") + .await + .expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + let url = format!("ws://{addr}"); + + let peer = tokio::spawn(async move { + let (ws, _) = connect_async(&url).await.expect("ws connect"); + let (mut write, mut read) = ws.split(); + let hello = serde_json::to_string(&Message::ClientHello(test_hello("peer-3"))).unwrap(); + write.send(WsMessage::Text(hello.into())).await.unwrap(); + read.next().await.expect("no server/hello").unwrap(); // discard server/hello + + let client_transmitted = 1_000_000i64; + let msg = + serde_json::to_string(&Message::ClientTime(ClientTime { client_transmitted })).unwrap(); + write.send(WsMessage::Text(msg.into())).await.unwrap(); + + let text = match read.next().await.expect("no server/time").unwrap() { + WsMessage::Text(t) => t, + other => panic!("expected text server/time, got {other:?}"), + }; + match serde_json::from_str::(&text).expect("server/time must deserialize") { + Message::ServerTime(st) => st, + other => panic!("expected server/time, got {other:?}"), + } + }); + + let (_conn, _) = timeout(Duration::from_secs(5), listener.accept()) + .await + .unwrap() + .unwrap(); + + let server_time = timeout(Duration::from_secs(5), peer) + .await + .expect("peer task timed out") + .expect("peer task panicked"); + + assert_eq!(server_time.client_transmitted, 1_000_000); + assert!( + server_time.server_transmitted >= server_time.server_received, + "server_transmitted ({}) must not precede server_received ({}) — it's stamped \ + immediately before the reply goes on the wire, strictly after receipt", + server_time.server_transmitted, + server_time.server_received + ); +} + +#[tokio::test] +async fn pushed_audio_and_stream_lifecycle_reach_the_client_intact() { + let listener = ServerListener::bind("127.0.0.1:0", "test-server", "Test Server") + .await + .expect("bind"); + let addr = listener.local_addr().expect("local_addr"); + let url = format!("ws://{addr}"); + + let peer = tokio::spawn(async move { + let (ws, _) = connect_async(&url).await.expect("ws connect"); + let (mut write, mut read) = ws.split(); + let hello = serde_json::to_string(&Message::ClientHello(test_hello("peer-4"))).unwrap(); + write.send(WsMessage::Text(hello.into())).await.unwrap(); + read.next().await.expect("no server/hello").unwrap(); + + let state = serde_json::to_string(&Message::ClientState(ClientState { + state: Some(ClientSyncState::Synchronized), + player: None, + })) + .unwrap(); + write.send(WsMessage::Text(state.into())).await.unwrap(); + + // stream/start + let text = match read.next().await.expect("no stream/start").unwrap() { + WsMessage::Text(t) => t, + other => panic!("expected text stream/start, got {other:?}"), + }; + assert!(matches!( + serde_json::from_str::(&text).unwrap(), + Message::StreamStart(_) + )); + + // one binary audio frame + let frame = match read.next().await.expect("no audio frame").unwrap() { + WsMessage::Binary(b) => b, + other => panic!("expected binary audio frame, got {other:?}"), + }; + let chunk = sendspin::protocol::client::AudioChunk::from_bytes(&frame) + .expect("must parse as a player audio chunk"); + + // stream/end + let text = match read.next().await.expect("no stream/end").unwrap() { + WsMessage::Text(t) => t, + other => panic!("expected text stream/end, got {other:?}"), + }; + assert!(matches!( + serde_json::from_str::(&text).unwrap(), + Message::StreamEnd(_) + )); + + chunk + }); + + let (conn, _) = timeout(Duration::from_secs(5), listener.accept()) + .await + .unwrap() + .unwrap(); + let sender = conn.sender(); + + sender + .send_stream_start(StreamPlayerConfig { + codec: "pcm".to_string(), + sample_rate: 44100, + channels: 2, + bit_depth: 16, + codec_header: None, + }) + .await + .expect("send_stream_start"); + let payload = vec![1u8, 2, 3, 4, 5, 6, 7, 8]; + sender + .send_audio_chunk(42_000, &payload) + .await + .expect("send_audio_chunk"); + sender.send_stream_end().await.expect("send_stream_end"); + + let chunk = timeout(Duration::from_secs(5), peer) + .await + .expect("peer task timed out") + .expect("peer task panicked"); + assert_eq!(chunk.timestamp, 42_000); + assert_eq!(&*chunk.data, &payload[..]); +}