Eliminate several singletons

just for better lifecycle management

Signed-off-by: x1a0b0 <xbtseng@gmail.com>

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<WsStream, AppError> {
+    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<WsStream, AppError> {
         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::<Event>::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;
 }

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;

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<JoinHandle<()>>,
+}
 
-static INSTANCE: LazyLock<FileMonitor> = 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<F, Fut>(&self, file_path: &str, on_update: F)
+    pub async fn start<F, Fut>(&mut self, file_path: &str, on_update: F)
     where
         F: Fn(FileMonitorEvent) -> Fut + Send + Sync + 'static,
         Fut: Future<Output = Result<(), AppError>> + 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<F, Fut>(file_path: &str, on_update: F) -> Result<(), AppError>

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<F, Fut>(on_update: F)
+    pub fn new() -> Self {
+        Self {
+            file_monitor: FileMonitor::new(),
+        }
+    }
+
+    pub async fn start<F, Fut>(&mut self, on_update: F)
     where
         F: Fn(FileMonitorEvent) -> Fut + Send + Sync + 'static,
         Fut: Future<Output = Result<(), AppError>> + 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;
     }
 }
 

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<F, Fut>(on_update: F)
+    pub fn new() -> Self {
+        Self {
+            file_monitor: FileMonitor::new(),
+        }
+    }
+
+    pub async fn start<F, Fut>(&mut self, on_update: F)
     where
         F: Fn(KwsMonitorEvent) -> Fut + Send + Sync + 'static,
         Fut: Future<Output = Result<(), AppError>> + 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::<Vec<&str>>();
                         let timestamp = data[0].parse::<u64>().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;
     }
 }

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<JoinHandle<()>>,
+}
+
+impl Default for PlayingMonitor {
+    fn default() -> Self {
+        Self::new()
+    }
+}
 
 impl PlayingMonitor {
-    pub async fn start<F, Fut>(on_update: F)
+    pub fn new() -> Self {
+        Self { task_holder: None }
+    }
+
+    pub async fn start<F, Fut>(&mut self, on_update: F)
     where
         F: Fn(PlayingMonitorEvent) -> Fut + Send + Sync + 'static,
         Fut: Future<Output = Result<(), AppError>> + 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<F, Fut>(on_update: F) -> Result<(), AppError>