From 224495fd8b9096ca093a206399908f5a11922674 Mon Sep 17 00:00:00 2001 From: sxyazi Date: Mon, 15 Apr 2024 07:27:26 +0800 Subject: [PATCH] feat: add a new `--send` option to allow standalone client processes to communicate with DDS from the command line --- yazi-dds/src/client.rs | 58 +++++------------------------- yazi-dds/src/lib.rs | 2 ++ yazi-dds/src/server.rs | 29 ++------------- yazi-dds/src/stream.rs | 58 ++++++++++++++++++++++++++++++ yazi-plugin/preset/plugins/dds.lua | 7 ++++ yazi-plugin/preset/setup.lua | 2 ++ yazi-plugin/src/isolate/entry.rs | 2 +- yazi-plugin/src/isolate/peek.rs | 2 +- yazi-plugin/src/isolate/preload.rs | 2 +- yazi-plugin/src/loader/loader.rs | 21 +++++++---- 10 files changed, 97 insertions(+), 86 deletions(-) create mode 100644 yazi-dds/src/stream.rs create mode 100644 yazi-plugin/preset/plugins/dds.lua diff --git a/yazi-dds/src/client.rs b/yazi-dds/src/client.rs index e2e4b7e7..a289977d 100644 --- a/yazi-dds/src/client.rs +++ b/yazi-dds/src/client.rs @@ -2,20 +2,15 @@ use std::{collections::{HashMap, HashSet}, mem, str::FromStr}; use parking_lot::RwLock; use serde::{Deserialize, Serialize}; -use tokio::{io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines, ReadHalf, WriteHalf}, select, sync::mpsc, task::JoinHandle, time}; +use tokio::{io::AsyncWriteExt, select, sync::mpsc, task::JoinHandle, time}; use yazi_shared::RoCell; -use crate::{body::Body, Payload, Pubsub, Server}; +use crate::{body::Body, ClientReader, ClientWriter, Payload, Pubsub, Server}; pub(super) static ID: RoCell = RoCell::new(); pub(super) static PEERS: RoCell>> = RoCell::new(); pub(super) static QUEUE: RoCell> = RoCell::new(); -#[cfg(not(unix))] -use tokio::net::TcpStream; -#[cfg(unix)] -use tokio::net::UnixStream; - #[derive(Debug)] pub struct Client { pub(super) id: u64, @@ -69,16 +64,14 @@ impl Client { #[inline] pub(super) fn able(&self, ability: &str) -> bool { self.abilities.contains(ability) } - #[cfg(unix)] - async fn connect( - server: &mut Option>, - ) -> (Lines>>, WriteHalf) { + async fn connect(server: &mut Option>) -> (ClientReader, ClientWriter) { + use crate::Stream; + let mut first = true; loop { - if let Ok(stream) = UnixStream::connect(Server::socket_file()).await { + if let Ok(conn) = Stream::connect().await { Pubsub::pub_from_hi(); - let (reader, writer) = tokio::io::split(stream); - return (BufReader::new(reader).lines(), writer); + return conn; } server.take().map(|h| h.abort()); @@ -94,42 +87,7 @@ impl Client { } } - #[cfg(not(unix))] - async fn connect( - server: &mut Option>, - ) -> (Lines>>, WriteHalf) { - let mut first = true; - loop { - if let Ok(stream) = TcpStream::connect("127.0.0.1:33581").await { - Pubsub::pub_from_hi(); - let (reader, writer) = tokio::io::split(stream); - return (BufReader::new(reader).lines(), writer); - } - - server.take().map(|h| h.abort()); - *server = Server::make().await.ok(); - if mem::replace(&mut first, false) && server.is_some() { - continue; - } - - time::sleep(time::Duration::from_secs(1)).await; - } - } - - #[cfg(unix)] - async fn reconnect( - server: &mut Option>, - ) -> (Lines>>, WriteHalf) { - PEERS.write().clear(); - - time::sleep(time::Duration::from_millis(500)).await; - Self::connect(server).await - } - - #[cfg(not(unix))] - async fn reconnect( - server: &mut Option>, - ) -> (Lines>>, WriteHalf) { + async fn reconnect(server: &mut Option>) -> (ClientReader, ClientWriter) { PEERS.write().clear(); time::sleep(time::Duration::from_millis(500)).await; diff --git a/yazi-dds/src/lib.rs b/yazi-dds/src/lib.rs index e923dc36..ee7ff6b6 100644 --- a/yazi-dds/src/lib.rs +++ b/yazi-dds/src/lib.rs @@ -7,6 +7,7 @@ mod pump; mod sendable; mod server; mod state; +mod stream; pub use client::*; pub use payload::*; @@ -15,6 +16,7 @@ pub use pump::*; pub use sendable::*; use server::*; pub use state::*; +pub use stream::*; pub fn serve() { let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); diff --git a/yazi-dds/src/server.rs b/yazi-dds/src/server.rs index dfd3d102..91cf6596 100644 --- a/yazi-dds/src/server.rs +++ b/yazi-dds/src/server.rs @@ -5,7 +5,7 @@ use parking_lot::RwLock; use tokio::{io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, select, sync::mpsc, task::JoinHandle, time}; use yazi_shared::RoCell; -use crate::{body::{Body, BodyHey}, Client, Payload, Peer, STATE}; +use crate::{body::{Body, BodyHey}, Client, Payload, Peer, Stream, STATE}; pub(super) static CLIENTS: RoCell>> = RoCell::new(); @@ -14,7 +14,7 @@ pub(super) struct Server; impl Server { pub(super) async fn make() -> Result> { CLIENTS.write().clear(); - let listener = Self::bind().await?; + let listener = Stream::bind().await?; Ok(tokio::spawn(async move { while let Ok((stream, _)) = listener.accept().await { @@ -78,31 +78,6 @@ impl Server { })) } - #[cfg(unix)] - #[inline] - pub(super) fn socket_file() -> std::path::PathBuf { - use uzers::Users; - use yazi_boot::USERS_CACHE; - use yazi_shared::Xdg; - - Xdg::cache_dir().join(format!(".dds-{}.sock", USERS_CACHE.get_current_uid())) - } - - #[cfg(unix)] - #[inline] - async fn bind() -> Result { - let p = Self::socket_file(); - - tokio::fs::remove_file(&p).await.ok(); - Ok(tokio::net::UnixListener::bind(p)?) - } - - #[cfg(not(unix))] - #[inline] - async fn bind() -> Result { - Ok(tokio::net::TcpListener::bind("127.0.0.1:33581").await?) - } - fn handle_hi(s: String, id: &mut Option, tx: mpsc::UnboundedSender) { let Ok(payload) = Payload::from_str(&s) else { return }; let Body::Hi(hi) = payload.body else { return }; diff --git a/yazi-dds/src/stream.rs b/yazi-dds/src/stream.rs new file mode 100644 index 00000000..8967a6b0 --- /dev/null +++ b/yazi-dds/src/stream.rs @@ -0,0 +1,58 @@ +use tokio::io::{BufReader, Lines, ReadHalf, WriteHalf}; + +pub struct Stream; + +use tokio::io::AsyncBufReadExt; + +#[cfg(unix)] +pub type ClientReader = Lines>>; +#[cfg(not(unix))] +pub type ClientReader = Lines>>; + +#[cfg(unix)] +pub type ClientWriter = WriteHalf; +#[cfg(not(unix))] +pub type ClientWriter = WriteHalf; + +#[cfg(unix)] +pub type ServerListener = tokio::net::UnixListener; +#[cfg(not(unix))] +pub type ServerListener = tokio::net::TcpListener; + +impl Stream { + #[cfg(unix)] + pub async fn connect() -> std::io::Result<(ClientReader, ClientWriter)> { + let stream = tokio::net::UnixStream::connect(Self::socket_file()).await?; + let (reader, writer) = tokio::io::split(stream); + Ok((BufReader::new(reader).lines(), writer)) + } + + #[cfg(not(unix))] + pub async fn connect() -> std::io::Result<(ClientReader, ClientWriter)> { + let stream = tokio::net::TcpStream::connect("127.0.0.1:33581").await?; + let (reader, writer) = tokio::io::split(stream); + Ok((BufReader::new(reader).lines(), writer)) + } + + #[cfg(unix)] + pub async fn bind() -> std::io::Result { + let p = Self::socket_file(); + + tokio::fs::remove_file(&p).await.ok(); + tokio::net::UnixListener::bind(p) + } + + #[cfg(not(unix))] + pub async fn bind() -> std::io::Result { + tokio::net::TcpListener::bind("127.0.0.1:33581").await + } + + #[cfg(unix)] + pub(super) fn socket_file() -> std::path::PathBuf { + use uzers::Users; + use yazi_boot::USERS_CACHE; + use yazi_shared::Xdg; + + Xdg::cache_dir().join(format!(".dds-{}.sock", USERS_CACHE.get_current_uid())) + } +} diff --git a/yazi-plugin/preset/plugins/dds.lua b/yazi-plugin/preset/plugins/dds.lua new file mode 100644 index 00000000..b8a0452f --- /dev/null +++ b/yazi-plugin/preset/plugins/dds.lua @@ -0,0 +1,7 @@ +local M = {} + +function M:setup() + ps.sub_remote("dds-cd", function(url) ya.manager_emit("cd", { url }) end) +end + +return M diff --git a/yazi-plugin/preset/setup.lua b/yazi-plugin/preset/setup.lua index e67aebbe..a6adcbb9 100644 --- a/yazi-plugin/preset/setup.lua +++ b/yazi-plugin/preset/setup.lua @@ -1 +1,3 @@ package.path = BOOT.plugin_dir .. "/?.yazi/init.lua;" .. package.path + +require("dds"):setup() diff --git a/yazi-plugin/src/isolate/entry.rs b/yazi-plugin/src/isolate/entry.rs index 7b776170..0edea618 100644 --- a/yazi-plugin/src/isolate/entry.rs +++ b/yazi-plugin/src/isolate/entry.rs @@ -11,7 +11,7 @@ pub async fn entry(name: String, args: Vec) -> mlua::Result<()> { tokio::task::spawn_blocking(move || { let lua = slim_lua(&name)?; let plugin: Table = if let Some(b) = LOADER.read().get(&name) { - lua.load(b).call(())? + lua.load(b.as_ref()).call(())? } else { return Err("unloaded plugin".into_lua_err()); }; diff --git a/yazi-plugin/src/isolate/peek.rs b/yazi-plugin/src/isolate/peek.rs index aba27316..2ad97463 100644 --- a/yazi-plugin/src/isolate/peek.rs +++ b/yazi-plugin/src/isolate/peek.rs @@ -26,7 +26,7 @@ pub fn peek(cmd: &Cmd, file: yazi_shared::fs::File, skip: usize) -> Cancellation ); let plugin: Table = if let Some(b) = LOADER.read().get(&name) { - lua.load(b).call(())? + lua.load(b.as_ref()).call(())? } else { return Err("unloaded plugin".into_lua_err()); }; diff --git a/yazi-plugin/src/isolate/preload.rs b/yazi-plugin/src/isolate/preload.rs index 19668597..3ce0f76a 100644 --- a/yazi-plugin/src/isolate/preload.rs +++ b/yazi-plugin/src/isolate/preload.rs @@ -16,7 +16,7 @@ pub async fn preload( tokio::task::spawn_blocking(move || { let lua = slim_lua(&name)?; let plugin: Table = if let Some(b) = LOADER.read().get(&name) { - lua.load(b).call(())? + lua.load(b.as_ref()).call(())? } else { return Err("unloaded plugin".into_lua_err()); }; diff --git a/yazi-plugin/src/loader/loader.rs b/yazi-plugin/src/loader/loader.rs index 4893a8eb..bb442760 100644 --- a/yazi-plugin/src/loader/loader.rs +++ b/yazi-plugin/src/loader/loader.rs @@ -13,7 +13,7 @@ pub static LOADER: RoCell = RoCell::new(); #[derive(Default)] pub struct Loader { - cache: RwLock>>, + cache: RwLock>>, } impl Loader { @@ -22,10 +22,20 @@ impl Loader { return Ok(()); } + let b = match name { + "dds" => Some(&include_bytes!("../../preset/plugins/dds.lua")[..]), + "noop" => Some(&include_bytes!("../../preset/plugins/noop.lua")[..]), + _ => None, + }; + if let Some(b) = b { + self.cache.write().insert(name.to_owned(), Cow::Borrowed(b)); + return Ok(()); + } + let path = BOOT.plugin_dir.join(format!("{name}.yazi/init.lua")); let b = fs::read(path).await.map(|v| v.into()).or_else(|_| { Ok(Cow::from(match name { - "archive" => include_bytes!("../../preset/plugins/archive.lua") as &[_], + "archive" => &include_bytes!("../../preset/plugins/archive.lua")[..], "code" => include_bytes!("../../preset/plugins/code.lua"), "file" => include_bytes!("../../preset/plugins/file.lua"), "folder" => include_bytes!("../../preset/plugins/folder.lua"), @@ -33,7 +43,6 @@ impl Loader { "image" => include_bytes!("../../preset/plugins/image.lua"), "json" => include_bytes!("../../preset/plugins/json.lua"), "mime" => include_bytes!("../../preset/plugins/mime.lua"), - "noop" => include_bytes!("../../preset/plugins/noop.lua"), "pdf" => include_bytes!("../../preset/plugins/pdf.lua"), "video" => include_bytes!("../../preset/plugins/video.lua"), "zoxide" => include_bytes!("../../preset/plugins/zoxide.lua"), @@ -41,7 +50,7 @@ impl Loader { })) })?; - self.cache.write().insert(name.to_owned(), b.into_owned()); + self.cache.write().insert(name.to_owned(), b); Ok(()) } @@ -53,7 +62,7 @@ impl Loader { } let t: Table = match self.read().get(name) { - Some(b) => LUA.load(b).call(())?, + Some(b) => LUA.load(b.as_ref()).call(())?, None => Err(format!("plugin `{name}` not found").into_lua_err())?, }; @@ -64,7 +73,7 @@ impl Loader { } impl Deref for Loader { - type Target = RwLock>>; + type Target = RwLock>>; #[inline] fn deref(&self) -> &Self::Target { &self.cache }