diff --git a/Cargo.lock b/Cargo.lock index 06d81cef..676ca9c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2773,6 +2773,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "tokio-stream", "tracing", "uzers", "yazi-boot", diff --git a/yazi-dds/Cargo.toml b/yazi-dds/Cargo.toml index 5568125d..49071c13 100644 --- a/yazi-dds/Cargo.toml +++ b/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" ] } diff --git a/yazi-dds/src/body/body.rs b/yazi-dds/src/body/body.rs index 4ee2024c..21549fd2 100644 --- a/yazi-dds/src/body/body.rs +++ b/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), } } diff --git a/yazi-dds/src/body/delete.rs b/yazi-dds/src/body/delete.rs new file mode 100644 index 00000000..c5dd3f6e --- /dev/null +++ b/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>, +} + +impl<'a> BodyDelete<'a> { + #[inline] + pub fn borrowed(targets: &'a Vec) -> Body<'a> { + Self { targets: Cow::Borrowed(targets) }.into() + } +} + +impl BodyDelete<'static> { + #[inline] + pub fn owned(targets: Vec) -> Body<'static> { Self { targets: Cow::Owned(targets) }.into() } +} + +impl<'a> From> 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> { + 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) + } +} diff --git a/yazi-dds/src/body/mod.rs b/yazi-dds/src/body/mod.rs index c7c50a9f..ade26d21 100644 --- a/yazi-dds/src/body/mod.rs +++ b/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::*; diff --git a/yazi-dds/src/body/move_.rs b/yazi-dds/src/body/move_.rs index 4a2f42a4..a38cc776 100644 --- a/yazi-dds/src/body/move_.rs +++ b/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>, } 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) -> 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) -> Body<'static> { + Self { items: Cow::Owned(items) }.into() } } @@ -31,11 +30,22 @@ impl<'a> From> for Body<'a> { } impl IntoLua<'_> for BodyMove<'static> { + fn into_lua(self, lua: &Lua) -> mlua::Result { 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 { 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) } diff --git a/yazi-dds/src/lib.rs b/yazi-dds/src/lib.rs index a148f92b..b9479569 100644 --- a/yazi-dds/src/lib.rs +++ b/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(); } diff --git a/yazi-dds/src/payload.rs b/yazi-dds/src/payload.rs index d7f50327..7deb80d5 100644 --- a/yazi-dds/src/payload.rs +++ b/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), }; diff --git a/yazi-dds/src/pubsub.rs b/yazi-dds/src/pubsub.rs index 89759dbc..8a6912bd 100644 --- a/yazi-dds/src/pubsub.rs +++ 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>>>> = 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) { 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) { + 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)); } } } diff --git a/yazi-dds/src/pump.rs b/yazi-dds/src/pump.rs new file mode 100644 index 00000000..0f339ad6 --- /dev/null +++ b/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> = RoCell::new(); +static DELETE_TX: RoCell> = 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(); + } +} diff --git a/yazi-fm/src/app/commands/quit.rs b/yazi-fm/src/app/commands/quit.rs index 655e1dd4..61f5440d 100644 --- a/yazi-fm/src/app/commands/quit.rs +++ b/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(); diff --git a/yazi-fm/src/main.rs b/yazi-fm/src/main.rs index d8d92685..0cc34d21 100644 --- a/yazi-fm/src/main.rs +++ b/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(); diff --git a/yazi-scheduler/src/file/file.rs b/yazi-scheduler/src/file/file.rs index 534cbdb1..4a9fb53c 100644 --- a/yazi-scheduler/src/file/file.rs +++ b/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; } diff --git a/yazi-scheduler/src/scheduler.rs b/yazi-scheduler/src/scheduler.rs index 7d134efa..2eacb560 100644 --- a/yazi-scheduler/src/scheduler.rs +++ b/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); }