From 38813413ec5826da0fcac5396728595686832c67 Mon Sep 17 00:00:00 2001 From: Mika Vilpas Date: Wed, 10 Apr 2024 20:09:29 +0300 Subject: [PATCH] feat: add new `move`, `trash`, and `delete` event kinds to DDS (#880) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: 三咲雅 · Misaki Masa --- .github/ISSUE_TEMPLATE/bug.yml | 2 +- Cargo.lock | 2 + yazi-dds/Cargo.toml | 14 +++--- yazi-dds/src/body/body.rs | 14 +++++- yazi-dds/src/body/delete.rs | 36 ++++++++++++++ yazi-dds/src/body/mod.rs | 6 +++ yazi-dds/src/body/move_.rs | 52 ++++++++++++++++++++ yazi-dds/src/body/trash.rs | 36 ++++++++++++++ yazi-dds/src/lib.rs | 7 ++- yazi-dds/src/payload.rs | 3 ++ yazi-dds/src/pubsub.rs | 38 ++++++++++++++- yazi-dds/src/pump.rs | 82 ++++++++++++++++++++++++++++++++ yazi-fm/src/app/commands/quit.rs | 1 + yazi-fm/src/main.rs | 2 +- yazi-scheduler/src/scheduler.rs | 11 +++-- 15 files changed, 291 insertions(+), 15 deletions(-) create mode 100644 yazi-dds/src/body/delete.rs create mode 100644 yazi-dds/src/body/move_.rs create mode 100644 yazi-dds/src/body/trash.rs create mode 100644 yazi-dds/src/pump.rs diff --git a/.github/ISSUE_TEMPLATE/bug.yml b/.github/ISSUE_TEMPLATE/bug.yml index 19e0d655..5d47b605 100644 --- a/.github/ISSUE_TEMPLATE/bug.yml +++ b/.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 diff --git a/Cargo.lock b/Cargo.lock index 06d81cef..0b8cc07a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2773,6 +2773,8 @@ dependencies = [ "serde", "serde_json", "tokio", + "tokio-stream", + "tokio-util", "tracing", "uzers", "yazi-boot", diff --git a/yazi-dds/Cargo.toml b/yazi-dds/Cargo.toml index 5568125d..51307cc9 100644 --- a/yazi-dds/Cargo.toml +++ b/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" ] } diff --git a/yazi-dds/src/body/body.rs b/yazi-dds/src/body/body.rs index 693d987b..c6b9f76e 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, 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), } } diff --git a/yazi-dds/src/body/delete.rs b/yazi-dds/src/body/delete.rs new file mode 100644 index 00000000..5397cacb --- /dev/null +++ b/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>, +} + +impl<'a> BodyDelete<'a> { + #[inline] + pub fn borrowed(urls: &'a Vec) -> Body<'a> { Self { urls: Cow::Borrowed(urls) }.into() } +} + +impl BodyDelete<'static> { + #[inline] + pub fn owned(urls: Vec) -> Body<'static> { Self { urls: Cow::Owned(urls) }.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.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) + } +} diff --git a/yazi-dds/src/body/mod.rs b/yazi-dds/src/body/mod.rs index 4570acd4..1c60c6ce 100644 --- a/yazi-dds/src/body/mod.rs +++ b/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::*; diff --git a/yazi-dds/src/body/move_.rs b/yazi-dds/src/body/move_.rs new file mode 100644 index 00000000..a38cc776 --- /dev/null +++ b/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>, +} + +impl<'a> BodyMove<'a> { + #[inline] + pub fn borrowed(items: &'a Vec) -> Body<'a> { + Self { items: Cow::Borrowed(items) }.into() + } +} + +impl BodyMove<'static> { + #[inline] + pub fn owned(items: Vec) -> Body<'static> { + Self { items: Cow::Owned(items) }.into() + } +} + +impl<'a> From> 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 { 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)?), + ("to", lua.create_any_userdata(self.to)?), + ])? + .into_lua(lua) + } +} diff --git a/yazi-dds/src/body/trash.rs b/yazi-dds/src/body/trash.rs new file mode 100644 index 00000000..88a2242b --- /dev/null +++ b/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>, +} + +impl<'a> BodyTrash<'a> { + #[inline] + pub fn borrowed(urls: &'a Vec) -> Body<'a> { Self { urls: Cow::Borrowed(urls) }.into() } +} + +impl BodyTrash<'static> { + #[inline] + pub fn owned(urls: Vec) -> Body<'static> { Self { urls: Cow::Owned(urls) }.into() } +} + +impl<'a> From> 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> { + 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) + } +} diff --git a/yazi-dds/src/lib.rs b/yazi-dds/src/lib.rs index a148f92b..14f7f4e9 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 async fn shutdown() { Pump::shutdown().await; } diff --git a/yazi-dds/src/payload.rs b/yazi-dds/src/payload.rs index e65b3c8f..3a6eee9a 100644 --- a/yazi-dds/src/payload.rs +++ b/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), }; diff --git a/yazi-dds/src/pubsub.rs b/yazi-dds/src/pubsub.rs index 11d47c03..67291052 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, 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>>>> = RoCell::new(); @@ -142,4 +142,40 @@ impl Pubsub { BodyYank::borrowed(cut, urls).with_receiver(*ID).flush(); } } + + pub(super) fn pub_from_move(items: Vec) { + 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) { + 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) { + 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)); + } + } } diff --git a/yazi-dds/src/pump.rs b/yazi-dds/src/pump.rs new file mode 100644 index 00000000..39504768 --- /dev/null +++ b/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 = RoCell::new(); +static MOVE_TX: Mutex>> = Mutex::new(None); +static TRASH_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) { + 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; + } +} diff --git a/yazi-fm/src/app/commands/quit.rs b/yazi-fm/src/app/commands/quit.rs index 655e1dd4..f7b7ec3f 100644 --- a/yazi-fm/src/app/commands/quit.rs +++ b/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 { 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/scheduler.rs b/yazi-scheduler/src/scheduler.rs index 7d134efa..8b66556f 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); } @@ -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,