From 7b473ad86a1913c6f15858d5904a2339698d90f5 Mon Sep 17 00:00:00 2001 From: x1a0b0 Date: Mon, 13 Oct 2025 14:20:11 +0800 Subject: [PATCH] Eliminate several singletons just for better lifecycle management Signed-off-by: x1a0b0 --- packages/client-rust/src/bin/client.rs | 75 +++++++++++-------- packages/client-rust/src/bin/monitor.rs | 17 +++-- .../client-rust/src/services/monitor/file.rs | 40 +++++----- .../src/services/monitor/instruction.rs | 26 +++++-- .../client-rust/src/services/monitor/kws.rs | 37 ++++++--- .../src/services/monitor/playing.rs | 29 +++++-- 6 files changed, 144 insertions(+), 80 deletions(-) diff --git a/packages/client-rust/src/bin/client.rs b/packages/client-rust/src/bin/client.rs index 532ef13..1da81a2 100644 --- a/packages/client-rust/src/bin/client.rs +++ b/packages/client-rust/src/bin/client.rs @@ -16,33 +16,45 @@ use open_xiaoai::services::connect::rpc::RPC; use open_xiaoai::services::monitor::instruction::InstructionMonitor; use open_xiaoai::services::monitor::playing::PlayingMonitor; -struct AppClient; +struct AppClient { + kws_monitor: KwsMonitor, + instruction_monitor: InstructionMonitor, + playing_monitor: PlayingMonitor, +} impl AppClient { - pub async fn connect(url: &str) -> Result { + pub fn new() -> Self { + Self { + kws_monitor: KwsMonitor::new(), + instruction_monitor: InstructionMonitor::new(), + playing_monitor: PlayingMonitor::new(), + } + } + + pub async fn connect(&self, url: &str) -> Result { let (ws_stream, _) = connect_async(url).await?; Ok(WsStream::Client(ws_stream)) } - pub async fn run() { + pub async fn run(&mut self) { let url = std::env::args().nth(1).expect("❌ 请输入服务器地址"); println!("✅ 已启动"); loop { - let Ok(ws_stream) = AppClient::connect(&url).await else { + let Ok(ws_stream) = self.connect(&url).await else { sleep(Duration::from_secs(1)).await; continue; }; println!("✅ 已连接: {:?}", url); - AppClient::init(ws_stream).await; + self.init(ws_stream).await; if let Err(e) = MessageManager::instance().process_messages().await { eprintln!("❌ 消息处理异常: {}", e); } - AppClient::dispose().await; + self.dispose().await; eprintln!("❌ 已断开连接"); } } - async fn init(ws_stream: WsStream) { + async fn init(&mut self, ws_stream: WsStream) { MessageManager::instance().init(ws_stream).await; MessageHandler::::instance() .set_handler(on_event) @@ -59,35 +71,38 @@ impl AppClient { rpc.add_command("start_recording", start_recording).await; rpc.add_command("stop_recording", stop_recording).await; - InstructionMonitor::start(|event| async move { - MessageManager::instance() - .send_event("instruction", Some(json!(event))) - .await - }) - .await; + self.instruction_monitor + .start(|event| async move { + MessageManager::instance() + .send_event("instruction", Some(json!(event))) + .await + }) + .await; - PlayingMonitor::start(|event| async move { - MessageManager::instance() - .send_event("playing", Some(json!(event))) - .await - }) - .await; + self.playing_monitor + .start(|event| async move { + MessageManager::instance() + .send_event("playing", Some(json!(event))) + .await + }) + .await; - KwsMonitor::start(|event| async move { - MessageManager::instance() - .send_event("kws", Some(json!(event))) - .await - }) - .await; + self.kws_monitor + .start(|event| async move { + MessageManager::instance() + .send_event("kws", Some(json!(event))) + .await + }) + .await; } - async fn dispose() { + async fn dispose(&mut self) { MessageManager::instance().dispose().await; let _ = AudioPlayer::instance().stop().await; let _ = AudioRecorder::instance().stop_recording().await; - InstructionMonitor::stop().await; - PlayingMonitor::stop().await; - KwsMonitor::stop().await; + self.instruction_monitor.stop().await; + self.playing_monitor.stop().await; + self.kws_monitor.stop().await; } } @@ -156,5 +171,5 @@ async fn on_stream(stream: Stream) -> Result<(), AppError> { #[tokio::main] async fn main() { - AppClient::run().await; + AppClient::new().run().await; } diff --git a/packages/client-rust/src/bin/monitor.rs b/packages/client-rust/src/bin/monitor.rs index 69c6dac..5befc9d 100644 --- a/packages/client-rust/src/bin/monitor.rs +++ b/packages/client-rust/src/bin/monitor.rs @@ -53,14 +53,15 @@ async fn main() { on_started().await; } - KwsMonitor::start(|event| async move { - match event { - KwsMonitorEvent::Started => on_started().await, - KwsMonitorEvent::Keyword(keyword) => on_keyword(keyword).await, - } - Ok(()) - }) - .await; + KwsMonitor::new() + .start(|event| async move { + match event { + KwsMonitorEvent::Started => on_started().await, + KwsMonitorEvent::Keyword(keyword) => on_keyword(keyword).await, + } + Ok(()) + }) + .await; loop { tokio::time::sleep(std::time::Duration::from_secs(1)).await; diff --git a/packages/client-rust/src/services/monitor/file.rs b/packages/client-rust/src/services/monitor/file.rs index 3c3bb90..c735e87 100644 --- a/packages/client-rust/src/services/monitor/file.rs +++ b/packages/client-rust/src/services/monitor/file.rs @@ -1,13 +1,12 @@ use serde::{Deserialize, Serialize}; use std::future::Future; use std::path::Path; -use std::sync::LazyLock; use tokio::fs::OpenOptions; use tokio::io::{AsyncBufReadExt, AsyncSeekExt, BufReader, SeekFrom}; +use tokio::task::JoinHandle; use tokio::time::{sleep, Duration}; use crate::base::AppError; -use crate::utils::task::TaskManager; #[derive(Debug, Serialize, Deserialize)] pub enum FileMonitorEvent { @@ -15,20 +14,22 @@ pub enum FileMonitorEvent { NewLine(String), } -pub struct FileMonitor; +pub struct FileMonitor { + task_holder: Option>, +} -static INSTANCE: LazyLock = LazyLock::new(FileMonitor::new); +impl Default for FileMonitor { + fn default() -> Self { + Self::new() + } +} impl FileMonitor { - fn new() -> Self { - Self {} + pub fn new() -> Self { + Self { task_holder: None } } - pub fn instance() -> &'static Self { - &INSTANCE - } - - pub async fn start(&self, file_path: &str, on_update: F) + pub async fn start(&mut self, file_path: &str, on_update: F) where F: Fn(FileMonitorEvent) -> Fut + Send + Sync + 'static, Fut: Future> + Send + 'static, @@ -36,18 +37,19 @@ impl FileMonitor { let file_path_clone = file_path.to_string(); let monitor = tokio::spawn(async move { - let _ = FileMonitor::start_monitor(file_path_clone.as_str(), on_update).await; + let _ = Self::start_monitor(file_path_clone.as_str(), on_update).await; }); - TaskManager::instance() - .add(&format!("FileMonitor-{}", file_path), monitor) - .await; + if let Some(old_task) = self.task_holder.replace(monitor) { + println!("Aborting old file monitor task"); + old_task.abort(); + } } - pub async fn stop(&self, file_path: &str) { - TaskManager::instance() - .dispose(&format!("FileMonitor-{}", file_path)) - .await; + pub async fn stop(&mut self) { + if let Some(handle) = self.task_holder.take() { + handle.abort(); + } } async fn start_monitor(file_path: &str, on_update: F) -> Result<(), AppError> diff --git a/packages/client-rust/src/services/monitor/instruction.rs b/packages/client-rust/src/services/monitor/instruction.rs index d1d5a50..40bf2e6 100644 --- a/packages/client-rust/src/services/monitor/instruction.rs +++ b/packages/client-rust/src/services/monitor/instruction.rs @@ -6,23 +6,37 @@ use crate::base::AppError; use super::file::{FileMonitor, FileMonitorEvent}; -pub struct InstructionMonitor; - static INSTRUCTION_FILE_PATH: &str = "/tmp/mico_aivs_lab/instruction.log"; +pub struct InstructionMonitor { + file_monitor: FileMonitor, +} + +impl Default for InstructionMonitor { + fn default() -> Self { + Self::new() + } +} + impl InstructionMonitor { - pub async fn start(on_update: F) + pub fn new() -> Self { + Self { + file_monitor: FileMonitor::new(), + } + } + + pub async fn start(&mut self, on_update: F) where F: Fn(FileMonitorEvent) -> Fut + Send + Sync + 'static, Fut: Future> + Send + 'static, { - FileMonitor::instance() + self.file_monitor .start(INSTRUCTION_FILE_PATH, on_update) .await; } - pub async fn stop() { - FileMonitor::instance().stop(INSTRUCTION_FILE_PATH).await; + pub async fn stop(&mut self) { + self.file_monitor.stop().await; } } diff --git a/packages/client-rust/src/services/monitor/kws.rs b/packages/client-rust/src/services/monitor/kws.rs index e7f3bba..66d80ca 100644 --- a/packages/client-rust/src/services/monitor/kws.rs +++ b/packages/client-rust/src/services/monitor/kws.rs @@ -8,36 +8,52 @@ use crate::base::AppError; use super::file::{FileMonitor, FileMonitorEvent}; +pub static KWS_FILE_PATH: &str = "/tmp/open-xiaoai/kws.log"; + #[derive(Debug, Serialize, Deserialize)] pub enum KwsMonitorEvent { Started, Keyword(String), } -pub struct KwsMonitor; +pub struct KwsMonitor { + file_monitor: FileMonitor, +} -pub static KWS_FILE_PATH: &str = "/tmp/open-xiaoai/kws.log"; - -static LAST_TIMESTAMP: AtomicU64 = AtomicU64::new(0); +impl Default for KwsMonitor { + fn default() -> Self { + Self::new() + } +} impl KwsMonitor { - pub async fn start(on_update: F) + pub fn new() -> Self { + Self { + file_monitor: FileMonitor::new(), + } + } + + pub async fn start(&mut self, on_update: F) where F: Fn(KwsMonitorEvent) -> Fut + Send + Sync + 'static, Fut: Future> + Send + 'static, { let on_update = Arc::new(on_update); - FileMonitor::instance() + let last_ts_store = Arc::new(AtomicU64::new(0)); + + self.file_monitor .start(KWS_FILE_PATH, move |event| { let on_update = Arc::clone(&on_update); + let last_ts_store = Arc::clone(&last_ts_store); + async move { if let FileMonitorEvent::NewLine(content) = event { let data = content.split('@').collect::>(); let timestamp = data[0].parse::().unwrap(); let keyword = data[1].to_string(); - let last_timestamp = LAST_TIMESTAMP.load(Ordering::Relaxed); + let last_timestamp = last_ts_store.load(Ordering::Relaxed); if timestamp != last_timestamp { - LAST_TIMESTAMP.store(timestamp, Ordering::Relaxed); + last_ts_store.store(timestamp, Ordering::Relaxed); let kws_event = if keyword == "__STARTED__" { KwsMonitorEvent::Started } else { @@ -52,8 +68,7 @@ impl KwsMonitor { .await; } - pub async fn stop() { - LAST_TIMESTAMP.store(0, Ordering::Relaxed); - FileMonitor::instance().stop(KWS_FILE_PATH).await; + pub async fn stop(&mut self) { + self.file_monitor.stop().await; } } diff --git a/packages/client-rust/src/services/monitor/playing.rs b/packages/client-rust/src/services/monitor/playing.rs index e5e293c..3789241 100644 --- a/packages/client-rust/src/services/monitor/playing.rs +++ b/packages/client-rust/src/services/monitor/playing.rs @@ -1,10 +1,10 @@ use serde::{Deserialize, Serialize}; use std::future::Future; +use tokio::task::JoinHandle; use tokio::time::{sleep, Duration}; use crate::base::AppError; use crate::utils::shell::run_shell; -use crate::utils::task::TaskManager; #[derive(Debug, Serialize, Deserialize, PartialEq, Clone)] pub enum PlayingMonitorEvent { @@ -13,10 +13,22 @@ pub enum PlayingMonitorEvent { Idle, } -pub struct PlayingMonitor; +pub struct PlayingMonitor { + task_holder: Option>, +} + +impl Default for PlayingMonitor { + fn default() -> Self { + Self::new() + } +} impl PlayingMonitor { - pub async fn start(on_update: F) + pub fn new() -> Self { + Self { task_holder: None } + } + + pub async fn start(&mut self, on_update: F) where F: Fn(PlayingMonitorEvent) -> Fut + Send + Sync + 'static, Fut: Future> + Send + 'static, @@ -25,11 +37,16 @@ impl PlayingMonitor { let _ = PlayingMonitor::start_monitor(on_update).await; }); - TaskManager::instance().add("PlayingMonitor", monitor).await; + if let Some(old_task) = self.task_holder.replace(monitor) { + println!("Aborting old playing monitor task"); + old_task.abort(); + }; } - pub async fn stop() { - TaskManager::instance().dispose("PlayingMonitor").await; + pub async fn stop(&mut self) { + if let Some(handle) = self.task_holder.take() { + handle.abort(); + } } async fn start_monitor(on_update: F) -> Result<(), AppError>