[agent] add publish test example, make it runnable with test

Author: timzaak

README.md

@@ -45,8 +45,12 @@ Install [Rust 1.70+](https://www.rust-lang.org/),
 [Node v18](https://nodejs.org/), [NPM v9](https://www.npmjs.com/), and
 [mprocs](https://github.com/pvolok/mprocs). Then, run
 ```shell
-cd web && npm install && cd ../
+cd web && npm ci && cd ../
+cd develop && ./run_mqtt.sh && cd ../
 mprocs
+
+# this is check MQTT agent if is OK
+cd agent && cargo run --example publish_command.rs
 ```
 
 ## 限制

agent/examples/publish_command.rs

@@ -0,0 +1,53 @@
+use std::path::PathBuf;
+use rumqttc::v5::{AsyncClient, MqttOptions, Incoming};
+use rumqttc::v5::mqttbytes::QoS;
+use tracing_subscriber::EnvFilter;
+use mproxy::message::{RequestMessage, ResponseMessage};
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+
+    let command = "ls";
+    let config_path:PathBuf = "./config.yml".parse().unwrap();
+
+    let is_atty = atty::is(atty::Stream::Stdout);
+    tracing_subscriber::fmt().with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::from("debug")),)
+        .with_ansi(is_atty).init();
+
+    let config = mproxy::config::Config::new(Some(config_path)).unwrap();
+
+    let mqtt_url = format!("{}?client_id={}", &config.server, "test_web");
+    let options = MqttOptions::parse_url(&mqtt_url).unwrap();
+    let (client, mut eventloop) = AsyncClient::new(options,20);
+
+    let topic = config.get_response_command_topic();
+    client.subscribe(topic, QoS::ExactlyOnce).await?;
+
+    tokio::spawn(async move {
+        loop {
+            let event = eventloop.poll().await;
+            match event {
+                Ok(event) =>  {
+                    match event {
+                        rumqttc::v5::Event::Incoming(Incoming::Publish(data)) => {
+                            let resp: ResponseMessage = serde_json::from_slice(data.payload.as_ref()).unwrap();
+
+                            match resp {
+                                ResponseMessage::Ok {data, seq,..} => println!("recv:{data}, seq: {seq}"),
+                                ResponseMessage::Err {message,..} => eprintln!("{message}"),
+                            }
+                        }
+                        _ => ()
+                    }
+                },
+                Err(err) => eprintln!("{:?}", err),
+            };
+        }
+    });
+    let command = RequestMessage::Cmd{command: command.to_string(), request_id: "test_request_id".to_string()};
+    let command = serde_json::to_vec(&command).unwrap();
+    client.publish(config.get_command_topic(), QoS::ExactlyOnce, false, command).await?;
+    tokio::signal::ctrl_c().await?;
+
+    Ok(())
+}

agent/src/connection.rs

@@ -7,7 +7,7 @@ use tokio::sync::broadcast;
 use tracing::{debug, error, info, warn};
 
 use crate::config::Config;
-use crate::handler::{RequestMessage, ResponseMessage};
+use crate::message::{RequestMessage, ResponseMessage};
 
 #[derive(Clone, Debug)]
 pub struct Connection {

agent/src/handler.rs

@@ -1,26 +1,9 @@
 use std::io::{BufRead, BufReader};
-use serde::{Deserialize, Serialize};
 use tokio::sync::broadcast;
 use tracing::{info};
 use crate::connection::Connection;
-use std::process::Command;
-
-
-#[derive(Serialize, Deserialize, Debug, Clone)]
-#[serde(tag = "type")]
-pub enum RequestMessage {
-    Cmd {
-        command: String,
-        request_id: String,
-    }
-}
-
-#[derive(Serialize, Deserialize, Debug, Clone)]
-#[serde(tag = "type")]
-pub enum ResponseMessage {
-    Ok{ request_id:String, seq:u32, data:String, pid:u32},
-    Err{ request_id:String, message:String}
-}
+use std::process::{Command, Stdio};
+use crate::message::{RequestMessage, ResponseMessage};
 
 
 type Receiver = broadcast::Receiver<(String,RequestMessage)>;
@@ -44,7 +27,9 @@ impl Handler {
             match cmd {
                 RequestMessage::Cmd { command,request_id } => {
                     let mut seq:u32 = 1;
-                    match Command::new(&command).spawn() {
+                    let mut command = Command::new(&command);
+                    command.stdout(Stdio::piped()).stderr(Stdio::piped());
+                    match command.spawn() {
                         Ok(mut child) => {
                             let pid = child.id();
                             if let Some(stdout) = child.stdout.take() {
@@ -70,7 +55,7 @@ impl Handler {
 #[cfg(test)]
 mod test {
     use std::io::{BufRead, BufReader};
-    use std::process::Command;
+    use std::process::{Command, Stdio};
     use std::time::Duration;
 
     use super::RequestMessage;
@@ -85,15 +70,19 @@ mod test {
 
     #[test]
     pub fn parse() {
-        let mut child = Command::new("top")
+        let mut cmd = Command::new("ls");
+        cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
+        let mut child = cmd
             .spawn().unwrap();
+
+
         if let Some(stdout) = child.stdout.take() {
             let reader = BufReader::new(stdout);
             reader.lines().filter_map(|line| line.ok()).for_each(|line| println!("{}", line));
         }
 
         println!("pid:{}",child.id());
-        std::thread::sleep(Duration::from_secs(10));
+        //std::thread::sleep(Duration::from_secs(10));
         child.kill().unwrap()
 
     }

agent/src/lib.rs

@@ -0,0 +1,27 @@
+pub mod config;
+pub mod message;
+mod connection;
+mod handler;
+
+use std::sync::Arc;
+use tokio::sync::broadcast;
+use crate::connection::Connection;
+use crate::handler::Handler;
+use crate::message::RequestMessage;
+
+pub async fn run(shutdown_rx:broadcast::Receiver<bool>, config: config::Config) -> anyhow::Result<()>{
+
+    let config = Arc::new(config);
+    //todo: 优化 Message clone 问题。 或是替换队列
+    let (cmd_tx, cmd_rx) = broadcast::channel::<(String, RequestMessage)>(5);
+
+    let (connection, event_loop) = Connection::connect(config.clone()).await?;
+    let mut handler = Handler::new(cmd_rx, connection, config.get_response_command_topic());
+    let shutdown_rx_1 = shutdown_rx.resubscribe();
+
+    tokio::spawn(async move {
+        handler.run(shutdown_rx_1).await;
+    });
+    Connection::loop_event(event_loop, shutdown_rx, cmd_tx).await;
+    Ok(())
+}

agent/src/main.rs

@@ -1,18 +1,13 @@
-mod config;
-mod connection;
-mod handler;
+
 
 use anyhow::Context;
 use clap::{Parser, ArgGroup};
 use std::path::PathBuf;
-use std::sync::Arc;
 use tokio::{signal, sync::broadcast};
 use tracing_subscriber::EnvFilter;
-use crate::connection::Connection;
-use crate::handler::{Handler, RequestMessage};
 
 #[derive(Parser, Debug)]
-#[clap(name = crate::config::APP_NAME, version = env!("CARGO_PKG_VERSION"))]
+#[clap(name = mproxy::config::APP_NAME, version = env!("CARGO_PKG_VERSION"))]
 #[clap(group(ArgGroup::new("cmds").required(true).args(&["CONFIG"]),))]
 struct Cli {
     #[clap(value_parser, name = "CONFIG")]
@@ -43,38 +38,18 @@ async fn main() -> anyhow::Result<()> {
     tracing_subscriber::fmt().with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::from("info")),)
         .with_ansi(is_atty).init();
 
-    let config = config::Config::new(ops.config_path).with_context(|| "Config can not load")?;
+    let config = mproxy::config::Config::new(ops.config_path).with_context(|| "Config can not load")?;
 
-    run(shutdown_rx, config).await?;
+    mproxy::run(shutdown_rx, config).await?;
     Ok(())
 }
 
 
-async fn run(shutdown_rx:broadcast::Receiver<bool>, config: config::Config) -> anyhow::Result<()>{
-
-    let config = Arc::new(config);
-    //todo: 优化 Message clone 问题。 或是替换队列
-    let (cmd_tx, cmd_rx) = broadcast::channel::<(String, RequestMessage)>(5);
-
-    let (connection, event_loop) = Connection::connect(config.clone()).await?;
-    let mut handler = Handler::new(cmd_rx, connection, config.get_response_command_topic());
-    let shutdown_rx_1 = shutdown_rx.resubscribe();
-
-    tokio::spawn(async move {
-        handler.run(shutdown_rx_1).await;
-    });
-    Connection::loop_event(event_loop, shutdown_rx, cmd_tx).await;
-    Ok(())
-
-
-
-}
-
 #[cfg(test)]
 mod test {
     use clap::Parser;
     use crate::Cli;
-    use crate::config::APP_NAME;
+    use mproxy::config::APP_NAME;
 
     #[test]
     fn parse_cli() {

agent/src/message.rs

@@ -0,0 +1,17 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+#[serde(tag = "type")]
+pub enum RequestMessage {
+    Cmd {
+        command: String,
+        request_id: String,
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+#[serde(tag = "type")]
+pub enum ResponseMessage {
+    Ok{ request_id:String, seq:u32, data:String, pid:u32},
+    Err{ request_id:String, message:String}
+}

mprocs.yaml

@@ -1,8 +1,10 @@
 procs:
-  mqtt:
-    shell: cd develop && ./run_mqtt.sh
+  #mqtt:
+  #  shell: cd develop && ./run_mqtt.sh
   mproxy:
-    shell: cd agent && cargo run -- config.toml
+    env:
+      RUST_LOG: debug
+    shell: cd agent && cargo run -- $(pwd)/config.yml
   web:
     shell: cd web && npm run dev
     stop: SIGKILL