From 76a11043e6ff502a52a7de1680dd5bb6ce401622 Mon Sep 17 00:00:00 2001 From: dev0 Date: Sun, 31 Dec 2023 06:34:08 +1100 Subject: [PATCH] wait a bit --- clash_lib/src/app/api/handlers/config.rs | 15 ++++++++---- clash_lib/src/app/logging.rs | 3 ++- clash_lib/src/lib.rs | 29 +++++++++++++++--------- 3 files changed, 31 insertions(+), 16 deletions(-) diff --git a/clash_lib/src/app/api/handlers/config.rs b/clash_lib/src/app/api/handlers/config.rs index d6f727fcd..c25c5ce3f 100644 --- a/clash_lib/src/app/api/handlers/config.rs +++ b/clash_lib/src/app/api/handlers/config.rs @@ -96,13 +96,17 @@ async fn update_configs( State(state): State, Json(req): Json, ) -> impl IntoResponse { + let (done, wait) = tokio::sync::oneshot::channel(); let g = state.global_state.lock().await; match (req.path, req.payload) { (_, Some(payload)) => { let msg = "config reloading from payload".to_string(); let cfg = crate::Config::Str(payload); - match g.reload_tx.send(cfg).await { - Ok(_) => (StatusCode::NO_CONTENT, msg).into_response(), + match g.reload_tx.send((cfg, done)).await { + Ok(_) => { + wait.await.unwrap(); + (StatusCode::NO_CONTENT, msg).into_response() + } Err(_) => ( StatusCode::INTERNAL_SERVER_ERROR, "could not signal config reload", @@ -127,8 +131,11 @@ async fn update_configs( let msg = format!("config reloading from file {}", path); let cfg: crate::Config = crate::Config::File(path); - match g.reload_tx.send(cfg).await { - Ok(_) => (StatusCode::NO_CONTENT, msg).into_response(), + match g.reload_tx.send((cfg, done)).await { + Ok(_) => { + wait.await.unwrap(); + (StatusCode::NO_CONTENT, msg).into_response() + } Err(_) => ( StatusCode::INTERNAL_SERVER_ERROR, diff --git a/clash_lib/src/app/logging.rs b/clash_lib/src/app/logging.rs index fa7e5b26c..ff6338ed8 100644 --- a/clash_lib/src/app/logging.rs +++ b/clash_lib/src/app/logging.rs @@ -140,7 +140,8 @@ pub fn setup_logging( .with( tracing_subscriber::fmt::Layer::new() .with_ansi(std::io::stdout().is_terminal()) - .pretty() + .compact() + .with_target(false) .with_file(true) .with_line_number(true) .with_writer(move || -> Box { Box::new(W(appender.clone())) }) diff --git a/clash_lib/src/lib.rs b/clash_lib/src/lib.rs index d40549a9a..389d6f337 100644 --- a/clash_lib/src/lib.rs +++ b/clash_lib/src/lib.rs @@ -22,7 +22,7 @@ use std::io; use std::path::PathBuf; use std::sync::{Arc, OnceLock}; use thiserror::Error; -use tokio::sync::{broadcast, mpsc, Mutex}; +use tokio::sync::{broadcast, mpsc, oneshot, Mutex}; use tokio::task::JoinHandle; use tracing::debug; use tracing::error; @@ -96,7 +96,7 @@ pub struct GlobalState { tunnel_listener_handle: Option>>, api_listener_handle: Option>>, dns_listener_handle: Option>>, - reload_tx: mpsc::Sender, + reload_tx: mpsc::Sender<(Config, oneshot::Sender<()>)>, cwd: String, } @@ -301,7 +301,7 @@ async fn start_async(opts: Options) -> Result<(), Error> { })); tasks.push(Box::pin(async move { - while let Some(config) = reload_rx.recv().await { + while let Some((config, done)) = reload_rx.recv().await { info!("reloading config"); let config = match config.try_parse() { Ok(c) => c, @@ -395,6 +395,9 @@ async fn start_async(opts: Options) -> Result<(), Error> { authenticator, )?)); + done.send(()).unwrap(); + + debug!("stopping listeners"); let mut g = global_state.lock().await; if let Some(h) = g.inbound_listener_handle.take() { h.abort(); @@ -409,11 +412,15 @@ async fn start_async(opts: Options) -> Result<(), Error> { h.abort(); } - let inbound_runner = inbound_manager.lock().await.get_runner()?; - let inbound_listener_handle = tokio::spawn(inbound_runner); + let inbound_listener_handle = inbound_manager + .lock() + .await + .get_runner() + .map(tokio::spawn)?; - let tun_runner = get_tun_runner(config.tun, dispatcher.clone(), dns_resolver.clone())?; - let tun_runner_handle = tun_runner.map(tokio::spawn); + let tun_runner_handle = + get_tun_runner(config.tun, dispatcher.clone(), dns_resolver.clone())? + .map(tokio::spawn); debug!("reloading dns listener"); let dns_listener_handle = dns::get_dns_listener(config.dns, dns_resolver.clone()) @@ -421,7 +428,7 @@ async fn start_async(opts: Options) -> Result<(), Error> { .map(tokio::spawn); debug!("reloading api listener"); - let api_runner = app::api::get_api_runner( + let api_listener_handle = app::api::get_api_runner( config.general.controller, log_tx.clone(), inbound_manager.clone(), @@ -433,13 +440,13 @@ async fn start_async(opts: Options) -> Result<(), Error> { cache_store, router, cwd.to_string_lossy().to_string(), - ); - let api_listener_handle = api_runner.map(tokio::spawn); + ) + .map(tokio::spawn); g.inbound_listener_handle = Some(inbound_listener_handle); g.tunnel_listener_handle = tun_runner_handle; g.dns_listener_handle = dns_listener_handle; - g.api_listener_handle = api_listener_handle + g.api_listener_handle = api_listener_handle; } Ok(()) }));