feat: 立体声 client 构建

apps/hello/src/bin/client.rs

@@ -1,12 +1,120 @@
+use anyhow::Result;
 use hello::audio::{AudioPlayer, OpusCodec};
 use hello::config::AudioConfig;
-use anyhow::Result;
+use hello::net::{AudioPacket, ChannelRole, ControlPacket, Discovery};
+use hello::sync::{ClockSync, now_us};
+use std::collections::VecDeque;
+use std::env;
+use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+    let args: Vec<String> = env::args().collect();
+    if args.len() < 2 {
+        eprintln!("Usage: {} [left|right]", args[0]);
+        return Ok(());
+    }
+    let role = if args[1] == "left" {
+        ChannelRole::Left
+    } else {
+        ChannelRole::Right
+    };
+
+    let config = AudioConfig {
+        sample_rate: 48000,
+        channels: 1,
+        frame_size: 960,
+        ..AudioConfig::default()
+    };
+
+    println!("Searching for server...");
+    let server_addr = Discovery::client_discover_server().await?;
+    let mut stream = TcpStream::connect(server_addr).await?;
+
+    // 1. Identify
+    stream
+        .write_all(&postcard::to_allocvec(&ControlPacket::ClientIdentify {
+            role: role.clone(),
+        })?)
+        .await?;
+
+    // 2. Sync
+    let mut clock = ClockSync::new();
+    let t1 = now_us();
+    stream
+        .write_all(&postcard::to_allocvec(&ControlPacket::Ping {
+            client_ts: t1,
+        })?)
+        .await?;
+    let mut buf = [0u8; 1024];
+    let n = stream.read(&mut buf).await?;
+    if let Ok(ControlPacket::Pong {
+        client_ts,
+        server_ts,
+    }) = postcard::from_bytes::<ControlPacket>(&buf[..n])
+    {
+        let t4 = now_us();
+        clock.update(client_ts, server_ts, t4);
+        println!(
+            "Clock synced. Offset: {}us, RTT: {}us",
+            clock.offset,
+            t4 - t1
+        );
+    }
 
-// todo 左右声道分发，播放音频
-fn main() -> Result<()> {
-    let config = AudioConfig::default();
     let player = AudioPlayer::new(&config)?;
     let mut codec = OpusCodec::new(&config)?;
+    let mut jitter_buffer: VecDeque<AudioPacket> = VecDeque::new();
+    let mut pcm_buf = vec![0i16; config.frame_size];
 
-    
+    let (tx, mut rx) = tokio::sync::mpsc::channel(100);
+    tokio::spawn(async move {
+        loop {
+            let mut s_buf = [0u8; 4];
+            if stream.read_exact(&mut s_buf).await.is_err() {
+                break;
+            }
+            let size = u32::from_be_bytes(s_buf) as usize;
+            let mut data = vec![0u8; size];
+            if stream.read_exact(&mut data).await.is_err() {
+                break;
+            }
+            if let Ok(p) = postcard::from_bytes::<AudioPacket>(&data) {
+                let _ = tx.send(p).await;
+            }
+        }
+    });
+
+    println!("Streaming started. Playing channel {:?}", role);
+
+    loop {
+        while let Ok(p) = rx.try_recv() {
+            jitter_buffer.push_back(p);
+        }
+
+        if let Some(p) = jitter_buffer.front() {
+            let target_client_time = clock.to_client_time(p.timestamp);
+            let now = now_us();
+
+            if now >= target_client_time {
+                let packet = jitter_buffer.pop_front().unwrap();
+                let len = codec.decode(&packet.data, &mut pcm_buf)?;
+                player.write(&pcm_buf[..len])?;
+            } else if target_client_time - now > 500_000 {
+                // Too far in the future, maybe clock jumped?
+                jitter_buffer.pop_front();
+            } else {
+                // Wait until it's time
+                let wait = (target_client_time - now) as u64;
+                if wait > 1000 {
+                    tokio::time::sleep(Duration::from_micros(wait)).await;
+                }
+            }
+        } else {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+    }
 }

apps/hello/src/bin/server.rs

@@ -1,11 +1,234 @@
-use hello::audio::{OpusCodec};
+use anyhow::{Context, Result};
+use hello::audio::OpusCodec;
 use hello::config::AudioConfig;
-use anyhow::Result;
+use hello::net::{AudioPacket, ChannelRole, ControlPacket, Discovery, SERVER_PORT};
+use hello::sync::now_us;
+use std::collections::HashMap;
+use std::fs::File;
+use std::path::Path;
+use std::sync::Arc;
+use symphonia::core::audio::SampleBuffer;
+use symphonia::core::codecs::DecoderOptions;
+use symphonia::core::formats::FormatOptions;
+use symphonia::core::io::MediaSourceStream;
+use symphonia::core::meta::MetadataOptions;
+use symphonia::core::probe::Hint;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream};
+use tokio::sync::Mutex;
 
-// todo 向 client 分发左右声道音频数据
-fn main() -> Result<()> {
-    let config = AudioConfig::default();
-    let mut codec = OpusCodec::new(&config)?;
+#[tokio::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+    let config = AudioConfig {
+        sample_rate: 48000,
+        channels: 1,
+        frame_size: 960, // 20ms at 48kHz
+        ..AudioConfig::default()
+    };
 
-    
+    println!("Starting Stereo Server on port {}", SERVER_PORT);
+
+    Discovery::start_server_beacon().await?;
+
+    let listener = TcpListener::bind(format!("0.0.0.0:{}", SERVER_PORT)).await?;
+    let clients: Arc<Mutex<HashMap<ChannelRole, TcpStream>>> = Arc::new(Mutex::new(HashMap::new()));
+
+    let clients_clone = Arc::clone(&clients);
+    tokio::spawn(async move {
+        while let Ok((mut stream, addr)) = listener.accept().await {
+            let mut buf = [0u8; 1024];
+            if let Ok(n) = stream.read(&mut buf).await {
+                if let Ok(packet) = postcard::from_bytes::<ControlPacket>(&buf[..n]) {
+                    match packet {
+                        ControlPacket::ClientIdentify { role } => {
+                            println!("Client {} identified as {:?}", addr, role);
+                            // Simple handshake for sync
+                            let mut sync_buf = [0u8; 1024];
+                            if let Ok(sn) = stream.read(&mut sync_buf).await {
+                                if let Ok(ControlPacket::Ping { client_ts }) =
+                                    postcard::from_bytes::<ControlPacket>(&sync_buf[..sn])
+                                {
+                                    let pong = ControlPacket::Pong {
+                                        client_ts,
+                                        server_ts: now_us(),
+                                    };
+                                    let pong_msg = postcard::to_allocvec(&pong).unwrap();
+                                    let _ = stream.write_all(&pong_msg).await;
+                                    clients_clone.lock().await.insert(role, stream);
+                                }
+                            }
+                        }
+                        _ => {}
+                    }
+                }
+            }
+        }
+    });
+
+    println!("Waiting for Left and Right clients...");
+    loop {
+        if clients.lock().await.len() >= 2 {
+            break;
+        }
+        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
+    }
+    println!("Both clients connected. Starting stream in 1s...");
+    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+
+    stream_mp3("test.mp3", &config, clients).await?;
+
+    Ok(())
+}
+
+async fn stream_mp3(
+    path: &str,
+    config: &AudioConfig,
+    clients: Arc<Mutex<HashMap<ChannelRole, TcpStream>>>,
+) -> Result<()> {
+    let src = File::open(Path::new(path)).context("Failed to open test.mp3")?;
+    let mss = MediaSourceStream::new(Box::new(src), Default::default());
+    let probed = symphonia::default::get_probe().format(
+        &Hint::new(),
+        mss,
+        &FormatOptions::default(),
+        &MetadataOptions::default(),
+    )?;
+
+    let mut format = probed.format;
+    let track = format
+        .tracks()
+        .iter()
+        .find(|t| t.codec_params.codec == symphonia::core::codecs::CODEC_TYPE_MP3)
+        .context("No track")?;
+    let mut decoder =
+        symphonia::default::get_codecs().make(&track.codec_params, &DecoderOptions::default())?;
+    let track_id = track.id;
+
+    let mut left_codec = OpusCodec::new(config)?;
+    let mut right_codec = OpusCodec::new(config)?;
+
+    let start_time = now_us() + 500_000;
+    let mut current_frame_ts = start_time;
+    let frame_duration_us =
+        (config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128;
+
+    let mut sample_buf = None;
+
+    loop {
+        let packet = match format.next_packet() {
+            Ok(packet) => packet,
+            Err(_) => break,
+        };
+        if packet.track_id() != track_id {
+            continue;
+        }
+
+        let decoded = decoder.decode(&packet)?;
+        if sample_buf.is_none() {
+            sample_buf = Some(SampleBuffer::<i16>::new(
+                decoded.capacity() as u64,
+                *decoded.spec(),
+            ));
+        }
+
+        if let Some(buf) = sample_buf.as_mut() {
+            buf.copy_interleaved_ref(decoded.clone());
+            let spec = decoded.spec();
+            let num_channels = spec.channels.count();
+
+            if num_channels == 2 {
+                for chunk in buf.samples().chunks(config.frame_size * 2) {
+                    if chunk.len() < config.frame_size * 2 {
+                        break;
+                    }
+
+                    let mut left_pcm = vec![0i16; config.frame_size];
+                    let mut right_pcm = vec![0i16; config.frame_size];
+
+                    for i in 0..config.frame_size {
+                        left_pcm[i] = chunk[i * 2];
+                        right_pcm[i] = chunk[i * 2 + 1];
+                    }
+
+                    send_packets(
+                        &clients,
+                        &mut left_codec,
+                        &mut right_codec,
+                        &left_pcm,
+                        &right_pcm,
+                        current_frame_ts,
+                    )
+                    .await?;
+                    current_frame_ts += frame_duration_us;
+                    pace(current_frame_ts).await;
+                }
+            } else {
+                for chunk in buf.samples().chunks(config.frame_size) {
+                    if chunk.len() < config.frame_size {
+                        break;
+                    }
+
+                    send_packets(
+                        &clients,
+                        &mut left_codec,
+                        &mut right_codec,
+                        chunk,
+                        chunk,
+                        current_frame_ts,
+                    )
+                    .await?;
+                    current_frame_ts += frame_duration_us;
+                    pace(current_frame_ts).await;
+                }
+            }
+        }
+    }
+    Ok(())
+}
+
+async fn send_packets(
+    clients: &Arc<Mutex<HashMap<ChannelRole, TcpStream>>>,
+    left_codec: &mut OpusCodec,
+    right_codec: &mut OpusCodec,
+    left_pcm: &[i16],
+    right_pcm: &[i16],
+    ts: u128,
+) -> Result<()> {
+    let mut left_opus = vec![0u8; 1275];
+    let mut right_opus = vec![0u8; 1275];
+
+    let l_len = left_codec.encode(left_pcm, &mut left_opus)?;
+    let r_len = right_codec.encode(right_pcm, &mut right_opus)?;
+
+    let l_p = postcard::to_allocvec(&AudioPacket {
+        timestamp: ts,
+        data: left_opus[..l_len].to_vec(),
+    })?;
+    let r_p = postcard::to_allocvec(&AudioPacket {
+        timestamp: ts,
+        data: right_opus[..r_len].to_vec(),
+    })?;
+
+    let mut lock = clients.lock().await;
+    if let Some(s) = lock.get_mut(&ChannelRole::Left) {
+        let _ = send_with_size(s, &l_p).await;
+    }
+    if let Some(s) = lock.get_mut(&ChannelRole::Right) {
+        let _ = send_with_size(s, &r_p).await;
+    }
+    Ok(())
+}
+
+async fn pace(current_frame_ts: u128) {
+    let now = now_us();
+    if current_frame_ts > now + 1_000_000 {
+        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
+    }
+}
+
+async fn send_with_size(stream: &mut TcpStream, data: &[u8]) -> Result<()> {
+    stream.write_all(&(data.len() as u32).to_be_bytes()).await?;
+    stream.write_all(data).await?;
+    Ok(())
 }

apps/hello/src/net/discovery.rs

@@ -0,0 +1,45 @@
+use std::net::SocketAddr;
+use tokio::net::UdpSocket;
+use std::time::Duration;
+use crate::net::protocol::ControlPacket;
+use anyhow::Result;
+
+pub const DISCOVERY_PORT: u16 = 8888;
+pub const SERVER_PORT: u16 = 8889;
+
+pub struct Discovery;
+
+impl Discovery {
+    pub async fn start_server_beacon() -> Result<()> {
+        let socket = UdpSocket::bind("0.0.0.0:0").await?;
+        socket.set_broadcast(true)?;
+        
+        let target_addr: SocketAddr = format!("255.255.255.255:{}", DISCOVERY_PORT).parse()?;
+        let hello = ControlPacket::ServerHello { port: SERVER_PORT };
+        let msg = postcard::to_allocvec(&hello)?;
+
+        tokio::spawn(async move {
+            loop {
+                let _ = socket.send_to(&msg, target_addr).await;
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            }
+        });
+
+        Ok(())
+    }
+
+    pub async fn client_discover_server() -> Result<SocketAddr> {
+        let socket = UdpSocket::bind(format!("0.0.0.0:{}", DISCOVERY_PORT)).await?;
+        let mut buf = [0u8; 1024];
+
+        loop {
+            let (len, addr) = socket.recv_from(&mut buf).await?;
+            if let Ok(ControlPacket::ServerHello { port }) = postcard::from_bytes::<ControlPacket>(&buf[..len]) {
+                let mut server_addr = addr;
+                server_addr.set_port(port);
+                return Ok(server_addr);
+            }
+        }
+    }
+}
+

apps/hello/src/net/protocol.rs

@@ -0,0 +1,37 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, Hash)]
+pub enum ChannelRole {
+    Left,
+    Right,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub enum ControlPacket {
+    // Discovery
+    ServerHello {
+        port: u16,
+    },
+    // Handshake
+    ClientIdentify {
+        role: ChannelRole,
+    },
+    // Time Sync
+    Ping {
+        client_ts: u128,
+    },
+    Pong {
+        client_ts: u128,
+        server_ts: u128,
+    },
+    // Stream control
+    StartStreaming {
+        start_time: u128, // Server time when streaming starts
+    },
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct AudioPacket {
+    pub timestamp: u128, // Target server time for playback
+    pub data: Vec<u8>,   // Opus encoded data
+}

apps/hello/src/sync.rs

@@ -0,0 +1,34 @@
+use std::time::{SystemTime, UNIX_EPOCH};
+
+pub fn now_us() -> u128 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .expect("Time went backwards")
+        .as_micros()
+}
+
+pub struct ClockSync {
+    pub offset: i128, // server_time - client_time
+}
+
+impl ClockSync {
+    pub fn new() -> Self {
+        Self { offset: 0 }
+    }
+
+    pub fn update(&mut self, client_send_ts: u128, server_ts: u128, client_recv_ts: u128) {
+        let rtt = (client_recv_ts - client_send_ts) as i128;
+        // Estimated server time when client_recv_ts occurred: server_ts + rtt/2
+        let estimated_server_now = server_ts as i128 + rtt / 2;
+        self.offset = estimated_server_now - client_recv_ts as i128;
+    }
+
+    pub fn to_server_time(&self, client_time: u128) -> u128 {
+        (client_time as i128 + self.offset) as u128
+    }
+
+    pub fn to_client_time(&self, server_time: u128) -> u128 {
+        (server_time as i128 - self.offset) as u128
+    }
+}
+