feat: add a new --send option to allow standalone client processes to communicate with DDS from the command line

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<u64> = RoCell::new();
 pub(super) static PEERS: RoCell<RwLock<HashMap<u64, Peer>>> = RoCell::new();
 pub(super) static QUEUE: RoCell<mpsc::UnboundedSender<String>> = 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<JoinHandle<()>>) -> (ClientReader, ClientWriter) {
-	async fn connect(
+		use crate::Stream;
-		server: &mut Option<JoinHandle<()>>,
+
-	) -> (Lines<BufReader<ReadHalf<UnixStream>>>, WriteHalf<UnixStream>) {
 		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 conn;
-				return (BufReader::new(reader).lines(), writer);
 			}
 
 			server.take().map(|h| h.abort());
@@ -94,42 +87,7 @@ impl Client {
 		}
 	}
 
-	#[cfg(not(unix))]
+	async fn reconnect(server: &mut Option<JoinHandle<()>>) -> (ClientReader, ClientWriter) {
-	async fn connect(
-		server: &mut Option<JoinHandle<()>>,
-	) -> (Lines<BufReader<ReadHalf<TcpStream>>>, WriteHalf<TcpStream>) {
-		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<JoinHandle<()>>,
-	) -> (Lines<BufReader<ReadHalf<UnixStream>>>, WriteHalf<UnixStream>) {
-		PEERS.write().clear();
-
-		time::sleep(time::Duration::from_millis(500)).await;
-		Self::connect(server).await
-	}
-
-	#[cfg(not(unix))]
-	async fn reconnect(
-		server: &mut Option<JoinHandle<()>>,
-	) -> (Lines<BufReader<ReadHalf<TcpStream>>>, WriteHalf<TcpStream>) {
 		PEERS.write().clear();
 
 		time::sleep(time::Duration::from_millis(500)).await;

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();

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<RwLock<HashMap<u64, Client>>> = RoCell::new();
 
@@ -14,7 +14,7 @@ pub(super) struct Server;
 impl Server {
 	pub(super) async fn make() -> Result<JoinHandle<()>> {
 		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<tokio::net::UnixListener> {
-		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<tokio::net::TcpListener> {
-		Ok(tokio::net::TcpListener::bind("127.0.0.1:33581").await?)
-	}
-
 	fn handle_hi(s: String, id: &mut Option<u64>, tx: mpsc::UnboundedSender<String>) {
 		let Ok(payload) = Payload::from_str(&s) else { return };
 		let Body::Hi(hi) = payload.body else { return };

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<BufReader<ReadHalf<tokio::net::UnixStream>>>;
+#[cfg(not(unix))]
+pub type ClientReader = Lines<BufReader<ReadHalf<tokio::net::TcpStream>>>;
+
+#[cfg(unix)]
+pub type ClientWriter = WriteHalf<tokio::net::UnixStream>;
+#[cfg(not(unix))]
+pub type ClientWriter = WriteHalf<tokio::net::TcpStream>;
+
+#[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<ServerListener> {
+		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<ServerListener> {
+		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()))
+	}
+}

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

yazi-plugin/preset/setup.lua

@@ -1 +1,3 @@
 package.path = BOOT.plugin_dir .. "/?.yazi/init.lua;" .. package.path
+
+require("dds"):setup()

yazi-plugin/src/isolate/entry.rs

@@ -11,7 +11,7 @@ pub async fn entry(name: String, args: Vec<ValueSendable>) -> 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());
 		};

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());
 			};

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());
 		};

yazi-plugin/src/loader/loader.rs

@@ -13,7 +13,7 @@ pub static LOADER: RoCell<Loader> = RoCell::new();
 
 #[derive(Default)]
 pub struct Loader {
-	cache: RwLock<HashMap<String, Vec<u8>>>,
+	cache: RwLock<HashMap<String, Cow<'static, [u8]>>>,
 }
 
 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<HashMap<String, Vec<u8>>>;
+	type Target = RwLock<HashMap<String, Cow<'static, [u8]>>>;
 
 	#[inline]
 	fn deref(&self) -> &Self::Target { &self.cache }