Add delete kind

Cargo.lock

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

yazi-dds/Cargo.toml

@@ -13,12 +13,13 @@ yazi-boot   = { path = "../yazi-boot", version = "0.2.4" }
 yazi-shared = { path = "../yazi-shared", version = "0.2.4" }
 
 # External dependencies
-anyhow      = "1.0.81"
-mlua        = { version = "0.9.6", features = [ "lua54", "vendored" ] }
-parking_lot = "0.12.1"
-serde       = { version = "1.0.197", features = [ "derive" ] }
-serde_json  = "1.0.115"
-tokio       = { version = "1.37.0", features = [ "full" ] }
+anyhow       = "1.0.81"
+mlua         = { version = "0.9.6", features = [ "lua54", "vendored" ] }
+parking_lot  = "0.12.1"
+serde        = { version = "1.0.197", features = [ "derive" ] }
+serde_json   = "1.0.115"
+tokio        = { version = "1.37.0", features = [ "full" ] }
+tokio-stream = "0.1.15"
 
 # Logging
 tracing = { version = "0.1.40", features = [ "max_level_debug", "release_max_level_warn" ] }

yazi-dds/src/body/body.rs

@@ -2,7 +2,7 @@ use anyhow::Result;
 use mlua::{ExternalResult, IntoLua, Lua, Value};
 use serde::Serialize;
 
-use super::{BodyBulk, BodyCd, BodyCustom, BodyHey, BodyHi, BodyHover, BodyMove, BodyRename, BodyYank};
+use super::{BodyBulk, BodyCd, BodyCustom, BodyDelete, BodyHey, BodyHi, BodyHover, BodyMove, BodyRename, BodyYank};
 use crate::Payload;
 
 #[derive(Debug, Serialize)]
@@ -16,6 +16,7 @@ pub enum Body<'a> {
 	Bulk(BodyBulk<'a>),
 	Yank(BodyYank<'a>),
 	Move(BodyMove<'a>),
+	Delete(BodyDelete<'a>),
 	Custom(BodyCustom),
 }
 
@@ -30,6 +31,7 @@ impl<'a> Body<'a> {
 			"bulk" => Body::Bulk(serde_json::from_str(body)?),
 			"yank" => Body::Yank(serde_json::from_str(body)?),
 			"move" => Body::Move(serde_json::from_str(body)?),
+			"delete" => Body::Delete(serde_json::from_str(body)?),
 			_ => BodyCustom::from_str(kind, body)?,
 		})
 	}
@@ -57,6 +59,7 @@ impl<'a> Body<'a> {
 			Self::Bulk(_) => "bulk",
 			Self::Yank(_) => "yank",
 			Body::Move(_) => "move",
+			Body::Delete(_) => "delete",
 			Self::Custom(b) => b.kind.as_str(),
 		}
 	}
@@ -98,6 +101,7 @@ impl IntoLua<'_> for Body<'static> {
 			Body::Bulk(b) => b.into_lua(lua),
 			Body::Yank(b) => b.into_lua(lua),
 			Body::Move(b) => b.into_lua(lua),
+			Body::Delete(b) => b.into_lua(lua),
 			Body::Custom(b) => b.into_lua(lua),
 		}
 	}

yazi-dds/src/body/delete.rs

@@ -0,0 +1,38 @@
+use std::borrow::Cow;
+
+use mlua::{IntoLua, Lua, Value};
+use serde::{Deserialize, Serialize};
+use yazi_shared::fs::Url;
+
+use super::Body;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BodyDelete<'a> {
+	pub targets: Cow<'a, Vec<Url>>,
+}
+
+impl<'a> BodyDelete<'a> {
+	#[inline]
+	pub fn borrowed(targets: &'a Vec<Url>) -> Body<'a> {
+		Self { targets: Cow::Borrowed(targets) }.into()
+	}
+}
+
+impl BodyDelete<'static> {
+	#[inline]
+	pub fn owned(targets: Vec<Url>) -> Body<'static> { Self { targets: Cow::Owned(targets) }.into() }
+}
+
+impl<'a> From<BodyDelete<'a>> for Body<'a> {
+	fn from(value: BodyDelete<'a>) -> Self { Self::Delete(value) }
+}
+
+impl IntoLua<'_> for BodyDelete<'static> {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value<'_>> {
+		let t = lua.create_table_with_capacity(self.targets.len(), 0)?;
+		for (i, url) in self.targets.into_owned().into_iter().enumerate() {
+			t.raw_set(i + 1, lua.create_any_userdata(url)?)?;
+		}
+		t.into_lua(lua)
+	}
+}

yazi-dds/src/body/mod.rs

@@ -4,6 +4,7 @@ mod body;
 mod bulk;
 mod cd;
 mod custom;
+mod delete;
 mod hey;
 mod hi;
 mod hover;
@@ -15,6 +16,7 @@ pub use body::*;
 pub use bulk::*;
 pub use cd::*;
 pub use custom::*;
+pub use delete::*;
 pub use hey::*;
 pub use hi::*;
 pub use hover::*;

yazi-dds/src/body/move_.rs

@@ -8,21 +8,20 @@ use super::Body;
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct BodyMove<'a> {
-	pub from: Cow<'a, Url>,
-	pub to:   Cow<'a, Url>,
+	pub items: Cow<'a, Vec<BodyMoveItem>>,
 }
 
 impl<'a> BodyMove<'a> {
 	#[inline]
-	pub fn borrowed(from: &'a Url, to: &'a Url) -> Body<'a> {
-		Self { from: Cow::Borrowed(from), to: Cow::Borrowed(to) }.into()
+	pub fn borrowed(items: &'a Vec<BodyMoveItem>) -> Body<'a> {
+		Self { items: Cow::Borrowed(items) }.into()
 	}
 }
 
 impl BodyMove<'static> {
 	#[inline]
-	pub fn dummy(from: &Url, to: &Url) -> Body<'static> {
-		Self { from: Cow::Owned(from.clone()), to: Cow::Owned(to.clone()) }.into()
+	pub fn owned(items: Vec<BodyMoveItem>) -> Body<'static> {
+		Self { items: Cow::Owned(items) }.into()
 	}
 }
 
@@ -31,11 +30,22 @@ impl<'a> From<BodyMove<'a>> for Body<'a> {
 }
 
 impl IntoLua<'_> for BodyMove<'static> {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> { self.items.into_owned().into_lua(lua) }
+}
+
+// --- Item
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BodyMoveItem {
+	pub from: Url,
+	pub to:   Url,
+}
+
+impl IntoLua<'_> for BodyMoveItem {
 	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> {
 		lua
 			.create_table_from([
-				("from", lua.create_any_userdata(self.from.into_owned())?.into_lua(lua)?),
-				("to", lua.create_any_userdata(self.to.into_owned())?.into_lua(lua)?),
+				("from", lua.create_any_userdata(self.from)?),
+				("to", lua.create_any_userdata(self.to)?),
 			])?
 			.into_lua(lua)
 	}

yazi-dds/src/lib.rs

@@ -3,6 +3,7 @@ pub mod body;
 mod client;
 mod payload;
 mod pubsub;
+mod pump;
 mod sendable;
 mod server;
 mod state;
@@ -10,11 +11,12 @@ mod state;
 pub use client::*;
 pub use payload::*;
 pub use pubsub::*;
+pub use pump::*;
 pub use sendable::*;
 use server::*;
 pub use state::*;
 
-pub fn init() {
+pub fn serve() {
 	let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
 
 	// Client
@@ -30,5 +32,8 @@ pub fn init() {
 	LOCAL.with(Default::default);
 	REMOTE.with(Default::default);
 
+	Pump::serve();
 	Client::serve(rx);
 }
+
+pub fn shutdown() { Pump::shutdown(); }

yazi-dds/src/payload.rs

@@ -85,6 +85,7 @@ impl Display for Payload<'_> {
 			Body::Bulk(b) => serde_json::to_string(b),
 			Body::Yank(b) => serde_json::to_string(b),
 			Body::Move(b) => serde_json::to_string(b),
+			Body::Delete(b) => serde_json::to_string(b),
 			Body::Custom(b) => serde_json::to_string(b),
 		};
 

yazi-dds/src/pubsub.rs

@@ -5,7 +5,7 @@ use parking_lot::RwLock;
 use yazi_boot::BOOT;
 use yazi_shared::{fs::Url, RoCell};
 
-use crate::{body::{Body, BodyCd, BodyHi, BodyHover, BodyMove, BodyRename, BodyYank}, Client, ID, PEERS};
+use crate::{body::{Body, BodyCd, BodyDelete, BodyHi, BodyHover, BodyMove, BodyMoveItem, BodyRename, BodyYank}, Client, ID, PEERS};
 
 pub static LOCAL: RoCell<RwLock<HashMap<String, HashMap<String, Function<'static>>>>> =
 	RoCell::new();
@@ -143,15 +143,27 @@ impl Pubsub {
 		}
 	}
 
-	pub fn pub_from_move(from: &Url, to: &Url) {
-		if LOCAL.read().contains_key("move") {
-			Self::pub_(BodyMove::dummy(from, to));
-		}
+	pub(super) fn pub_from_move(items: Vec<BodyMoveItem>) {
 		if PEERS.read().values().any(|p| p.able("move")) {
-			Client::push(BodyMove::borrowed(from, to));
+			Client::push(BodyMove::borrowed(&items));
 		}
 		if BOOT.local_events.contains("move") {
-			BodyMove::borrowed(from, to).with_receiver(*ID).flush();
+			BodyMove::borrowed(&items).with_receiver(*ID).flush();
+		}
+		if LOCAL.read().contains_key("move") {
+			Self::pub_(BodyMove::owned(items));
+		}
+	}
+
+	pub(super) fn pub_from_delete(targets: Vec<Url>) {
+		if PEERS.read().values().any(|p| p.able("delete")) {
+			Client::push(BodyDelete::borrowed(&targets));
+		}
+		if BOOT.local_events.contains("delete") {
+			BodyDelete::borrowed(&targets).with_receiver(*ID).flush();
+		}
+		if LOCAL.read().contains_key("delete") {
+			Self::pub_(BodyDelete::owned(targets));
 		}
 	}
 }

yazi-dds/src/pump.rs

@@ -0,0 +1,51 @@
+use std::time::Duration;
+
+use tokio::{pin, select, sync::mpsc};
+use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt};
+use yazi_shared::{fs::Url, RoCell};
+
+use crate::{body::BodyMoveItem, Pubsub};
+
+static MOVE_TX: RoCell<mpsc::UnboundedSender<BodyMoveItem>> = RoCell::new();
+static DELETE_TX: RoCell<mpsc::UnboundedSender<Url>> = RoCell::new();
+
+pub struct Pump;
+
+impl Pump {
+	#[inline]
+	pub fn push_move(from: Url, to: Url) { MOVE_TX.send(BodyMoveItem { from, to }).ok(); }
+
+	#[inline]
+	pub fn push_delete(target: Url) { DELETE_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);
+
+		tokio::spawn(async move {
+			let move_rx =
+				UnboundedReceiverStream::new(move_rx).chunks_timeout(1000, Duration::from_millis(500));
+			let delete_rx =
+				UnboundedReceiverStream::new(delete_rx).chunks_timeout(1000, Duration::from_millis(500));
+
+			pin!(move_rx);
+			pin!(delete_rx);
+
+			loop {
+				select! {
+					Some(items) = move_rx.next() => Pubsub::pub_from_move(items),
+					Some(targets) = delete_rx.next() => Pubsub::pub_from_delete(targets),
+					else => break,
+				}
+			}
+		});
+	}
+
+	pub(super) fn shutdown() {
+		MOVE_TX.drop();
+		DELETE_TX.drop();
+	}
+}

yazi-fm/src/app/commands/quit.rs

@@ -7,6 +7,7 @@ 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::STATE.drain()).ok();

yazi-fm/src/main.rs

@@ -49,7 +49,7 @@ async fn main() -> anyhow::Result<()> {
 
 	yazi_proxy::init();
 
-	yazi_dds::init();
+	yazi_dds::serve();
 
 	yazi_plugin::init();
 

yazi-scheduler/src/file/file.rs

@@ -5,7 +5,6 @@ use futures::{future::BoxFuture, FutureExt};
 use tokio::{fs, io::{self, ErrorKind::{AlreadyExists, NotFound}}, sync::mpsc};
 use tracing::warn;
 use yazi_config::TASKS;
-use yazi_dds::Pubsub;
 use yazi_shared::fs::{accessible, calculate_size, copy_with_progress, path_relative_to, Url};
 
 use super::{FileOp, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash};
@@ -38,7 +37,6 @@ impl File {
 						Ok(0) => {
 							if task.cut {
 								fs::remove_file(&task.from).await.ok();
-								Pubsub::pub_from_move(&task.from, &task.to);
 							}
 							break;
 						}

yazi-scheduler/src/scheduler.rs

@@ -4,7 +4,7 @@ use futures::{future::BoxFuture, FutureExt};
 use parking_lot::Mutex;
 use tokio::{fs, select, sync::{mpsc::{self, UnboundedReceiver}, oneshot}, task::JoinHandle};
 use yazi_config::{open::Opener, plugin::PluginRule, TASKS};
-use yazi_dds::ValueSendable;
+use yazi_dds::{Pump, ValueSendable};
 use yazi_shared::{fs::{unique_path, Url}, Throttle};
 
 use super::{Ongoing, TaskProg, TaskStage};
@@ -72,13 +72,14 @@ impl Scheduler {
 		let id = ongoing.add(TaskKind::User, format!("Cut {:?} to {:?}", from, to));
 
 		ongoing.hooks.insert(id, {
-			let from = from.clone();
 			let ongoing = self.ongoing.clone();
+			let (from, to) = (from.clone(), to.clone());
 
 			Box::new(move |canceled: bool| {
 				async move {
 					if !canceled {
 						File::remove_empty_dirs(&from).await;
+						Pump::push_move(from, to);
 					}
 					ongoing.lock().try_remove(id, TaskStage::Hooked);
 				}
@@ -147,7 +148,8 @@ impl Scheduler {
 			Box::new(move |canceled: bool| {
 				async move {
 					if !canceled {
-						fs::remove_dir_all(target).await.ok();
+						fs::remove_dir_all(&target).await.ok();
+						Pump::push_delete(target);
 					}
 					ongoing.lock().try_remove(id, TaskStage::Hooked);
 				}