feat: 开放 Rust 补丁程序源代码和编译说明

packages/client-rust/src/base.rs

@@ -0,0 +1,3 @@
+pub type AppError = Box<dyn std::error::Error>;
+
+pub const VERSION: &str = env!("CARGO_PKG_VERSION");

packages/client-rust/src/bin/client.rs

@@ -0,0 +1,155 @@
+use open_xiaoai::services::audio::config::AudioConfig;
+use serde_json::json;
+use std::time::Duration;
+use tokio::time::sleep;
+use tokio_tungstenite::connect_async;
+
+use open_xiaoai::base::AppError;
+use open_xiaoai::base::VERSION;
+use open_xiaoai::services::audio::play::AudioPlayer;
+use open_xiaoai::services::audio::record::AudioRecorder;
+use open_xiaoai::services::connect::data::{Event, Request, Response, Stream};
+use open_xiaoai::services::connect::handler::MessageHandler;
+use open_xiaoai::services::connect::message::{MessageManager, WsStream};
+use open_xiaoai::services::connect::rpc::RPC;
+use open_xiaoai::services::monitor::instruction::InstructionMonitor;
+use open_xiaoai::services::monitor::playing::PlayingMonitor;
+
+struct AppClient;
+
+impl AppClient {
+    pub async fn connect(url: &str) -> Result<WsStream, AppError> {
+        let (ws_stream, _) = connect_async(url).await?;
+        Ok(WsStream::Client(ws_stream))
+    }
+
+    pub async fn run() {
+        let url = std::env::args().nth(1).expect("❌ 请输入服务器地址");
+        println!("✅ 已启动");
+        loop {
+            let Ok(ws_stream) = AppClient::connect(&url).await else {
+                sleep(Duration::from_secs(1)).await;
+                continue;
+            };
+            println!("✅ 已连接: {:?}", url);
+            AppClient::init(ws_stream).await;
+            if let Err(e) = MessageManager::instance().process_messages().await {
+                eprintln!("❌ 消息处理异常: {}", e);
+            }
+            AppClient::dispose().await;
+            eprintln!("❌ 已断开连接");
+        }
+    }
+
+    async fn init(ws_stream: WsStream) {
+        MessageManager::instance().init(ws_stream).await;
+        MessageHandler::<Event>::instance()
+            .set_handler(on_event)
+            .await;
+        MessageHandler::<Stream>::instance()
+            .set_handler(on_stream)
+            .await;
+
+        let rpc = RPC::instance();
+        rpc.add_command("get_version", get_version).await;
+        rpc.add_command("run_shell", run_shell).await;
+        rpc.add_command("start_play", start_play).await;
+        rpc.add_command("stop_play", stop_play).await;
+        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;
+
+        PlayingMonitor::instance()
+            .start(|event| async move {
+                MessageManager::instance()
+                    .send_event("playing", Some(json!(event)))
+                    .await
+            })
+            .await;
+    }
+
+    async fn dispose() {
+        MessageManager::instance().dispose().await;
+        let _ = AudioPlayer::instance().stop().await;
+        let _ = AudioRecorder::instance().stop_recording().await;
+        InstructionMonitor::stop().await;
+        PlayingMonitor::instance().stop().await;
+    }
+}
+
+async fn get_version(_: Request) -> Result<Response, AppError> {
+    let data = json!(VERSION.to_string());
+    Ok(Response::from_data(data))
+}
+
+async fn start_play(request: Request) -> Result<Response, AppError> {
+    let config = request
+        .payload
+        .and_then(|payload| serde_json::from_value::<AudioConfig>(payload).ok());
+    AudioPlayer::instance().start(config).await?;
+    Ok(Response::success())
+}
+
+async fn stop_play(_: Request) -> Result<Response, AppError> {
+    AudioPlayer::instance().stop().await?;
+    Ok(Response::success())
+}
+
+async fn start_recording(request: Request) -> Result<Response, AppError> {
+    let config = request
+        .payload
+        .and_then(|payload| serde_json::from_value::<AudioConfig>(payload).ok());
+    AudioRecorder::instance()
+        .start_recording(
+            |bytes| async {
+                MessageManager::instance()
+                    .send_stream("record", bytes, None)
+                    .await
+            },
+            config,
+        )
+        .await?;
+    Ok(Response::success())
+}
+
+async fn stop_recording(_: Request) -> Result<Response, AppError> {
+    AudioRecorder::instance().stop_recording().await?;
+    Ok(Response::success())
+}
+
+async fn run_shell(request: Request) -> Result<Response, AppError> {
+    let script = match request.payload {
+        Some(payload) => serde_json::from_value::<String>(payload)?,
+        _ => return Err("empty command".into()),
+    };
+    let res = open_xiaoai::utils::shell::run_shell(script.as_str()).await?;
+    Ok(Response::from_data(json!(res)))
+}
+
+async fn on_event(event: Event) -> Result<(), AppError> {
+    println!("🔥 收到事件: {:?}", event);
+    Ok(())
+}
+
+async fn on_stream(stream: Stream) -> Result<(), AppError> {
+    let Stream { tag, bytes, .. } = stream;
+    match tag.as_str() {
+        "play" => {
+            // 播放接收到的音频流
+            let _ = AudioPlayer::instance().play(bytes).await;
+        }
+        _ => {}
+    }
+    Ok(())
+}
+
+#[tokio::main]
+async fn main() {
+    AppClient::run().await;
+}

packages/client-rust/src/lib.rs

@@ -0,0 +1,3 @@
+pub mod base;
+pub mod services;
+pub mod utils;

packages/client-rust/src/services/audio/config.rs

@@ -0,0 +1,22 @@
+use std::sync::LazyLock;
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct AudioConfig {
+    pub pcm: String,
+    pub channels: u16,
+    pub bits_per_sample: u16,
+    pub sample_rate: u32,
+    pub period_size: u32,
+    pub buffer_size: u32,
+}
+
+pub static AUDIO_CONFIG: LazyLock<AudioConfig> = LazyLock::new(|| AudioConfig {
+    pcm: "noop".into(),
+    channels: 1,
+    bits_per_sample: 16,
+    sample_rate: 24000,
+    period_size: 1440 / 4,
+    buffer_size: 1440,
+});

packages/client-rust/src/services/audio/mod.rs

@@ -0,0 +1,3 @@
+pub mod config;
+pub mod play;
+pub mod record;

packages/client-rust/src/services/audio/play.rs

@@ -0,0 +1,117 @@
+use std::process::Stdio;
+use std::sync::{Arc, LazyLock};
+use tokio::io::AsyncWriteExt;
+use tokio::process::{Child, ChildStdin, Command};
+use tokio::sync::{mpsc, Mutex};
+use tokio::task::JoinHandle;
+
+use crate::base::AppError;
+
+use super::config::{AudioConfig, AUDIO_CONFIG};
+
+pub struct AudioPlayer {
+    aplay_thread: Arc<Mutex<Option<Child>>>,
+    write_thread: Arc<Mutex<Option<ChildStdin>>>,
+    sender: Arc<Mutex<Option<mpsc::Sender<Vec<u8>>>>>,
+    player_task: Arc<Mutex<Option<JoinHandle<()>>>>,
+}
+
+static INSTANCE: LazyLock<AudioPlayer> = LazyLock::new(AudioPlayer::new);
+
+impl AudioPlayer {
+    fn new() -> Self {
+        Self {
+            aplay_thread: Arc::new(Mutex::new(None)),
+            write_thread: Arc::new(Mutex::new(None)),
+            sender: Arc::new(Mutex::new(None)),
+            player_task: Arc::new(Mutex::new(None)),
+        }
+    }
+
+    pub fn instance() -> &'static Self {
+        &INSTANCE
+    }
+
+    pub async fn stop(&self) -> Result<(), AppError> {
+        let mut sender_guard = self.sender.lock().await;
+        if let Some(sender) = sender_guard.take() {
+            drop(sender);
+        }
+
+        if let Some(task) = self.player_task.lock().await.take() {
+            let _ = task.abort();
+        }
+
+        if let Some(mut write_thread) = self.write_thread.lock().await.take() {
+            let _ = write_thread.shutdown().await;
+            drop(write_thread);
+        }
+
+        if let Some(mut aplay_thread) = self.aplay_thread.lock().await.take() {
+            let _ = aplay_thread.kill().await;
+        }
+
+        Ok(())
+    }
+
+    pub async fn start(&self, config: Option<AudioConfig>) -> Result<(), AppError> {
+        let is_started = self.sender.lock().await.is_some();
+        if is_started {
+            self.stop().await?;
+        }
+
+        let config = config.unwrap_or_else(|| (*AUDIO_CONFIG).clone());
+
+        let mut aplay_thread = Command::new("aplay")
+            .args([
+                "--quiet",
+                "-t",
+                "raw",
+                "-f",
+                &format!("S{}_LE", config.bits_per_sample),
+                "-r",
+                &config.sample_rate.to_string(),
+                "-c",
+                &config.channels.to_string(),
+                "--buffer-size",
+                &config.buffer_size.to_string(),
+                "--period-size",
+                &config.period_size.to_string(),
+                "-",
+            ])
+            .stdin(Stdio::piped())
+            .spawn()?;
+
+        let stdin = aplay_thread.stdin.take().unwrap();
+        self.aplay_thread.lock().await.replace(aplay_thread);
+        self.write_thread.lock().await.replace(stdin);
+
+        let (tx, mut rx) = mpsc::channel::<Vec<u8>>(100);
+
+        let write_thread_clone = self.write_thread.clone();
+        let player_task = tokio::spawn(async move {
+            while let Some(bytes) = rx.recv().await {
+                let mut write_guard = write_thread_clone.lock().await;
+                if let Some(write_thread) = write_guard.as_mut() {
+                    let _ = write_thread.write_all(&bytes).await;
+                } else {
+                    break;
+                }
+            }
+        });
+
+        self.player_task.lock().await.replace(player_task);
+        self.sender.lock().await.replace(tx);
+
+        Ok(())
+    }
+
+    pub async fn play(&self, bytes: Vec<u8>) -> Result<(), AppError> {
+        let sender_guard = self.sender.lock().await;
+        if let Some(sender) = sender_guard.as_ref() {
+            sender.send(bytes).await?;
+        }
+
+        Ok(())
+    }
+}

packages/client-rust/src/services/audio/record.rs

@@ -0,0 +1,126 @@
+use std::future::Future;
+use std::process::Stdio;
+use std::sync::{Arc, LazyLock};
+use tokio::io::AsyncReadExt;
+use tokio::process::{Child, Command};
+use tokio::sync::Mutex;
+use tokio::task::JoinHandle;
+
+use crate::base::AppError;
+
+use super::config::{AudioConfig, AUDIO_CONFIG};
+
+#[derive(PartialEq)]
+enum State {
+    Idle,
+    Recording,
+}
+
+pub struct AudioRecorder {
+    state: Arc<Mutex<State>>,
+    arecord_thread: Arc<Mutex<Option<Child>>>,
+    read_thread: Arc<Mutex<Option<JoinHandle<()>>>>,
+}
+
+static INSTANCE: LazyLock<AudioRecorder> = LazyLock::new(AudioRecorder::new);
+
+impl AudioRecorder {
+    fn new() -> Self {
+        Self {
+            state: Arc::new(Mutex::new(State::Idle)),
+            arecord_thread: Arc::new(Mutex::new(None)),
+            read_thread: Arc::new(Mutex::new(None)),
+        }
+    }
+
+    pub fn instance() -> &'static Self {
+        &INSTANCE
+    }
+
+    pub async fn stop_recording(&self) -> Result<(), AppError> {
+        let mut state = self.state.lock().await;
+        if *state == State::Idle {
+            return Ok(());
+        }
+
+        if let Some(read_thread) = self.read_thread.lock().await.take() {
+            read_thread.abort();
+        }
+
+        if let Some(mut arecord_thread) = self.arecord_thread.lock().await.take() {
+            let _ = arecord_thread.kill().await;
+        }
+
+        *state = State::Idle;
+        Ok(())
+    }
+
+    pub async fn start_recording<F, Fut>(
+        &self,
+        on_stream: F,
+        config: Option<AudioConfig>,
+    ) -> Result<(), AppError>
+    where
+        F: Fn(Vec<u8>) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), AppError>> + Send + 'static,
+    {
+        let mut state = self.state.lock().await;
+        if *state == State::Recording {
+            return Ok(());
+        }
+
+        let config = config.unwrap_or_else(|| (*AUDIO_CONFIG).clone());
+
+        let mut arecord_thread = Command::new("arecord")
+            .args([
+                "--quiet",
+                "-t",
+                "raw",
+                "-D",
+                &config.pcm,
+                "-f",
+                &format!("S{}_LE", config.bits_per_sample),
+                "-r",
+                &config.sample_rate.to_string(),
+                "-c",
+                &config.channels.to_string(),
+                "--buffer-size",
+                &config.buffer_size.to_string(),
+                "--period-size",
+                &config.period_size.to_string(),
+            ])
+            .stdout(Stdio::piped())
+            .spawn()?;
+
+        let mut stdout = arecord_thread.stdout.take().unwrap();
+        let read_thread = tokio::spawn(async move {
+            let size = (config.bits_per_sample / 8) as usize;
+            let target_size = config.buffer_size as usize * size;
+
+            let mut accumulated_data = Vec::new();
+            let mut buffer = vec![0u8; config.period_size as usize * size];
+
+            loop {
+                match stdout.read(&mut buffer).await {
+                    Ok(size) if size > 0 => {
+                        accumulated_data.extend_from_slice(&buffer[..size]);
+                        if accumulated_data.len() >= target_size {
+                            let data_to_send =
+                                accumulated_data.drain(..target_size).collect::<Vec<u8>>();
+                            let _ = on_stream(data_to_send).await;
+                        }
+                    }
+                    _ => break,
+                }
+            }
+
+            let _ = AudioRecorder::instance().stop_recording().await;
+        });
+
+        self.arecord_thread.lock().await.replace(arecord_thread);
+        self.read_thread.lock().await.replace(read_thread);
+
+        *state = State::Recording;
+        Ok(())
+    }
+}

packages/client-rust/src/services/connect/data.rs

@@ -0,0 +1,96 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use uuid::Uuid;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum AppMessage {
+    Request(Request),
+    Response(Response),
+    Event(Event),
+    Stream(Stream),
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Stream {
+    pub id: String,
+    pub tag: String,
+    pub bytes: Vec<u8>,
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub data: Option<Value>,
+}
+
+impl Stream {
+    pub fn new(tag: &str, bytes: Vec<u8>, data: Option<Value>) -> Self {
+        Self {
+            id: Uuid::new_v4().to_string(),
+            tag: tag.to_string(),
+            bytes,
+            data,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Event {
+    pub id: String,
+    pub event: String,
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub data: Option<Value>,
+}
+
+impl Event {
+    pub fn new(event: &str, data: Option<Value>) -> Self {
+        Self {
+            id: Uuid::new_v4().to_string(),
+            event: event.to_string(),
+            data,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Request {
+    pub id: String,
+    pub command: String,
+    pub payload: Option<Value>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Response {
+    pub id: String,
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub code: Option<i32>,
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub msg: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none", default)]
+    pub data: Option<Value>,
+}
+
+impl Response {
+    pub fn success() -> Self {
+        Self {
+            id: 0.to_string(),
+            code: Some(0),
+            msg: Some("success".to_string()),
+            data: None,
+        }
+    }
+
+    pub fn from_data(data: Value) -> Self {
+        Self {
+            id: 0.to_string(),
+            code: None,
+            msg: None,
+            data: Some(data),
+        }
+    }
+
+    pub fn from_error(id: &str, e: impl std::fmt::Display) -> Self {
+        Self {
+            id: id.to_string(),
+            code: Some(-1),
+            msg: Some(e.to_string()),
+            data: None,
+        }
+    }
+}

packages/client-rust/src/services/connect/handler.rs

@@ -0,0 +1,93 @@
+use futures::future::BoxFuture;
+use std::future::Future;
+use std::sync::{Arc, LazyLock};
+use tokio::sync::Mutex;
+use tokio_tungstenite::tungstenite::Message;
+
+use crate::base::AppError;
+
+use super::data::{AppMessage, Event, Request, Response, Stream};
+use super::message::MessageManager;
+use super::rpc::RPC;
+
+type Handler<T> = Arc<dyn Fn(T) -> BoxFuture<'static, Result<(), AppError>> + Send + Sync>;
+
+pub struct MessageHandler<T> {
+    handler: Arc<Mutex<Option<Handler<T>>>>,
+}
+
+static REQUEST_INSTANCE: LazyLock<MessageHandler<Request>> = LazyLock::new(MessageHandler::new);
+static RESPONSE_INSTANCE: LazyLock<MessageHandler<Response>> = LazyLock::new(MessageHandler::new);
+static EVENT_INSTANCE: LazyLock<MessageHandler<Event>> = LazyLock::new(MessageHandler::new);
+static BYTES_INSTANCE: LazyLock<MessageHandler<Stream>> = LazyLock::new(MessageHandler::new);
+
+impl<T> MessageHandler<T> {
+    fn new() -> Self {
+        Self {
+            handler: Arc::new(Mutex::new(None)),
+        }
+    }
+
+    pub async fn set_handler<F, Fut>(&self, handler: F)
+    where
+        F: Fn(T) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), AppError>> + Send + 'static,
+    {
+        *self.handler.lock().await = Some(Arc::new(move |data| Box::pin(handler(data))));
+    }
+
+    pub async fn on(&self, data: T) -> Result<(), AppError> {
+        let handler = {
+            let guard = self.handler.lock().await;
+            match &*guard {
+                Some(h) => Arc::clone(h),
+                None => return Err("handler is not initialized".into()),
+            }
+        };
+        handler(data).await
+    }
+}
+
+impl MessageHandler<Request> {
+    pub fn instance() -> &'static MessageHandler<Request> {
+        &REQUEST_INSTANCE
+    }
+
+    pub async fn on_request(&self, request: Request) -> Result<(), AppError> {
+        println!("🚗 收到指令: {:?}", request);
+        let id = request.id.clone();
+        let response: Response = match RPC::instance().on_request(request).await {
+            Ok(resp) => Response { id, ..resp },
+            Err(e) => Response::from_error(&id, e),
+        };
+        if let Ok(data) = serde_json::to_string(&AppMessage::Response(response)) {
+            MessageManager::instance()
+                .send(Message::Text(data.into()))
+                .await?;
+        }
+        Ok(())
+    }
+}
+
+impl MessageHandler<Response> {
+    pub fn instance() -> &'static MessageHandler<Response> {
+        &RESPONSE_INSTANCE
+    }
+
+    pub async fn on_response(&self, response: Response) -> Result<(), AppError> {
+        RPC::instance().on_response(response).await;
+        Ok(())
+    }
+}
+
+impl MessageHandler<Event> {
+    pub fn instance() -> &'static MessageHandler<Event> {
+        &EVENT_INSTANCE
+    }
+}
+
+impl MessageHandler<Stream> {
+    pub fn instance() -> &'static MessageHandler<Stream> {
+        &BYTES_INSTANCE
+    }
+}

packages/client-rust/src/services/connect/message.rs

@@ -0,0 +1,203 @@
+use futures::stream::{SplitSink, SplitStream};
+use futures::{SinkExt, StreamExt};
+use serde_json::Value;
+use std::future::Future;
+use std::sync::{Arc, LazyLock};
+use tokio::net::TcpStream;
+use tokio::sync::{Mutex, Semaphore};
+use tokio_tungstenite::MaybeTlsStream;
+use tokio_tungstenite::{tungstenite::Message, WebSocketStream};
+
+use super::rpc::RPC;
+use crate::base::AppError;
+use crate::utils::task::TaskManager;
+
+use super::data::{AppMessage, Event, Request, Response, Stream};
+use super::handler::MessageHandler;
+
+pub enum WsStream {
+    Server(WebSocketStream<TcpStream>),
+    Client(WebSocketStream<MaybeTlsStream<TcpStream>>),
+}
+
+pub enum WsReader {
+    Server(SplitStream<WebSocketStream<TcpStream>>),
+    Client(SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>),
+}
+
+pub enum WsWriter {
+    Server(SplitSink<WebSocketStream<TcpStream>, Message>),
+    Client(SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>),
+}
+
+pub struct MessageManager {
+    semaphore: Arc<Semaphore>,
+    reader: Arc<Mutex<Option<WsReader>>>,
+    writer: Arc<Mutex<Option<WsWriter>>>,
+}
+
+static INSTANCE: LazyLock<MessageManager> = LazyLock::new(MessageManager::new);
+
+impl MessageManager {
+    fn new() -> Self {
+        Self {
+            reader: Arc::new(Mutex::new(None)),
+            writer: Arc::new(Mutex::new(None)),
+            semaphore: Arc::new(Semaphore::new(32)),
+        }
+    }
+
+    pub fn instance() -> &'static Self {
+        &INSTANCE
+    }
+
+    pub async fn init(&self, ws_stream: WsStream) {
+        match ws_stream {
+            WsStream::Client(stream) => {
+                let (tx, rx) = stream.split();
+                self.reader.lock().await.replace(WsReader::Client(rx));
+                self.writer.lock().await.replace(WsWriter::Client(tx));
+            }
+            WsStream::Server(stream) => {
+                let (tx, rx) = stream.split();
+                self.reader.lock().await.replace(WsReader::Server(rx));
+                self.writer.lock().await.replace(WsWriter::Server(tx));
+            }
+        }
+        RPC::instance()
+            .init(|request| async {
+                let data = serde_json::to_string(&AppMessage::Request(request)).unwrap();
+                MessageManager::instance()
+                    .send(Message::Text(data.into()))
+                    .await
+            })
+            .await;
+    }
+
+    pub async fn dispose(&self) {
+        *self.reader.lock().await = None;
+        *self.writer.lock().await = None;
+        RPC::instance().dispose().await;
+        TaskManager::instance().dispose("MessageManager").await;
+    }
+
+    pub async fn send(&self, msg: Message) -> Result<(), AppError> {
+        let mut writer_guard = self.writer.lock().await;
+
+        let Some(writer) = &mut *writer_guard else {
+            return Err("WebSocket writer is not initialized".into());
+        };
+
+        match writer {
+            WsWriter::Client(w) => w.send(msg).await?,
+            WsWriter::Server(w) => w.send(msg).await?,
+        }
+
+        Ok(())
+    }
+
+    pub async fn send_event(&self, event: &str, data: Option<Value>) -> Result<(), AppError> {
+        let event: Event = Event::new(event, data);
+        let data = serde_json::to_string(&AppMessage::Event(event)).unwrap();
+        MessageManager::instance()
+            .send(Message::Text(data.into()))
+            .await
+    }
+
+    pub async fn send_stream(
+        &self,
+        tag: &str,
+        bytes: Vec<u8>,
+        data: Option<Value>,
+    ) -> Result<(), AppError> {
+        let stream = serde_json::to_vec(&Stream::new(tag, bytes, data)).unwrap();
+        MessageManager::instance()
+            .send(Message::Binary(stream.into()))
+            .await
+    }
+
+    pub async fn process_messages(&self) -> Result<(), AppError> {
+        if self.reader.lock().await.is_none() {
+            return Err("WebSocket reader is not initialized".into());
+        }
+
+        loop {
+            let next_msg = {
+                let mut reader = self.reader.lock().await;
+                match reader.as_mut() {
+                    None => break,
+                    Some(WsReader::Client(reader)) => reader.next().await,
+                    Some(WsReader::Server(reader)) => reader.next().await,
+                }
+            };
+            match next_msg {
+                None => break,
+                Some(Ok(Message::Close(_))) => break,
+                Some(Err(e)) => return Err(e.into()),
+                Some(Ok(msg)) => {
+                    match msg {
+                        Message::Text(text) => {
+                            let _ = self.on_text(text.to_string()).await;
+                        }
+                        Message::Binary(bytes) => {
+                            let _ = self.on_bytes(bytes.into()).await;
+                        }
+                        _ => {}
+                    };
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    async fn on_bytes(&self, bytes: Vec<u8>) -> Result<(), AppError> {
+        let data = serde_json::from_slice::<Stream>(&bytes)?;
+        MessageHandler::<Stream>::instance().on(data).await
+    }
+
+    async fn on_text(&self, text: String) -> Result<(), AppError> {
+        let msg = serde_json::from_str::<AppMessage>(&text)?;
+
+        match msg {
+            AppMessage::Request(request) => {
+                self.run_concurrently(move || {
+                    let request = request.clone();
+                    async move {
+                        MessageHandler::<Request>::instance()
+                            .on_request(request)
+                            .await?;
+                        Ok(())
+                    }
+                })
+                .await
+            }
+            AppMessage::Response(response) => {
+                MessageHandler::<Response>::instance()
+                    .on_response(response)
+                    .await
+            }
+            AppMessage::Event(event) => MessageHandler::<Event>::instance().on(event).await,
+            _ => Ok(()),
+        }
+    }
+
+    async fn run_concurrently<F, Fut>(&self, run: F) -> Result<(), AppError>
+    where
+        F: Fn() -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), AppError>> + Send + 'static,
+    {
+        let permit = match self.semaphore.clone().try_acquire_owned() {
+            Ok(permit) => permit,
+            Err(_) => self.semaphore.clone().acquire_owned().await?,
+        };
+
+        let task = tokio::spawn(async move {
+            let _ = run().await;
+            drop(permit);
+        });
+
+        TaskManager::instance().add("MessageManager", task).await;
+        Ok(())
+    }
+}

packages/client-rust/src/services/connect/mod.rs

@@ -0,0 +1,4 @@
+pub mod data;
+pub mod handler;
+pub mod message;
+pub mod rpc;

packages/client-rust/src/services/connect/rpc.rs

@@ -0,0 +1,150 @@
+use futures::future::BoxFuture;
+use std::collections::HashMap;
+use std::future::Future;
+use std::sync::{Arc, LazyLock};
+use tokio::sync::{oneshot, Mutex, RwLock};
+use tokio::time::{timeout, Duration};
+use uuid::Uuid;
+
+use crate::base::AppError;
+
+use super::data::{Request, Response};
+
+type SendRequestFn = Arc<dyn Fn(Request) -> BoxFuture<'static, Result<(), AppError>> + Send + Sync>;
+
+type RequestHandler =
+    Arc<dyn Fn(Request) -> BoxFuture<'static, Result<Response, AppError>> + Send + Sync>;
+
+static INSTANCE: LazyLock<RPC> = LazyLock::new(RPC::new);
+
+pub struct RPC {
+    send_request: Arc<RwLock<Option<SendRequestFn>>>,
+    request_handlers: Arc<RwLock<HashMap<String, RequestHandler>>>,
+    pending_requests: Arc<Mutex<HashMap<String, oneshot::Sender<Response>>>>,
+}
+
+impl RPC {
+    fn new() -> Self {
+        Self {
+            send_request: Arc::new(RwLock::new(None)),
+            request_handlers: Arc::new(RwLock::new(HashMap::new())),
+            pending_requests: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    pub fn instance() -> &'static Self {
+        &INSTANCE
+    }
+
+    pub async fn init<F, Fut>(&self, send_request: F)
+    where
+        F: Fn(Request) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), AppError>> + Send + 'static,
+    {
+        let mut lock = self.send_request.write().await;
+        *lock = Some(Arc::new(move |msg| Box::pin(send_request(msg))));
+    }
+
+    pub async fn dispose(&self) {
+        {
+            let mut lock = self.send_request.write().await;
+            *lock = None;
+        }
+        {
+            let mut lock = self.request_handlers.write().await;
+            lock.clear();
+        }
+        {
+            let mut lock = self.pending_requests.lock().await;
+            lock.clear();
+        }
+    }
+
+    /// 添加 local 方法
+    pub async fn add_command<F, Fut>(&self, command: &str, handler: F)
+    where
+        F: Fn(Request) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<Response, AppError>> + Send + 'static,
+    {
+        let mut handlers = self.request_handlers.write().await;
+        handlers.insert(
+            command.to_string(),
+            Arc::new(move |request| Box::pin(handler(request))),
+        );
+    }
+
+    /// local 收到 remote 调用
+    pub async fn on_request(&self, request: Request) -> Result<Response, AppError> {
+        let handler = self.get_request_handler(&request.command).await;
+        match handler {
+            Some(handler) => handler(request).await,
+            None => Err("command not found".into()),
+        }
+    }
+
+    /// local 收到 remote 响应
+    pub async fn on_response(&self, response: Response) {
+        let tx = {
+            let mut pending = self.pending_requests.lock().await;
+            pending.remove(&response.id)
+        };
+
+        if let Some(tx) = tx {
+            let _ = tx.send(response);
+        }
+    }
+
+    /// 调用 remote 方法
+    pub async fn call_remote(
+        &self,
+        command: &str,
+        payload: Option<serde_json::Value>,
+        timeout_millis: Option<u64>,
+    ) -> Result<Response, AppError> {
+        let send_request = self.get_send_request().await;
+        let Some(send_request) = send_request else {
+            return Err("send_request is not initialized".into());
+        };
+
+        let uid = Uuid::new_v4().to_string();
+        let (tx, rx) = oneshot::channel();
+
+        let request = Request {
+            id: uid.clone(),
+            command: command.to_string(),
+            payload,
+        };
+
+        send_request(request).await?;
+
+        {
+            let mut pending = self.pending_requests.lock().await;
+            pending.insert(uid.clone(), tx);
+        }
+
+        let timeout_duration = Duration::from_millis(timeout_millis.unwrap_or(10 * 1000));
+        match timeout(timeout_duration, rx).await {
+            Ok(Ok(response)) => Ok(response),
+            Ok(Err(_)) => {
+                let mut pending = self.pending_requests.lock().await;
+                pending.remove(&uid);
+                Err("response channel closed".into())
+            }
+            Err(_) => {
+                let mut pending = self.pending_requests.lock().await;
+                pending.remove(&uid);
+                Err("request timeout".into())
+            }
+        }
+    }
+
+    async fn get_send_request(&self) -> Option<SendRequestFn> {
+        let send_request = self.send_request.read().await;
+        send_request.clone()
+    }
+
+    async fn get_request_handler(&self, event: &str) -> Option<RequestHandler> {
+        let handlers = self.request_handlers.read().await;
+        handlers.get(event).cloned()
+    }
+}

packages/client-rust/src/services/mod.rs

@@ -0,0 +1,4 @@
+pub mod audio;
+pub mod connect;
+pub mod monitor;
+pub mod speaker;

packages/client-rust/src/services/monitor/file.rs

@@ -0,0 +1,106 @@
+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::time::{sleep, Duration};
+
+use crate::base::AppError;
+use crate::utils::task::TaskManager;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum FileMonitorEvent {
+    NewFile,
+    NewLine(String),
+}
+
+pub struct FileMonitor;
+
+static INSTANCE: LazyLock<FileMonitor> = LazyLock::new(FileMonitor::new);
+
+impl FileMonitor {
+    fn new() -> Self {
+        Self {}
+    }
+
+    pub fn instance() -> &'static Self {
+        &INSTANCE
+    }
+
+    pub async fn start<F, Fut>(&self, file_path: &str, on_update: F)
+    where
+        F: Fn(FileMonitorEvent) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), AppError>> + Send + 'static,
+    {
+        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;
+        });
+
+        TaskManager::instance()
+            .add(&format!("FileMonitor-{}", file_path), monitor)
+            .await;
+    }
+
+    pub async fn stop(&self, file_path: &str) {
+        TaskManager::instance()
+            .dispose(&format!("FileMonitor-{}", file_path))
+            .await;
+    }
+
+    async fn start_monitor<F, Fut>(file_path: &str, on_update: F) -> Result<(), AppError>
+    where
+        F: Fn(FileMonitorEvent) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), AppError>> + Send + 'static,
+    {
+        while !Path::new(file_path).exists() {
+            sleep(Duration::from_millis(100)).await;
+        }
+
+        let file = OpenOptions::new().read(true).open(file_path).await?;
+        let mut reader = BufReader::new(file);
+
+        let metadata = reader.get_ref().metadata().await.unwrap();
+        let mut position = metadata.len();
+
+        loop {
+            let metadata = reader.get_ref().metadata().await.unwrap();
+
+            let current_size = metadata.len();
+            if current_size < position {
+                position = 0;
+                let _ = on_update(FileMonitorEvent::NewFile).await;
+            }
+
+            if reader.stream_position().await? != position {
+                reader.seek(SeekFrom::Start(position)).await?;
+            }
+
+            let mut line = String::new();
+            let mut new_content_found = false;
+
+            while let Ok(bytes_read) = reader.read_line(&mut line).await {
+                if bytes_read == 0 {
+                    break;
+                }
+
+                let trimmed_line = line.trim();
+                if !trimmed_line.is_empty() {
+                    new_content_found = true;
+                    let _ = on_update(FileMonitorEvent::NewLine(trimmed_line.to_string())).await;
+                }
+
+                position = reader.stream_position().await?;
+                line.clear();
+            }
+
+            if !new_content_found {
+                sleep(Duration::from_millis(100)).await;
+            } else {
+                sleep(Duration::from_millis(10)).await;
+            }
+        }
+    }
+}

packages/client-rust/src/services/monitor/instruction.rs

@@ -0,0 +1,134 @@
+use std::future::Future;
+
+use serde::{Deserialize, Serialize};
+
+use crate::base::AppError;
+
+use super::file::{FileMonitor, FileMonitorEvent};
+
+pub struct InstructionMonitor;
+
+static INSTRUCTION_FILE_PATH: &str = "/tmp/mico_aivs_lab/instruction.log";
+
+impl InstructionMonitor {
+    pub async fn start<F, Fut>(on_update: F)
+    where
+        F: Fn(FileMonitorEvent) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), AppError>> + Send + 'static,
+    {
+        FileMonitor::instance()
+            .start(INSTRUCTION_FILE_PATH, on_update)
+            .await;
+    }
+
+    pub async fn stop() {
+        FileMonitor::instance().stop(INSTRUCTION_FILE_PATH).await;
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Header {
+    pub dialog_id: String,
+    pub id: String,
+    pub name: String,
+    pub namespace: String,
+}
+
+#[derive(Debug, Serialize, Deserialize, Default)]
+pub struct RecognizeResult {
+    #[serde(default)]
+    pub confidence: f64,
+    #[serde(default)]
+    pub text: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub asr_binary_offset: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub begin_offset: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub end_offset: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub is_nlp_request: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub is_stop: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub origin_text: Option<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum Payload {
+    RecognizeResultPayload {
+        is_final: bool,
+        is_vad_begin: bool,
+        results: Vec<RecognizeResult>,
+    },
+    StopCapturePayload {
+        stop_time: u64,
+    },
+    SpeakPayload {
+        text: String,
+        #[serde(skip_serializing_if = "Option::is_none")]
+        emotion: Option<Emotion>,
+    },
+    PlayPayload {
+        audio_items: Vec<AudioItem>,
+        audio_type: String,
+        loadmore_token: String,
+        needs_loadmore: bool,
+        origin_id: String,
+        play_behavior: String,
+    },
+    SetPropertyPayload {
+        name: String,
+        value: String,
+    },
+    InstructionControlPayload {
+        behavior: String,
+    },
+    EmptyPayload {},
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Emotion {
+    pub category: String,
+    pub level: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct AudioItem {
+    pub item_id: ItemId,
+    pub log: Log,
+    pub stream: Stream,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ItemId {
+    pub audio_id: String,
+    pub cp: Cp,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Cp {
+    pub id: String,
+    pub name: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Log {
+    pub eid: String,
+    pub refer: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Stream {
+    pub authentication: bool,
+    pub duration_in_ms: u64,
+    pub offset_in_ms: u64,
+    pub url: String,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct LogMessage {
+    pub header: Header,
+    pub payload: Payload,
+}

packages/client-rust/src/services/monitor/mod.rs

@@ -0,0 +1,3 @@
+pub mod file;
+pub mod instruction;
+pub mod playing;

packages/client-rust/src/services/monitor/playing.rs

@@ -0,0 +1,70 @@
+use serde::{Deserialize, Serialize};
+use std::future::Future;
+use std::sync::LazyLock;
+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 {
+    Playing,
+    Paused,
+    Idle,
+}
+
+pub struct PlayingMonitor;
+
+static INSTANCE: LazyLock<PlayingMonitor> = LazyLock::new(PlayingMonitor::new);
+
+impl PlayingMonitor {
+    fn new() -> Self {
+        Self {}
+    }
+
+    pub fn instance() -> &'static Self {
+        &INSTANCE
+    }
+
+    pub async fn start<F, Fut>(&self, on_update: F)
+    where
+        F: Fn(PlayingMonitorEvent) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), AppError>> + Send + 'static,
+    {
+        let monitor = tokio::spawn(async move {
+            let _ = PlayingMonitor::start_monitor(on_update).await;
+        });
+
+        TaskManager::instance().add("PlayingMonitor", monitor).await;
+    }
+
+    pub async fn stop(&self) {
+        TaskManager::instance().dispose("PlayingMonitor").await;
+    }
+
+    async fn start_monitor<F, Fut>(on_update: F) -> Result<(), AppError>
+    where
+        F: Fn(PlayingMonitorEvent) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), AppError>> + Send + 'static,
+    {
+        let mut last_status = PlayingMonitorEvent::Idle;
+        loop {
+            let res = run_shell("mphelper mute_stat").await?;
+            let status = if res.stdout.contains("1") {
+                PlayingMonitorEvent::Playing
+            } else if res.stdout.contains("2") {
+                PlayingMonitorEvent::Paused
+            } else {
+                PlayingMonitorEvent::Idle
+            };
+
+            if last_status != status {
+                last_status = status.clone();
+                let _ = on_update(status).await;
+            }
+
+            sleep(Duration::from_millis(100)).await;
+        }
+    }
+}

packages/client-rust/src/services/speaker.rs

@@ -0,0 +1,174 @@
+use serde_json::json;
+
+use crate::utils::shell::CommandResult;
+use crate::{base::AppError, services::connect::rpc::RPC};
+
+pub struct SpeakerManager;
+
+impl SpeakerManager {
+    /// 获取启动分区
+    pub async fn get_boot() -> Result<String, AppError> {
+        const COMMAND: &str = r#"
+            echo $(fw_env -g boot_part)
+        "#;
+        let res = SpeakerManager::run_shell(COMMAND).await?;
+        Ok(res.stdout.trim().to_string())
+    }
+
+    /// 设置启动分区
+    pub async fn set_boot(boot_part: &str) -> Result<bool, AppError> {
+        const COMMAND: &str = r#"
+            fw_env -s boot_part %s >/dev/null 2>&1 && echo $(fw_env -g boot_part)
+        "#;
+        let script = COMMAND.replace("%s", boot_part);
+        let res = SpeakerManager::run_shell(&script).await?;
+        Ok(res.stdout.contains(boot_part))
+    }
+
+    /// 获取设备型号
+    pub async fn get_device_model() -> Result<String, AppError> {
+        const COMMAND: &str = r#"
+            echo $(micocfg_model)
+        "#;
+        let res = SpeakerManager::run_shell(COMMAND).await?;
+        Ok(res.stdout.trim().to_string())
+    }
+
+    /// 获取设备序列号
+    pub async fn get_device_sn() -> Result<String, AppError> {
+        const COMMAND: &str = r#"
+            echo $(micocfg_sn)
+        "#;
+        let res = SpeakerManager::run_shell(COMMAND).await?;
+        Ok(res.stdout.trim().to_string())
+    }
+
+    /// 获取播放状态
+    pub async fn get_play_status() -> Result<String, AppError> {
+        const COMMAND: &str = r#"
+            mphelper mute_stat
+        "#;
+        let res = SpeakerManager::run_shell(COMMAND).await?;
+        let status = if res.stdout.contains("1") {
+            "playing"
+        } else if res.stdout.contains("2") {
+            "paused"
+        } else {
+            "idle"
+        };
+        Ok(status.to_string())
+    }
+
+    /// 播放
+    pub async fn play() -> Result<bool, AppError> {
+        const COMMAND: &str = r#"
+            mphelper play
+        "#;
+        let res = SpeakerManager::run_shell(COMMAND).await?;
+        Ok(res.stdout.contains("\"code\": 0"))
+    }
+
+    /// 暂停
+    pub async fn pause() -> Result<bool, AppError> {
+        const COMMAND: &str = r#"
+            mphelper pause
+        "#;
+        let res = SpeakerManager::run_shell(COMMAND).await?;
+        Ok(res.stdout.contains("\"code\": 0"))
+    }
+
+    /// TTS
+    pub async fn play_text(text: &str) -> Result<bool, AppError> {
+        const COMMAND: &str = r#"
+            /usr/sbin/tts_play2.sh '%s'
+        "#;
+        let script = COMMAND.replace("%s", text);
+        let res = SpeakerManager::run_shell(&script).await?;
+        Ok(res.stdout.contains("\"code\": 0"))
+    }
+
+    /// 播放音频
+    pub async fn play_url(url: &str) -> Result<bool, AppError> {
+        const COMMAND: &str = r#"
+            ubus call mediaplayer player_play_url '{"url":"%s","type": 1}'
+        "#;
+        let script = COMMAND.replace("%s", url);
+        let res = SpeakerManager::run_shell(&script).await?;
+        Ok(res.stdout.contains("\"code\": 0"))
+    }
+
+    /// 获取麦克风状态
+    pub async fn get_mic_status() -> Result<String, AppError> {
+        const COMMAND: &str = r#"
+            [ ! -f /tmp/mipns/mute ] && echo on || echo off
+        "#;
+        let res = SpeakerManager::run_shell(COMMAND).await?;
+        let status = if res.stdout.contains("on") {
+            "on"
+        } else {
+            "off"
+        };
+        Ok(status.to_string())
+    }
+
+    /// 打开麦克风
+    pub async fn mic_on() -> Result<bool, AppError> {
+        const COMMAND: &str = r#"
+            ubus -t1 -S call pnshelper event_notify '{"src":3, "event":7}' 2>&1
+        "#;
+        let res = SpeakerManager::run_shell(COMMAND).await?;
+        Ok(res.stdout.contains("\"code\":0"))
+    }
+
+    /// 关闭麦克风
+    pub async fn mic_off() -> Result<bool, AppError> {
+        const COMMAND: &str = r#"
+            ubus -t1 -S call pnshelper event_notify '{"src":3, "event":8}' 2>&1
+        "#;
+        let res = SpeakerManager::run_shell(COMMAND).await?;
+        Ok(res.stdout.contains("\"code\":0"))
+    }
+
+    /// 执行命令
+    pub async fn ask_xiaoai(text: &str) -> Result<bool, AppError> {
+        const COMMAND: &str = r#"
+            ubus call mibrain ai_service '{"tts":1,"nlp":1,"nlp_text":"%s"}'
+        "#;
+        let script = COMMAND.replace("%s", text);
+        let res = SpeakerManager::run_shell(&script).await?;
+        Ok(res.stdout.contains("\"code\": 0"))
+    }
+
+    /// 中断运行
+    pub async fn abort_xiaoai() -> Result<bool, AppError> {
+        const COMMAND: &str = r#"
+            /etc/init.d/mico_aivs_lab restart >/dev/null 2>&1
+        "#;
+        let res = SpeakerManager::run_shell(COMMAND).await?;
+        Ok(res.exit_code == 0)
+    }
+
+    /// 唤醒
+    pub async fn wake_up(flag: bool) -> Result<bool, AppError> {
+        let command = if flag {
+            r#"
+                ubus call pnshelper event_notify '{"src":1,"event":0}'
+            "#
+        } else {
+            r#"
+                ubus call pnshelper event_notify '{"src":3, "event":7}'
+                sleep 0.1
+                ubus call pnshelper event_notify '{"src":3, "event":8}'
+            "#
+        };
+        let res = SpeakerManager::run_shell(command).await?;
+        Ok(res.stdout.contains("\"code\": 0"))
+    }
+
+    async fn run_shell(script: &str) -> Result<CommandResult, AppError> {
+        let res = RPC::instance()
+            .call_remote("run_shell", Some(json!(script)), None)
+            .await?;
+        Ok(serde_json::from_value::<CommandResult>(res.data.unwrap()).unwrap())
+    }
+}

packages/client-rust/src/utils/event.rs

@@ -0,0 +1,81 @@
+use futures::future::BoxFuture;
+use std::{
+    collections::HashMap,
+    future::Future,
+    sync::{Arc, LazyLock},
+};
+use tokio::sync::Mutex;
+
+use crate::{base::AppError, services::connect::data::Event};
+
+use super::task::TaskManager;
+
+type EventCallback = Arc<dyn Fn(Event) -> BoxFuture<'static, Result<(), AppError>> + Send + Sync>;
+
+pub struct EventBus {
+    subscribers: Arc<Mutex<HashMap<String, Vec<EventCallback>>>>,
+}
+
+static INSTANCE: LazyLock<EventBus> = LazyLock::new(EventBus::new);
+
+impl EventBus {
+    fn new() -> Self {
+        EventBus {
+            subscribers: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    pub fn instance() -> &'static Self {
+        &INSTANCE
+    }
+
+    pub async fn subscribe<F, Fut>(&self, event: &str, callback: F)
+    where
+        F: Fn(Event) -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<(), AppError>> + Send + 'static,
+    {
+        let mut subscribers = self.subscribers.lock().await;
+        subscribers
+            .entry(event.to_string())
+            .or_insert_with(Vec::new)
+            .push(Arc::new(move |event| Box::pin(callback(event))));
+    }
+
+    pub async fn unsubscribe(&self, event: &str) {
+        self.subscribers.lock().await.remove(event);
+        let tag = format!("EventBus-{}", event);
+        TaskManager::instance().dispose(&tag).await;
+    }
+
+    pub async fn publish(&self, event: Event) {
+        let Some(callbacks) = self.get_callbacks(&event.event).await else {
+            return;
+        };
+        for callback in callbacks {
+            let event = event.clone();
+            let _ = callback(event).await;
+        }
+    }
+
+    pub async fn publish_async(&self, event: Event) {
+        let Some(callbacks) = self.get_callbacks(&event.event).await else {
+            return;
+        };
+        let tag = format!("EventBus-{}", event.event);
+        for callback in callbacks {
+            let event = event.clone();
+            let task = tokio::spawn(async move {
+                let _ = callback(event).await;
+            });
+            TaskManager::instance().add(&tag, task).await;
+        }
+    }
+
+    async fn get_callbacks(&self, event: &str) -> Option<Vec<EventCallback>> {
+        let subscribers = self.subscribers.lock().await;
+        match subscribers.get(event) {
+            Some(callbacks) => Some(callbacks.clone()),
+            None => None,
+        }
+    }
+}

packages/client-rust/src/utils/mod.rs

@@ -0,0 +1,3 @@
+pub mod event;
+pub mod shell;
+pub mod task;

packages/client-rust/src/utils/shell.rs

@@ -0,0 +1,29 @@
+use serde::{Deserialize, Serialize};
+use tokio::process::Command;
+
+use crate::base::AppError;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct CommandResult {
+    pub stdout: String,
+    pub stderr: String,
+    pub exit_code: i32,
+}
+
+pub async fn run_shell(script: &str) -> Result<CommandResult, AppError> {
+    let output = Command::new("/bin/sh")
+        .arg("-c")
+        .arg(script)
+        .output()
+        .await?;
+
+    let stdout = String::from_utf8_lossy(&output.stdout).to_string();
+    let stderr = String::from_utf8_lossy(&output.stderr).to_string();
+    let exit_code = output.status.code().unwrap_or(-1);
+
+    Ok(CommandResult {
+        stdout,
+        stderr,
+        exit_code,
+    })
+}

packages/client-rust/src/utils/task.rs

@@ -0,0 +1,58 @@
+use std::{
+    collections::HashMap,
+    future::Future,
+    sync::{Arc, LazyLock},
+};
+
+use tokio::{sync::Mutex, task::JoinHandle};
+
+/// 批量管理异步任务，在合适的时机终止
+pub struct TaskManager {
+    tasks: Arc<Mutex<HashMap<String, Vec<JoinHandle<()>>>>>,
+}
+
+static INSTANCE: LazyLock<TaskManager> = LazyLock::new(TaskManager::new);
+
+impl TaskManager {
+    fn new() -> Self {
+        Self {
+            tasks: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+
+    pub fn instance() -> &'static Self {
+        &INSTANCE
+    }
+
+    pub async fn add(&self, tag: &str, handle: JoinHandle<()>) {
+        let mut tasks = self.tasks.lock().await;
+        let handles = tasks.entry(tag.to_string()).or_insert_with(Vec::new);
+        handles.retain(|h| !h.is_finished());
+        handles.push(handle);
+    }
+
+    pub async fn dispose(&self, tag: &str) {
+        let mut tasks = self.tasks.lock().await;
+        if let Some(handles) = tasks.remove(tag) {
+            for handle in handles {
+                let _ = handle.abort();
+            }
+        }
+    }
+
+    pub async fn run_async<F>(future: F)
+    where
+        F: Future + Send + 'static,
+        F::Output: Send + 'static,
+    {
+        let task = tokio::spawn(async move {
+            let _ = future.await;
+        });
+
+        TaskManager::instance().add("async", task).await;
+    }
+
+    pub async fn dispose_async() {
+        TaskManager::instance().dispose("async").await;
+    }
+}