feat: 支持动态切换主从播放模式

apps/hello/src/stereo_core/master.rs

@@ -7,15 +7,23 @@ use crate::stereo_core::discovery::Discovery;
 use crate::stereo_core::network::{ControlConnection, MasterNetwork};
 use crate::stereo_core::protocol::{AudioPacket, ChannelRole, ControlPacket};
 use crate::stereo_core::sync::now_us;
-use anyhow::Result;
+use anyhow::{Result, anyhow};
+use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::sync::Mutex;
 
 pub const SERVER_TCP_PORT: u16 = 53531;
 
-pub async fn run_master(role: ChannelRole) -> Result<()> {
-    println!("--- 主节点模式 ({}) ---", role.to_string());
+#[derive(Clone)]
+struct SlaveSession {
+    udp_addr: SocketAddr,
+    role: ChannelRole,
+}
+
+pub async fn run_master(master_role: ChannelRole) -> Result<()> {
+    println!("--- 主节点模式 ({}) ---", master_role.to_string());
 
     // 0. 设置 ALSA 重定向
     let _alsa_guard = AlsaRedirector::new()?;
@@ -29,21 +37,180 @@ pub async fn run_master(role: ChannelRole) -> Result<()> {
 
     println!("✅ 服务已启动，等待连接...");
 
-    loop {
-        let (control_conn, client_addr) = network.accept().await?;
+    let slaves = Arc::new(Mutex::new(Vec::<SlaveSession>::new()));
 
-        let audio_socket = audio_socket.clone();
-        let role = role.clone();
-
-        // 启动会话处理句柄
-        tokio::spawn(async move {
-            if let Err(e) =
-                handle_master_session(control_conn, audio_socket, role, client_addr.to_string())
-                    .await
-            {
-                eprintln!("❌ {:?}", e);
+    // 3. 启动连接监听任务
+    let slaves_clone = slaves.clone();
+    let audio_socket_clone = audio_socket.clone();
+    tokio::spawn(async move {
+        loop {
+            match network.accept().await {
+                Ok((control_conn, client_addr)) => {
+                    let slaves_for_session = slaves_clone.clone();
+                    let audio_socket_for_session = audio_socket_clone.clone();
+                    tokio::spawn(async move {
+                        if let Err(e) = handle_master_session(
+                            control_conn,
+                            audio_socket_for_session,
+                            slaves_for_session,
+                            client_addr.to_string(),
+                        )
+                        .await
+                        {
+                            eprintln!("❌ 会话错误: {:?}", e);
+                        }
+                    });
+                }
+                Err(e) => {
+                    eprintln!("❌ Accept 错误: {:?}", e);
+                }
             }
-        });
+        }
+    });
+
+    // 4. 音频处理主循环
+    let config = AudioConfig {
+        sample_rate: 48000,
+        channels: 2,
+        frame_size: 960,
+        bitrate: 64000,
+        ..AudioConfig::default()
+    };
+
+    let mut mono_codec = OpusCodec::new(&AudioConfig {
+        channels: 1,
+        ..config.clone()
+    })?;
+
+    let mut current_player_channels = 0;
+    let mut player: Option<AudioPlayer> = None;
+
+    let mut raw_buf = vec![0u8; config.frame_size * 2 * 2];
+    let mut opus_out = vec![0u8; 1500];
+    let mut seq = 0u32;
+
+    let delay_us = 200_000;
+    let frame_duration_us =
+        (config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128;
+
+    let mut stream_start_ts = 0;
+    let mut stream_start_seq = 0;
+
+    loop {
+        // 打开 FIFO
+        let mut fifo = match tokio::fs::File::open(AlsaRedirector::fifo_path()).await {
+            Ok(f) => f,
+            Err(e) => {
+                eprintln!("❌ 无法打开 FIFO: {:?}, 重试...", e);
+                tokio::time::sleep(Duration::from_secs(1)).await;
+                continue;
+            }
+        };
+
+        loop {
+            // 从 FIFO 读取
+            if let Err(_) = fifo.read_exact(&mut raw_buf).await {
+                break; // FIFO 关闭，重新打开
+            }
+
+            let active_slaves = slaves.lock().await.clone();
+
+            // 检查是否需要切换播放器模式
+            let target_channels = if active_slaves.is_empty() { 2 } else { 1 };
+            if player.is_none() || current_player_channels != target_channels {
+                println!(
+                    "🔄 切换播放模式: {}",
+                    if target_channels == 2 {
+                        "本地立体声"
+                    } else {
+                        "主从同步 (单声道)"
+                    }
+                );
+                let playback_config = AudioConfig {
+                    channels: target_channels,
+                    playback_device: "plug:original_default".into(),
+                    ..config.clone()
+                };
+                player = Some(AudioPlayer::new(&playback_config)?);
+                current_player_channels = target_channels;
+            }
+
+            let now = now_us();
+            if stream_start_ts == 0 {
+                stream_start_ts = now;
+                stream_start_seq = seq;
+            }
+
+            if active_slaves.is_empty() {
+                // 情况 1: 没有从节点，本地立体声播放
+                let mut pcm = Vec::with_capacity(config.frame_size * 2);
+                for i in 0..config.frame_size {
+                    let l = i16::from_le_bytes([raw_buf[i * 4], raw_buf[i * 4 + 1]]);
+                    let r = i16::from_le_bytes([raw_buf[i * 4 + 2], raw_buf[i * 4 + 3]]);
+                    pcm.push(l);
+                    pcm.push(r);
+                }
+                if let Some(p) = &player {
+                    p.write(&pcm)?;
+                }
+            } else {
+                // 情况 2: 有从节点，主从同步
+                let mut local_pcm = Vec::with_capacity(config.frame_size);
+                let mut remote_pcm = Vec::with_capacity(config.frame_size);
+
+                // 提取左右声道 (假设当前逻辑只处理一个从节点的情况，或所有从节点角色一致)
+                // 如果有多个从节点角色不同，这里需要更复杂的逻辑
+                let slave_role = active_slaves[0].role;
+
+                for i in 0..config.frame_size {
+                    let l = i16::from_le_bytes([raw_buf[i * 4], raw_buf[i * 4 + 1]]);
+                    let r = i16::from_le_bytes([raw_buf[i * 4 + 2], raw_buf[i * 4 + 3]]);
+                    if master_role == ChannelRole::Left {
+                        local_pcm.push(l);
+                    } else {
+                        local_pcm.push(r);
+                    }
+                    if slave_role == ChannelRole::Left {
+                        remote_pcm.push(l);
+                    } else {
+                        remote_pcm.push(r);
+                    }
+                }
+
+                // 编码并发送给所有从节点
+                let len = mono_codec.encode(&remote_pcm, &mut opus_out)?;
+                let target_ts = stream_start_ts
+                    + ((seq - stream_start_seq) as u128 * frame_duration_us)
+                    + delay_us;
+
+                let packet = AudioPacket {
+                    seq,
+                    timestamp: target_ts,
+                    data: opus_out[..len].to_vec(),
+                };
+
+                let bytes = postcard::to_allocvec(&packet)?;
+                for slave in &active_slaves {
+                    let _ = audio_socket.send_to(&bytes, slave.udp_addr).await;
+                }
+
+                // 本地回放同步
+                let now = now_us();
+                if now < target_ts {
+                    let wait = target_ts - now;
+                    if wait > 1000 {
+                        tokio::time::sleep(Duration::from_micros(wait as u64)).await;
+                    }
+                }
+                if let Some(p) = &player {
+                    p.write(&local_pcm)?;
+                }
+            }
+
+            seq += 1;
+        }
+        // 重置流计时
+        stream_start_ts = 0;
     }
 }
 
@@ -51,19 +218,16 @@ pub async fn run_master(role: ChannelRole) -> Result<()> {
 async fn handle_master_session(
     mut control: ControlConnection,
     audio_socket: Arc<tokio::net::UdpSocket>,
-    master_role: ChannelRole,
+    slaves: Arc<Mutex<Vec<SlaveSession>>>,
     client_tcp_addr: String,
 ) -> Result<()> {
     let mut buf = [0u8; 1024];
 
-    #[allow(unused_assignments)]
-    let mut slave_role = ChannelRole::Left;
-
     // 握手
     let pkt = control.recv_packet(&mut buf).await?;
-    match pkt {
-        ControlPacket::ClientIdentify { role } => slave_role = role,
-        _ => return Err(anyhow::anyhow!("无效的握手协议")),
+    let slave_role = match pkt {
+        ControlPacket::ClientIdentify { role } => role,
+        _ => return Err(anyhow!("无效的握手协议")),
     };
 
     let hello = ControlPacket::ServerHello {
@@ -81,149 +245,56 @@ async fn handle_master_session(
         slave_role.to_string(),
     );
 
-    // 配置音频
-    let config = AudioConfig {
-        sample_rate: 48000,
-        channels: 2,
-        frame_size: 960,
-        bitrate: 64000,
-        ..AudioConfig::default()
+    // 添加到从节点列表
+    let session = SlaveSession {
+        udp_addr: client_udp_addr,
+        role: slave_role,
     };
+    {
+        let mut s = slaves.lock().await;
+        s.push(session.clone());
+    }
 
-    let mono_config = AudioConfig {
-        channels: 1,
-        ..config.clone()
-    };
-    let mut codec = OpusCodec::new(&mono_config)?;
-    let playback_config = AudioConfig {
-        channels: 1,
-        playback_device: "plug:original_default".into(),
-        ..config.clone()
-    };
-    let player = AudioPlayer::new(&playback_config)?;
-
-    let mut raw_buf = vec![0u8; config.frame_size * 2 * 2];
-    let mut opus_out = vec![0u8; 1500];
-    let mut seq = 0u32;
-
-    let delay_us = 200_000; // 200ms 延迟，确保主从节点音频同步
-    let frame_duration_us =
-        (config.frame_size as f64 / config.sample_rate as f64 * 1_000_000.0) as u128;
-
-    // 分离 TCP 读写，以便在不同任务中使用
+    // 分离 TCP 读写，处理控制消息和心跳
     let (mut tcp_rx, mut tcp_tx) = control.split();
-    let (stop_tx, mut stop_rx) = tokio::sync::mpsc::channel::<()>(1);
 
-    // 处理来自从节点的控制消息
-    tokio::spawn(async move {
-        let mut buf = [0u8; 1024];
-        loop {
-            match tcp_rx.read(&mut buf).await {
-                Ok(0) | Err(_) => {
-                    let _ = stop_tx.send(()).await;
-                    println!(
-                        "❌ 从节点已断开: {} {}",
-                        client_tcp_addr,
-                        slave_role.to_string(),
-                    );
-                    break;
-                }
-                Ok(n) => {
-                    if let Ok(ControlPacket::Ping { client_ts, seq }) =
-                        postcard::from_bytes(&buf[..n])
+    let mut buf = [0u8; 1024];
+    loop {
+        match tcp_rx.read(&mut buf).await {
+            Ok(0) | Err(_) => {
+                break;
+            }
+            Ok(n) => {
+                if let Ok(ControlPacket::Ping { client_ts, seq }) = postcard::from_bytes(&buf[..n])
+                {
+                    let pong = ControlPacket::Pong {
+                        client_ts,
+                        server_ts: now_us(),
+                        seq,
+                    };
+                    if tcp_tx
+                        .write_all(&postcard::to_allocvec(&pong).unwrap())
+                        .await
+                        .is_err()
                     {
-                        let pong = ControlPacket::Pong {
-                            client_ts,
-                            server_ts: now_us(),
-                            seq,
-                        };
-                        if tcp_tx
-                            .write_all(&postcard::to_allocvec(&pong).unwrap())
-                            .await
-                            .is_err()
-                        {
-                            let _ = stop_tx.send(()).await;
-                            break;
-                        }
+                        break;
                     }
                 }
             }
         }
-    });
-
-    loop {
-        // 打开 FIFO
-        let mut fifo = tokio::select! {
-            _ = stop_rx.recv() => return Ok(()),
-            f = tokio::fs::File::open(AlsaRedirector::fifo_path()) => f?,
-        };
-
-        let mut stream_start_ts = 0;
-        let stream_start_seq = seq;
-
-        loop {
-            // 从 FIFO 读取
-            let read_res = tokio::select! {
-                _ = stop_rx.recv() => return Ok(()),
-                res = fifo.read_exact(&mut raw_buf) => res,
-            };
-
-            if let Err(_) = read_res {
-                // todo 继续等待音频流继续播放
-                break;
-            }
-
-            let now = now_us();
-            if stream_start_ts == 0 {
-                stream_start_ts = now;
-            }
-
-            let mut local_pcm = Vec::with_capacity(config.frame_size);
-            let mut remote_pcm = Vec::with_capacity(config.frame_size);
-
-            // 提取左右声道
-            for i in 0..config.frame_size {
-                let l = i16::from_le_bytes([raw_buf[i * 4], raw_buf[i * 4 + 1]]);
-                let r = i16::from_le_bytes([raw_buf[i * 4 + 2], raw_buf[i * 4 + 3]]);
-                if master_role == ChannelRole::Left {
-                    local_pcm.push(l);
-                } else {
-                    local_pcm.push(r);
-                }
-                if slave_role == ChannelRole::Left {
-                    remote_pcm.push(l);
-                } else {
-                    remote_pcm.push(r);
-                }
-            }
-
-            // 编码并发送给从节点
-            let len = codec.encode(&remote_pcm, &mut opus_out)?;
-            let target_ts =
-                stream_start_ts + ((seq - stream_start_seq) as u128 * frame_duration_us) + delay_us;
-
-            let packet = AudioPacket {
-                seq,
-                timestamp: target_ts,
-                data: opus_out[..len].to_vec(),
-            };
-
-            let bytes = postcard::to_allocvec(&packet)?;
-            if let Err(e) = audio_socket.send_to(&bytes, client_udp_addr).await {
-                return Err(anyhow::anyhow!("UDP 发送错误: {:?}", e));
-            }
-
-            // 本地回放同步
-            let now = now_us();
-            if now < target_ts {
-                let wait = target_ts - now;
-                if wait > 1000 {
-                    tokio::time::sleep(Duration::from_micros(wait as u64)).await;
-                }
-            }
-            player.write(&local_pcm)?;
-
-            seq += 1;
-        }
     }
+
+    println!(
+        "❌ 从节点已断开: {} {}",
+        client_tcp_addr,
+        slave_role.to_string(),
+    );
+
+    // 从列表中移除
+    {
+        let mut s = slaves.lock().await;
+        s.retain(|x| x.udp_addr != client_udp_addr);
+    }
+
+    Ok(())
 }