..

Cargo.lock

@@ -2774,6 +2774,7 @@ dependencies = [
  "serde_json",
  "tokio",
  "tokio-stream",
+ "tokio-util",
  "tracing",
  "uzers",
  "yazi-boot",

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" ] }

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; }

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<mpsc::UnboundedSender<BodyMoveItem>> = RoCell::new();
+static CT: RoCell<CancellationToken> = RoCell::new();
-static DELETE_TX: RoCell<mpsc::UnboundedSender<Url>> = RoCell::new();
+static MOVE_TX: Mutex<Option<mpsc::UnboundedSender<BodyMoveItem>>> = Mutex::new(None);
+static DELETE_TX: Mutex<Option<mpsc::UnboundedSender<Url>>> = 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);
+		CT.with(Default::default);
-		DELETE_TX.init(delete_tx);
+		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() {
+	pub(super) async fn shutdown() {
-		MOVE_TX.drop();
+		drop(MOVE_TX.lock().take());
-		DELETE_TX.drop();
+		drop(DELETE_TX.lock().take());
+		CT.cancelled().await;
 	}
 }

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 {