将日志通过WebSocket发送出去

src/lib.rs

@@ -1,11 +1,24 @@
-use ::log::{debug, error, info, trace, warn};
-use encoding_rs::Encoding;
-use tokio::runtime::Runtime;
+use crate::protobuf::gou::{MessageType, Starpoles};
 use crate::utils::cxx_string_to_string;
+use ::log::{debug, error, info, trace, warn};
+use bytes::Bytes;
+use cxx::CxxString;
+use encoding_rs::Encoding;
+use futures_util::SinkExt;
+use prost::Message as WebMessage;
+use spin::Mutex;
+use std::sync::mpsc;
+use tokio::net::TcpStream;
+use tokio::runtime::Runtime;
+use tokio_tungstenite::tungstenite::Message;
+use std::sync::OnceLock;
 
+static LOGGER_SENDER: OnceLock<mpsc::Sender<Starpoles>> = OnceLock::new();
+
+mod config;
 mod log;
 mod network;
-mod config;
+pub mod protobuf;
 pub mod utils;
 
 #[cxx::bridge]
@@ -15,7 +28,7 @@ mod ffi {
 
         fn http_get(url: &CxxString) -> Result<String>;
 
-        fn init_log(is_debug: bool);
+        fn init_log(is_debug: bool, ws_uel: &CxxString);
 
         fn log_trace(msg: &CxxString);
 
@@ -31,30 +44,85 @@ mod ffi {
 
 fn log_error(msg: &cxx::CxxString) {
     let msg = cxx_string_to_string(msg);
-    error!("{}", msg);
+    let lock = LOGGER_SENDER.get().unwrap();
+    let _ = lock.send(Starpoles {
+        r#type: MessageType::Error.into(),
+        message: msg,
+    });
+    //error!("{}", msg);
 }
 
 fn log_warning(msg: &cxx::CxxString) {
     let msg = cxx_string_to_string(msg);
-    warn!("{}", msg);
+    let lock = LOGGER_SENDER.get().unwrap();
+    let _ = lock.send(Starpoles {
+        r#type: MessageType::Warning.into(),
+        message: msg,
+    });
 }
 
 fn log_info(msg: &cxx::CxxString) {
     let msg = cxx_string_to_string(msg);
-    info!("{}", msg);
+    let lock = LOGGER_SENDER.get().unwrap();
+    let _ = lock.send(Starpoles {
+        r#type: MessageType::Info.into(),
+        message: msg,
+    });
 }
 
 fn log_debug(msg: &cxx::CxxString) {
     let msg = cxx_string_to_string(msg);
-    debug!("{}", msg);
+    let lock = LOGGER_SENDER.get().unwrap();
+    let _ = lock.send(Starpoles {
+        r#type: MessageType::Error.into(),
+        message: msg,
+    });
 }
 
 fn log_trace(msg: &cxx::CxxString) {
     let msg = cxx_string_to_string(msg);
-    trace!("{}", msg);
+    let lock = LOGGER_SENDER.get().unwrap();
+    let _ = lock.send(Starpoles {
+        r#type: MessageType::Trace.into(),
+        message: msg,
+    });
 }
 
-fn init_log(is_debug: bool) {
+fn init_log(is_debug: bool, ws_uel: &CxxString) {
+    let url = cxx_string_to_string(ws_uel);
+    let (tx, rx) = mpsc::channel::<Starpoles>();
+    LOGGER_SENDER.set(tx).expect("已经初始化过了");
+    std::thread::spawn(move || {
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+
+        rt.block_on(async move {
+            loop {
+                match tokio_tungstenite::connect_async(url.clone()).await {
+                    Ok((mut stream, _resp)) => {
+                        while let Ok(line) = rx.recv() {
+                            let mut buffer = vec![];
+                            //  Vec总是具备足够的空间
+                            line.encode(&mut buffer).unwrap();
+                            if stream
+                                .send(Message::Binary(Bytes::from(buffer)))
+                                .await
+                                .is_err()
+                            {
+                                break;
+                            }
+                        }
+                    }
+                    Err(_) => {
+                        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+                    }
+                }
+            }
+        });
+    });
+
     log::init_log(is_debug);
 }
 
@@ -78,5 +146,3 @@ fn http_get(url: &cxx::CxxString) -> Result<String, Box<dyn std::error::Error>>
     let url = cxx_string_to_string(url);
     rt.block_on(network::http_get(&url))
 }
-
-