feat: add new move, trash, and delete event kinds to DDS (#880)

Co-authored-by: 三咲雅 · Misaki Masa <sxyazi@gmail.com>

.github/ISSUE_TEMPLATE/bug.yml

@@ -25,7 +25,7 @@ body:
   - type: dropdown
     id: tried_main
     attributes:
-      label: Did you try the latest main branch to see if the problem got fixed?
+      label: Did you try the latest code to see if this problem got fixed?
       options:
         - Tried, but the problem still
         - Not tried, and I'll explain why below

Cargo.lock

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

yazi-dds/Cargo.toml

@@ -13,12 +13,14 @@ 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"
+tokio-util   = "0.7.10"
 
 # 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, BodyRename, BodyYank};
+use super::{BodyBulk, BodyCd, BodyCustom, BodyDelete, BodyHey, BodyHi, BodyHover, BodyMove, BodyRename, BodyTrash, BodyYank};
 use crate::Payload;
 
 #[derive(Debug, Serialize)]
@@ -15,6 +15,9 @@ pub enum Body<'a> {
 	Rename(BodyRename<'a>),
 	Bulk(BodyBulk<'a>),
 	Yank(BodyYank<'a>),
+	Move(BodyMove<'a>),
+	Trash(BodyTrash<'a>),
+	Delete(BodyDelete<'a>),
 	Custom(BodyCustom),
 }
 
@@ -28,6 +31,9 @@ impl<'a> Body<'a> {
 			"rename" => Body::Rename(serde_json::from_str(body)?),
 			"bulk" => Body::Bulk(serde_json::from_str(body)?),
 			"yank" => Body::Yank(serde_json::from_str(body)?),
+			"move" => Body::Move(serde_json::from_str(body)?),
+			"trash" => Body::Trash(serde_json::from_str(body)?),
+			"delete" => Body::Delete(serde_json::from_str(body)?),
 			_ => BodyCustom::from_str(kind, body)?,
 		})
 	}
@@ -54,6 +60,9 @@ impl<'a> Body<'a> {
 			Self::Rename(_) => "rename",
 			Self::Bulk(_) => "bulk",
 			Self::Yank(_) => "yank",
+			Body::Move(_) => "move",
+			Body::Trash(_) => "trash",
+			Body::Delete(_) => "delete",
 			Self::Custom(b) => b.kind.as_str(),
 		}
 	}
@@ -94,6 +103,9 @@ impl IntoLua<'_> for Body<'static> {
 			Body::Rename(b) => b.into_lua(lua),
 			Body::Bulk(b) => b.into_lua(lua),
 			Body::Yank(b) => b.into_lua(lua),
+			Body::Move(b) => b.into_lua(lua),
+			Body::Trash(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,36 @@
+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 urls: Cow<'a, Vec<Url>>,
+}
+
+impl<'a> BodyDelete<'a> {
+	#[inline]
+	pub fn borrowed(urls: &'a Vec<Url>) -> Body<'a> { Self { urls: Cow::Borrowed(urls) }.into() }
+}
+
+impl BodyDelete<'static> {
+	#[inline]
+	pub fn owned(urls: Vec<Url>) -> Body<'static> { Self { urls: Cow::Owned(urls) }.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.urls.len(), 0)?;
+		for (i, url) in self.urls.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,18 +4,24 @@ mod body;
 mod bulk;
 mod cd;
 mod custom;
+mod delete;
 mod hey;
 mod hi;
 mod hover;
+mod move_;
 mod rename;
+mod trash;
 mod yank;
 
 pub use body::*;
 pub use bulk::*;
 pub use cd::*;
 pub use custom::*;
+pub use delete::*;
 pub use hey::*;
 pub use hi::*;
 pub use hover::*;
+pub use move_::*;
 pub use rename::*;
+pub use trash::*;
 pub use yank::*;

yazi-dds/src/body/move_.rs

@@ -0,0 +1,52 @@
+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 BodyMove<'a> {
+	pub items: Cow<'a, Vec<BodyMoveItem>>,
+}
+
+impl<'a> BodyMove<'a> {
+	#[inline]
+	pub fn borrowed(items: &'a Vec<BodyMoveItem>) -> Body<'a> {
+		Self { items: Cow::Borrowed(items) }.into()
+	}
+}
+
+impl BodyMove<'static> {
+	#[inline]
+	pub fn owned(items: Vec<BodyMoveItem>) -> Body<'static> {
+		Self { items: Cow::Owned(items) }.into()
+	}
+}
+
+impl<'a> From<BodyMove<'a>> for Body<'a> {
+	fn from(value: BodyMove<'a>) -> Self { Self::Move(value) }
+}
+
+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)?),
+				("to", lua.create_any_userdata(self.to)?),
+			])?
+			.into_lua(lua)
+	}
+}

yazi-dds/src/body/trash.rs

@@ -0,0 +1,36 @@
+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 BodyTrash<'a> {
+	pub urls: Cow<'a, Vec<Url>>,
+}
+
+impl<'a> BodyTrash<'a> {
+	#[inline]
+	pub fn borrowed(urls: &'a Vec<Url>) -> Body<'a> { Self { urls: Cow::Borrowed(urls) }.into() }
+}
+
+impl BodyTrash<'static> {
+	#[inline]
+	pub fn owned(urls: Vec<Url>) -> Body<'static> { Self { urls: Cow::Owned(urls) }.into() }
+}
+
+impl<'a> From<BodyTrash<'a>> for Body<'a> {
+	fn from(value: BodyTrash<'a>) -> Self { Self::Trash(value) }
+}
+
+impl IntoLua<'_> for BodyTrash<'static> {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value<'_>> {
+		let t = lua.create_table_with_capacity(self.urls.len(), 0)?;
+		for (i, url) in self.urls.into_owned().into_iter().enumerate() {
+			t.raw_set(i + 1, lua.create_any_userdata(url)?)?;
+		}
+		t.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 async fn shutdown() { Pump::shutdown().await; }

yazi-dds/src/payload.rs

@@ -84,6 +84,9 @@ impl Display for Payload<'_> {
 			Body::Rename(b) => serde_json::to_string(b),
 			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::Trash(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, BodyRename, BodyYank}, Client, ID, PEERS};
+use crate::{body::{Body, BodyCd, BodyDelete, BodyHi, BodyHover, BodyMove, BodyMoveItem, BodyRename, BodyTrash, BodyYank}, Client, ID, PEERS};
 
 pub static LOCAL: RoCell<RwLock<HashMap<String, HashMap<String, Function<'static>>>>> =
 	RoCell::new();
@@ -142,4 +142,40 @@ impl Pubsub {
 			BodyYank::borrowed(cut, urls).with_receiver(*ID).flush();
 		}
 	}
+
+	pub(super) fn pub_from_move(items: Vec<BodyMoveItem>) {
+		if PEERS.read().values().any(|p| p.able("move")) {
+			Client::push(BodyMove::borrowed(&items));
+		}
+		if BOOT.local_events.contains("move") {
+			BodyMove::borrowed(&items).with_receiver(*ID).flush();
+		}
+		if LOCAL.read().contains_key("move") {
+			Self::pub_(BodyMove::owned(items));
+		}
+	}
+
+	pub(super) fn pub_from_trash(urls: Vec<Url>) {
+		if PEERS.read().values().any(|p| p.able("trash")) {
+			Client::push(BodyTrash::borrowed(&urls));
+		}
+		if BOOT.local_events.contains("trash") {
+			BodyTrash::borrowed(&urls).with_receiver(*ID).flush();
+		}
+		if LOCAL.read().contains_key("trash") {
+			Self::pub_(BodyTrash::owned(urls));
+		}
+	}
+
+	pub(super) fn pub_from_delete(urls: Vec<Url>) {
+		if PEERS.read().values().any(|p| p.able("delete")) {
+			Client::push(BodyDelete::borrowed(&urls));
+		}
+		if BOOT.local_events.contains("delete") {
+			BodyDelete::borrowed(&urls).with_receiver(*ID).flush();
+		}
+		if LOCAL.read().contains_key("delete") {
+			Self::pub_(BodyDelete::owned(urls));
+		}
+	}
 }

yazi-dds/src/pump.rs

@@ -0,0 +1,82 @@
+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 CT: RoCell<CancellationToken> = RoCell::new();
+static MOVE_TX: Mutex<Option<mpsc::UnboundedSender<BodyMoveItem>>> = Mutex::new(None);
+static TRASH_TX: Mutex<Option<mpsc::UnboundedSender<Url>>> = 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) {
+		if let Some(tx) = &*MOVE_TX.lock() {
+			tx.send(BodyMoveItem { from, to }).ok();
+		}
+	}
+
+	#[inline]
+	pub fn push_trash(target: Url) {
+		if let Some(tx) = &*TRASH_TX.lock() {
+			tx.send(target).ok();
+		}
+	}
+
+	#[inline]
+	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 (trash_tx, trash_rx) = mpsc::unbounded_channel();
+		let (delete_tx, delete_rx) = mpsc::unbounded_channel();
+
+		CT.with(Default::default);
+		MOVE_TX.lock().replace(move_tx);
+		TRASH_TX.lock().replace(trash_tx);
+		DELETE_TX.lock().replace(delete_tx);
+
+		tokio::spawn(async move {
+			let move_rx =
+				UnboundedReceiverStream::new(move_rx).chunks_timeout(1000, Duration::from_millis(500));
+			let trash_rx =
+				UnboundedReceiverStream::new(trash_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!(trash_rx);
+			pin!(delete_rx);
+
+			loop {
+				select! {
+					Some(items) = move_rx.next() => Pubsub::pub_from_move(items),
+					Some(urls) = trash_rx.next() => Pubsub::pub_from_trash(urls),
+					Some(urls) = delete_rx.next() => Pubsub::pub_from_delete(urls),
+					else => {
+						CT.cancel();
+						break;
+					},
+				}
+			}
+		});
+	}
+
+	pub(super) async fn shutdown() {
+		drop(MOVE_TX.lock().take());
+		drop(TRASH_TX.lock().take());
+		drop(DELETE_TX.lock().take());
+		CT.cancelled().await;
+	}
+}

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

@@ -9,6 +9,7 @@ impl App {
 	pub(crate) fn quit(&mut self, opt: EventQuit) -> ! {
 		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 {

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/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);
 				}
@@ -172,7 +174,8 @@ impl Scheduler {
 		let file = self.file.clone();
 		_ = self.micro.try_send(
 			async move {
-				file.trash(FileOpTrash { id, target, length: 0 }).await.ok();
+				file.trash(FileOpTrash { id, target: target.clone(), length: 0 }).await.ok();
+				Pump::push_trash(target);
 			}
 			.boxed(),
 			LOW,