feat(core): stop core process when exit

venus-core/src/error.rs

@@ -2,7 +2,7 @@ use std::{borrow::Cow, fmt::Display};
 
 use log::error;
 
-use crate::config::error::ConfigError;
+use crate::{config::error::ConfigError, message::MessageType};
 
 #[derive(thiserror::Error, Debug)]
 pub enum SubscriptionError {
@@ -30,6 +30,8 @@ pub enum VenusError {
     Decode(#[from] base64::DecodeError),
     #[error("from utf8 error {0}")]
     Utf8(#[from] std::string::FromUtf8Error),
+    #[error("subscription error {0:?}")]
+    Channel(#[from] std::sync::mpsc::SendError<MessageType>),
     #[error("subscription error {0}")]
     Subscription(#[from] SubscriptionError),
 }

venus-core/src/lib.rs

@@ -142,6 +142,7 @@ impl VenusCore for Venus {
     /// Kill core process if exist
     fn kill_core(&mut self) -> VenusResult<()> {
         if let Some(core) = self.child.as_mut() {
+            self.message_tx.send(MessageType::Terminate)?;
             core.kill()?;
             self.child = None;
             Ok(())

venus-core/src/message.rs

@@ -6,4 +6,6 @@ pub type Message = (Sender<MessageType>, Receiver<MessageType>);
 pub enum MessageType {
     /// Message from v2ray core process
     Core(String),
+    /// Shutdown signal, the core process will exit
+    Terminate,
 }

venus/src/main.rs

@@ -1,6 +1,7 @@
 use core::{global_core, global_message};
 use std::{env, error::Error, net::SocketAddr};
 
+use anyhow::Context;
 use axum::Router;
 use consts::{DEFAULT_PORT, RUA_COMPILER};
 use dotenvy::dotenv;
@@ -24,35 +25,40 @@ async fn main() -> Result<()> {
     dotenv().ok();
     init_logger();
 
-    tokio::spawn(async move {
+    {
         info!("venus {RUA_COMPILER}");
         let venus = &mut global_core().await.lock().await;
         info!("string core");
         venus
             .config
             .reload_rua()
-            .expect("reading venus configuration failed");
+            .with_context(|| "reading venus configuration failed")?;
         venus
             .config
             .reload_core()
-            .expect("reading core configuration failed");
+            .with_context(|| "reading core configuration failed")?;
         venus
             .config
             .write_core()
-            .expect("write core configuration failed");
-        venus.spawn_core().expect("staring core failed");
-
+            .with_context(|| "write core configuration failed")?;
+        venus.spawn_core().with_context(|| "staring core failed")?;
+    }
+    tokio::spawn(async move {
         // global message handler
         let child_rx = &global_message().await.lock().await.1;
+        let core_span = span!(Level::INFO, "CORE").entered();
         while let Ok(msg) = child_rx.recv() {
             match msg {
                 MessageType::Core(msg) => {
-                    let core_span = span!(Level::INFO, "CORE").entered();
                     info!("{msg}");
-                    core_span.exit();
+                }
+                MessageType::Terminate => {
+                    info!("core stopping");
+                    break;
                 }
             }
         }
+        core_span.exit();
     });
 
     let port = env::var("VENUS_PORT")

venus/src/utils/mod.rs

@@ -1,7 +1,5 @@
-use std::thread;
-
 use tokio::signal;
-use tracing::{debug, info};
+use tracing::info;
 use tracing_subscriber::{fmt, prelude::*, registry, EnvFilter};
 use venus_core::{error::log_err, VenusCore};
 
@@ -111,7 +109,6 @@ where
     tokio::select! {
         _ = ctrl_c => {
             shutdown_cb()
-            // let _ = stop_core().map_err(log_err);
         },
         _ = terminate => {
             shutdown_cb()