diff --git a/Cargo.lock b/Cargo.lock index 676ca9c6..0b8cc07a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2774,6 +2774,7 @@ dependencies = [ "serde_json", "tokio", "tokio-stream", + "tokio-util", "tracing", "uzers", "yazi-boot", diff --git a/yazi-dds/Cargo.toml b/yazi-dds/Cargo.toml index 49071c13..51307cc9 100644 --- a/yazi-dds/Cargo.toml +++ b/yazi-dds/Cargo.toml @@ -20,6 +20,7 @@ serde = { version = "1.0.197", features = [ "derive" ] } serde_json = "1.0.115" tokio = { version = "1.37.0", features = [ "full" ] } tokio-stream = "0.1.15" +tokio-util = "0.7.10" # Logging tracing = { version = "0.1.40", features = [ "max_level_debug", "release_max_level_warn" ] } diff --git a/yazi-dds/src/lib.rs b/yazi-dds/src/lib.rs index b9479569..14f7f4e9 100644 --- a/yazi-dds/src/lib.rs +++ b/yazi-dds/src/lib.rs @@ -36,4 +36,4 @@ pub fn serve() { Client::serve(rx); } -pub fn shutdown() { Pump::shutdown(); } +pub async fn shutdown() { Pump::shutdown().await; } diff --git a/yazi-dds/src/pump.rs b/yazi-dds/src/pump.rs index 0f339ad6..afabdadc 100644 --- a/yazi-dds/src/pump.rs +++ b/yazi-dds/src/pump.rs @@ -1,29 +1,41 @@ use std::time::Duration; +use parking_lot::Mutex; use tokio::{pin, select, sync::mpsc}; use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt}; +use tokio_util::sync::CancellationToken; use yazi_shared::{fs::Url, RoCell}; use crate::{body::BodyMoveItem, Pubsub}; -static MOVE_TX: RoCell> = RoCell::new(); -static DELETE_TX: RoCell> = RoCell::new(); +static CT: RoCell = RoCell::new(); +static MOVE_TX: Mutex>> = Mutex::new(None); +static DELETE_TX: Mutex>> = Mutex::new(None); pub struct Pump; impl Pump { #[inline] - pub fn push_move(from: Url, to: Url) { MOVE_TX.send(BodyMoveItem { from, to }).ok(); } + pub fn push_move(from: Url, to: Url) { + if let Some(tx) = &*MOVE_TX.lock() { + tx.send(BodyMoveItem { from, to }).ok(); + } + } #[inline] - pub fn push_delete(target: Url) { DELETE_TX.send(target).ok(); } + pub fn push_delete(target: Url) { + if let Some(tx) = &*DELETE_TX.lock() { + tx.send(target).ok(); + } + } pub(super) fn serve() { let (move_tx, move_rx) = mpsc::unbounded_channel(); let (delete_tx, delete_rx) = mpsc::unbounded_channel(); - MOVE_TX.init(move_tx); - DELETE_TX.init(delete_tx); + CT.with(Default::default); + MOVE_TX.lock().replace(move_tx); + DELETE_TX.lock().replace(delete_tx); tokio::spawn(async move { let move_rx = @@ -38,14 +50,18 @@ impl Pump { select! { Some(items) = move_rx.next() => Pubsub::pub_from_move(items), Some(targets) = delete_rx.next() => Pubsub::pub_from_delete(targets), - else => break, + else => { + CT.cancel(); + break; + }, } } }); } - pub(super) fn shutdown() { - MOVE_TX.drop(); - DELETE_TX.drop(); + pub(super) async fn shutdown() { + drop(MOVE_TX.lock().take()); + drop(DELETE_TX.lock().take()); + CT.cancelled().await; } } diff --git a/yazi-fm/src/app/commands/quit.rs b/yazi-fm/src/app/commands/quit.rs index 61f5440d..f7b7ec3f 100644 --- a/yazi-fm/src/app/commands/quit.rs +++ b/yazi-fm/src/app/commands/quit.rs @@ -7,9 +7,9 @@ use crate::app::App; impl App { pub(crate) fn quit(&mut self, opt: EventQuit) -> ! { - yazi_dds::shutdown(); self.cx.tasks.shutdown(); self.cx.manager.shutdown(); + futures::executor::block_on(yazi_dds::shutdown()); futures::executor::block_on(yazi_dds::STATE.drain()).ok(); if !opt.no_cwd_file {