diff --git a/apps/hello/src/stereo_core/master.rs b/apps/hello/src/stereo_core/master.rs index 21cba64..9d5ab3f 100644 --- a/apps/hello/src/stereo_core/master.rs +++ b/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::::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 = 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, - master_role: ChannelRole, + slaves: Arc>>, 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(()) }