feat: add a new ya send command to allow standalone client processes to communicate with DDS from the command line (#913)

.github/workflows/release.yml

@@ -64,6 +64,7 @@ jobs:
           TARGET_NAME: yazi-${{ matrix.target }}
         run: |
           New-Item -ItemType Directory -Path ${env:TARGET_NAME}
+          Copy-Item -Path "target\${{ matrix.target }}\release\ya.exe" -Destination ${env:TARGET_NAME}
           Copy-Item -Path "target\${{ matrix.target }}\release\yazi.exe" -Destination ${env:TARGET_NAME}
           Copy-Item -Path "yazi-boot\completions" -Destination ${env:TARGET_NAME} -Recurse
           Copy-Item -Path "README.md", "LICENSE" -Destination ${env:TARGET_NAME}

Cargo.lock

@@ -2712,13 +2712,22 @@ dependencies = [
  "clap_complete_fig",
  "clap_complete_nushell",
  "serde",
- "uzers",
  "vergen",
  "yazi-adaptor",
  "yazi-config",
  "yazi-shared",
 ]
 
+[[package]]
+name = "yazi-cli"
+version = "0.2.4"
+dependencies = [
+ "anyhow",
+ "clap",
+ "tokio",
+ "yazi-dds",
+]
+
 [[package]]
 name = "yazi-config"
 version = "0.2.4"

scripts/build.sh

@@ -14,6 +14,7 @@ cargo build --release --locked --target "$1"
 
 # Create the artifact
 mkdir "$ARTIFACT_NAME"
+cp "target/$1/release/ya" "$ARTIFACT_NAME"
 cp "target/$1/release/yazi" "$ARTIFACT_NAME"
 cp -r yazi-boot/completions "$ARTIFACT_NAME"
 cp README.md LICENSE "$ARTIFACT_NAME"

yazi-boot/Cargo.toml

@@ -23,6 +23,3 @@ clap_complete         = "4.5.2"
 clap_complete_nushell = "4.5.1"
 clap_complete_fig     = "4.5.0"
 vergen                = { version = "8.3.1", features = [ "build", "git", "gitcl" ] }
-
-[target."cfg(unix)".dependencies]
-uzers = "0.11.3"

yazi-boot/src/lib.rs

@@ -9,13 +9,7 @@ pub use boot::*;
 pub static ARGS: RoCell<Args> = RoCell::new();
 pub static BOOT: RoCell<Boot> = RoCell::new();
 
-#[cfg(unix)]
-pub static USERS_CACHE: yazi_shared::RoCell<uzers::UsersCache> = yazi_shared::RoCell::new();
-
 pub fn init() {
 	ARGS.with(Default::default);
 	BOOT.with(Default::default);
-
-	#[cfg(unix)]
-	USERS_CACHE.with(Default::default);
 }

yazi-cli/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name        = "yazi-cli"
+version     = "0.2.4"
+edition     = "2021"
+license     = "MIT"
+authors     = [ "sxyazi <sxyazi@gmail.com>" ]
+description = "Yazi command-line interface"
+homepage    = "https://yazi-rs.github.io"
+repository  = "https://github.com/sxyazi/yazi"
+
+[dependencies]
+yazi-dds  = { path = "../yazi-dds", version = "0.2.4" }
+
+# External dependencies
+anyhow = "1.0.82"
+clap   = { version = "4.5.4", features = [ "derive" ] }
+tokio  = { version = "1.37.0", features = [ "full" ] }
+
+[[bin]]
+name = "ya"
+path = "src/main.rs"

yazi-cli/src/args.rs

@@ -0,0 +1,20 @@
+use clap::{command, Parser, Subcommand};
+
+#[derive(Parser)]
+#[command(name = "ya",version, about, long_about = None)]
+#[command(propagate_version = true)]
+pub(super) struct Args {
+	#[command(subcommand)]
+	pub(super) command: Command,
+}
+
+#[derive(Subcommand)]
+pub(super) enum Command {
+	/// Send a message to remote instances.
+	Send(CommandSend),
+}
+
+#[derive(clap::Args)]
+pub(super) struct CommandSend {
+	pub(super) message: String,
+}

yazi-cli/src/main.rs

@@ -0,0 +1,21 @@
+mod args;
+
+use args::*;
+use clap::Parser;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+	let args = Args::parse();
+
+	match &args.command {
+		Command::Send(cmd) => {
+			yazi_dds::init();
+			if let Err(e) = yazi_dds::Client::shot(&cmd.message).await {
+				eprintln!("Cannot send message: {e}");
+				std::process::exit(1);
+			}
+		}
+	}
+
+	Ok(())
+}

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, BodyDelete, BodyHey, BodyHi, BodyHover, BodyMove, BodyRename, BodyTrash, BodyYank};
+use super::{BodyBulk, BodyBye, BodyCd, BodyCustom, BodyDelete, BodyHey, BodyHi, BodyHover, BodyMove, BodyRename, BodyTrash, BodyYank};
 use crate::Payload;
 
 #[derive(Debug, Serialize)]
@@ -10,6 +10,7 @@ use crate::Payload;
 pub enum Body<'a> {
 	Hi(BodyHi<'a>),
 	Hey(BodyHey),
+	Bye(BodyBye),
 	Cd(BodyCd<'a>),
 	Hover(BodyHover<'a>),
 	Rename(BodyRename<'a>),
@@ -21,28 +22,28 @@ pub enum Body<'a> {
 	Custom(BodyCustom),
 }
 
-impl<'a> Body<'a> {
+impl Body<'static> {
 	pub fn from_str(kind: &str, body: &str) -> Result<Self> {
 		Ok(match kind {
-			"hi" => Body::Hi(serde_json::from_str(body)?),
-			"hey" => Body::Hey(serde_json::from_str(body)?),
-			"cd" => Body::Cd(serde_json::from_str(body)?),
-			"hover" => Body::Hover(serde_json::from_str(body)?),
-			"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)?),
+			"hi" => Self::Hi(serde_json::from_str(body)?),
+			"hey" => Self::Hey(serde_json::from_str(body)?),
+			"bye" => Self::Bye(serde_json::from_str(body)?),
+			"cd" => Self::Cd(serde_json::from_str(body)?),
+			"hover" => Self::Hover(serde_json::from_str(body)?),
+			"rename" => Self::Rename(serde_json::from_str(body)?),
+			"bulk" => Self::Bulk(serde_json::from_str(body)?),
+			"yank" => Self::Yank(serde_json::from_str(body)?),
+			"move" => Self::Move(serde_json::from_str(body)?),
+			"trash" => Self::Trash(serde_json::from_str(body)?),
+			"delete" => Self::Delete(serde_json::from_str(body)?),
 			_ => BodyCustom::from_str(kind, body)?,
 		})
 	}
 
 	pub fn from_lua(kind: &str, value: Value) -> Result<Self> {
 		Ok(match kind {
-			"hi" | "hey" | "cd" | "hover" | "rename" | "bulk" | "yank" | "move" | "trash" | "delete" => {
-				Err("Cannot construct system event from Lua").into_lua_err()?
-			}
+			"hi" | "hey" | "bye" | "cd" | "hover" | "rename" | "bulk" | "yank" | "move" | "trash"
+			| "delete" => Err("Cannot construct system event").into_lua_err()?,
 			_ if !kind.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-') => {
 				Err("Kind must be alphanumeric with dashes").into_lua_err()?
 			}
@@ -50,23 +51,6 @@ impl<'a> Body<'a> {
 		})
 	}
 
-	#[inline]
-	pub fn kind(&self) -> &str {
-		match self {
-			Self::Hi(_) => "hi",
-			Self::Hey(_) => "hey",
-			Self::Cd(_) => "cd",
-			Self::Hover(_) => "hover",
-			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(),
-		}
-	}
-
 	pub fn tab(kind: &str, body: &str) -> usize {
 		match kind {
 			"cd" | "hover" | "bulk" | "rename" => {}
@@ -74,19 +58,42 @@ impl<'a> Body<'a> {
 		}
 
 		match Self::from_str(kind, body) {
-			Ok(Body::Cd(b)) => b.tab,
-			Ok(Body::Hover(b)) => b.tab,
-			Ok(Body::Bulk(b)) => b.tab,
-			Ok(Body::Rename(b)) => b.tab,
+			Ok(Self::Cd(b)) => b.tab,
+			Ok(Self::Hover(b)) => b.tab,
+			Ok(Self::Bulk(b)) => b.tab,
+			Ok(Self::Rename(b)) => b.tab,
 			_ => 0,
 		}
 	}
+}
+
+impl<'a> Body<'a> {
+	#[inline]
+	pub fn kind(&self) -> &str {
+		match self {
+			Self::Hi(_) => "hi",
+			Self::Hey(_) => "hey",
+			Self::Bye(_) => "bye",
+			Self::Cd(_) => "cd",
+			Self::Hover(_) => "hover",
+			Self::Rename(_) => "rename",
+			Self::Bulk(_) => "bulk",
+			Self::Yank(_) => "yank",
+			Self::Move(_) => "move",
+			Self::Trash(_) => "trash",
+			Self::Delete(_) => "delete",
+			Self::Custom(b) => b.kind.as_str(),
+		}
+	}
 
 	#[inline]
 	pub fn with_receiver(self, receiver: u64) -> Payload<'a> {
 		Payload::new(self).with_receiver(receiver)
 	}
 
+	#[inline]
+	pub fn with_sender(self, sender: u64) -> Payload<'a> { Payload::new(self).with_sender(sender) }
+
 	#[inline]
 	pub fn with_severity(self, severity: u16) -> Payload<'a> {
 		Payload::new(self).with_severity(severity)
@@ -96,17 +103,18 @@ impl<'a> Body<'a> {
 impl IntoLua<'_> for Body<'static> {
 	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> {
 		match self {
-			Body::Hi(b) => b.into_lua(lua),
-			Body::Hey(b) => b.into_lua(lua),
-			Body::Cd(b) => b.into_lua(lua),
-			Body::Hover(b) => b.into_lua(lua),
-			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),
+			Self::Hi(b) => b.into_lua(lua),
+			Self::Hey(b) => b.into_lua(lua),
+			Self::Bye(b) => b.into_lua(lua),
+			Self::Cd(b) => b.into_lua(lua),
+			Self::Hover(b) => b.into_lua(lua),
+			Self::Rename(b) => b.into_lua(lua),
+			Self::Bulk(b) => b.into_lua(lua),
+			Self::Yank(b) => b.into_lua(lua),
+			Self::Move(b) => b.into_lua(lua),
+			Self::Trash(b) => b.into_lua(lua),
+			Self::Delete(b) => b.into_lua(lua),
+			Self::Custom(b) => b.into_lua(lua),
 		}
 	}
 }

yazi-dds/src/body/bye.rs

@@ -0,0 +1,22 @@
+use mlua::{ExternalResult, IntoLua, Lua, Value};
+use serde::{Deserialize, Serialize};
+
+use super::Body;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BodyBye {}
+
+impl BodyBye {
+	#[inline]
+	pub fn borrowed() -> Body<'static> { Self {}.into() }
+}
+
+impl<'a> From<BodyBye> for Body<'a> {
+	fn from(value: BodyBye) -> Self { Self::Bye(value) }
+}
+
+impl IntoLua<'_> for BodyBye {
+	fn into_lua(self, _: &Lua) -> mlua::Result<Value<'_>> {
+		Err("BodyBye cannot be converted to Lua").into_lua_err()
+	}
+}

yazi-dds/src/body/mod.rs

@@ -2,6 +2,7 @@
 
 mod body;
 mod bulk;
+mod bye;
 mod cd;
 mod custom;
 mod delete;
@@ -15,6 +16,7 @@ mod yank;
 
 pub use body::*;
 pub use bulk::*;
+pub use bye::*;
 pub use cd::*;
 pub use custom::*;
 pub use delete::*;

yazi-dds/src/client.rs

@@ -1,20 +1,18 @@
 use std::{collections::{HashMap, HashSet}, mem, str::FromStr};
 
+use anyhow::{bail, Result};
 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, BodyBye, BodyHi}, ClientReader, ClientWriter, Payload, Pubsub, Server, Stream};
 
 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;
+pub(super) static QUEUE_TX: RoCell<mpsc::UnboundedSender<String>> = RoCell::new();
+pub(super) static QUEUE_RX: RoCell<mpsc::UnboundedReceiver<String>> = RoCell::new();
 
 #[derive(Debug)]
 pub struct Client {
@@ -29,7 +27,8 @@ pub struct Peer {
 }
 
 impl Client {
-	pub(super) fn serve(mut rx: mpsc::UnboundedReceiver<String>) {
+	pub(super) fn serve() {
+		let mut rx = QUEUE_RX.drop();
 		while rx.try_recv().is_ok() {}
 
 		tokio::spawn(async move {
@@ -50,7 +49,9 @@ impl Client {
 							continue;
 						};
 
-						if line.starts_with("hey,") {
+						if line.is_empty() {
+							continue;
+						} else if line.starts_with("hey,") {
 							Self::handle_hey(line);
 						} else {
 							Payload::from_str(&line).map(|p| p.emit()).ok();
@@ -61,24 +62,47 @@ impl Client {
 		});
 	}
 
+	pub async fn shot(s: &str) -> Result<()> {
+		let (kind, receiver, sender, body) = Payload::split(s)?;
+		if receiver != 0 && sender <= u16::MAX as u64 {
+			bail!("Sender must be greater than 65535 if receiver is non-zero");
+		}
+
+		let payload = format!(
+			"{}\n{kind},{receiver},{sender},{}\n{}\n",
+			Payload::new(BodyHi::borrowed(Default::default())),
+			serde_json::to_string(body)?,
+			Payload::new(BodyBye::borrowed())
+		);
+
+		let (mut lines, mut writer) = Stream::connect().await?;
+		writer.write_all(payload.as_bytes()).await?;
+		writer.flush().await?;
+		drop(writer);
+
+		while let Ok(Some(s)) = lines.next_line().await {
+			if matches!(Payload::split(&s), Ok((kind, ..)) if kind == "bye") {
+				break;
+			}
+		}
+
+		Ok(())
+	}
+
 	#[inline]
 	pub(super) fn push<'a>(payload: impl Into<Payload<'a>>) {
-		QUEUE.send(format!("{}\n", payload.into())).ok();
+		QUEUE_TX.send(format!("{}\n", payload.into())).ok();
 	}
 
 	#[inline]
 	pub(super) fn able(&self, ability: &str) -> bool { self.abilities.contains(ability) }
 
-	#[cfg(unix)]
-	async fn connect(
-		server: &mut Option<JoinHandle<()>>,
-	) -> (Lines<BufReader<ReadHalf<UnixStream>>>, WriteHalf<UnixStream>) {
+	async fn connect(server: &mut Option<JoinHandle<()>>) -> (ClientReader, ClientWriter) {
 		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 +118,7 @@ impl Client {
 		}
 	}
 
-	#[cfg(not(unix))]
-	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>) {
+	async fn reconnect(server: &mut Option<JoinHandle<()>>) -> (ClientReader, ClientWriter) {
 		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,14 +16,19 @@ pub use pump::*;
 pub use sendable::*;
 use server::*;
 pub use state::*;
+use stream::*;
 
-pub fn serve() {
+#[cfg(unix)]
+pub static USERS_CACHE: yazi_shared::RoCell<uzers::UsersCache> = yazi_shared::RoCell::new();
+
+pub fn init() {
 	let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
 
 	// Client
 	ID.init(yazi_shared::timestamp_us());
 	PEERS.with(Default::default);
-	QUEUE.init(tx);
+	QUEUE_TX.init(tx);
+	QUEUE_RX.init(rx);
 
 	// Server
 	CLIENTS.with(Default::default);
@@ -32,15 +38,20 @@ pub fn serve() {
 	LOCAL.with(Default::default);
 	REMOTE.with(Default::default);
 
+	#[cfg(unix)]
+	USERS_CACHE.with(Default::default);
+
 	// Env
 	std::env::set_var("YAZI_ID", ID.to_string());
 	std::env::set_var(
 		"YAZI_LEVEL",
 		(std::env::var("YAZI_LEVEL").unwrap_or_default().parse().unwrap_or(0u16) + 1).to_string(),
 	);
+}
 
+pub fn serve() {
 	Pump::serve();
-	Client::serve(rx);
+	Client::serve();
 }
 
 pub async fn shutdown() { Pump::shutdown().await; }

yazi-dds/src/payload.rs

@@ -37,6 +37,11 @@ impl<'a> Payload<'a> {
 		self
 	}
 
+	pub(super) fn with_sender(mut self, sender: u64) -> Self {
+		self.sender = sender;
+		self
+	}
+
 	pub(super) fn with_severity(mut self, severity: u16) -> Self {
 		self.sender = severity as u64;
 		self
@@ -44,16 +49,7 @@ impl<'a> Payload<'a> {
 }
 
 impl Payload<'static> {
-	pub(super) fn emit(self) {
-		self.try_flush();
-		emit!(Call(Cmd::new("accept_payload").with_data(self), Layer::App));
-	}
-}
-
-impl FromStr for Payload<'_> {
-	type Err = anyhow::Error;
-
-	fn from_str(s: &str) -> Result<Self, Self::Err> {
+	pub fn split(s: &str) -> Result<(&str, u64, u64, &str)> {
 		let mut parts = s.splitn(4, ',');
 
 		let kind = parts.next().ok_or_else(|| anyhow!("empty kind"))?;
@@ -66,6 +62,20 @@ impl FromStr for Payload<'_> {
 
 		let body = parts.next().ok_or_else(|| anyhow!("empty body"))?;
 
+		Ok((kind, receiver, sender, body))
+	}
+
+	pub(super) fn emit(self) {
+		self.try_flush();
+		emit!(Call(Cmd::new("accept_payload").with_data(self), Layer::App));
+	}
+}
+
+impl FromStr for Payload<'static> {
+	type Err = anyhow::Error;
+
+	fn from_str(s: &str) -> Result<Self, Self::Err> {
+		let (kind, receiver, sender, body) = Self::split(s)?;
 		Ok(Self { receiver, sender, body: Body::from_str(kind, body)? })
 	}
 }
@@ -79,6 +89,7 @@ impl Display for Payload<'_> {
 		let result = match &self.body {
 			Body::Hi(b) => serde_json::to_string(b),
 			Body::Hey(b) => serde_json::to_string(b),
+			Body::Bye(b) => serde_json::to_string(b),
 			Body::Cd(b) => serde_json::to_string(b),
 			Body::Hover(b) => serde_json::to_string(b),
 			Body::Rename(b) => serde_json::to_string(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, BodyBye, 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 {
@@ -42,8 +42,13 @@ impl Server {
 									continue;
 								}
 
-								let mut parts = line.splitn(4, ',');
 								let Some(id) = id else { continue };
+								if line.starts_with("bye,") {
+									writer.write_all(BodyBye::borrowed().with_receiver(id).with_sender(0).to_string().as_bytes()).await.ok();
+									break;
+								}
+
+								let mut parts = line.splitn(4, ',');
 								let Some(kind) = parts.next() else { continue };
 								let Some(receiver) = parts.next().and_then(|s| s.parse().ok()) else { continue };
 								let Some(sender) = parts.next().and_then(|s| s.parse::<u64>().ok()) else { continue };
@@ -78,31 +83,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,59 @@
+use tokio::io::{BufReader, Lines, ReadHalf, WriteHalf};
+
+pub(super) struct Stream;
+
+use tokio::io::AsyncBufReadExt;
+
+#[cfg(unix)]
+pub(super) type ClientReader = Lines<BufReader<ReadHalf<tokio::net::UnixStream>>>;
+#[cfg(not(unix))]
+pub(super) type ClientReader = Lines<BufReader<ReadHalf<tokio::net::TcpStream>>>;
+
+#[cfg(unix)]
+pub(super) type ClientWriter = WriteHalf<tokio::net::UnixStream>;
+#[cfg(not(unix))]
+pub(super) type ClientWriter = WriteHalf<tokio::net::TcpStream>;
+
+#[cfg(unix)]
+pub(super) type ServerListener = tokio::net::UnixListener;
+#[cfg(not(unix))]
+pub(super) type ServerListener = tokio::net::TcpListener;
+
+impl Stream {
+	#[cfg(unix)]
+	pub(super) 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(super) 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(super) 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(super) async fn bind() -> std::io::Result<ServerListener> {
+		tokio::net::TcpListener::bind("127.0.0.1:33581").await
+	}
+
+	#[cfg(unix)]
+	fn socket_file() -> std::path::PathBuf {
+		use uzers::Users;
+		use yazi_shared::Xdg;
+
+		use crate::USERS_CACHE;
+
+		Xdg::cache_dir().join(format!(".dds-{}.sock", USERS_CACHE.get_current_uid()))
+	}
+}

yazi-fm/Cargo.toml

@@ -9,27 +9,27 @@ homepage    = "https://yazi-rs.github.io"
 repository  = "https://github.com/sxyazi/yazi"
 
 [dependencies]
-yazi-adaptor   = { path = "../yazi-adaptor", version = "0.2.4" }
-yazi-boot      = { path = "../yazi-boot", version = "0.2.4" }
-yazi-config    = { path = "../yazi-config", version = "0.2.4" }
-yazi-core      = { path = "../yazi-core", version = "0.2.4" }
-yazi-dds       = { path = "../yazi-dds", version = "0.2.4" }
-yazi-plugin    = { path = "../yazi-plugin", version = "0.2.4" }
-yazi-proxy     = { path = "../yazi-proxy", version = "0.2.4" }
-yazi-shared    = { path = "../yazi-shared", version = "0.2.4" }
+yazi-adaptor = { path = "../yazi-adaptor", version = "0.2.4" }
+yazi-boot    = { path = "../yazi-boot", version = "0.2.4" }
+yazi-config  = { path = "../yazi-config", version = "0.2.4" }
+yazi-core    = { path = "../yazi-core", version = "0.2.4" }
+yazi-dds     = { path = "../yazi-dds", version = "0.2.4" }
+yazi-plugin  = { path = "../yazi-plugin", version = "0.2.4" }
+yazi-proxy   = { path = "../yazi-proxy", version = "0.2.4" }
+yazi-shared  = { path = "../yazi-shared", version = "0.2.4" }
 
 # External dependencies
-anyhow        = "1.0.82"
-better-panic  = "0.3.0"
-crossterm     = { version = "0.27.0", features = [ "event-stream" ] }
-fdlimit       = "0.3.0"
-futures       = "0.3.30"
-mlua          = { version = "0.9.7", features = [ "lua54", "vendored" ] }
-ratatui       = "0.26.1"
-scopeguard    = "1.2.0"
-syntect       = { version = "5.2.0", default-features = false, features = [ "parsing", "plist-load", "regex-onig" ] }
-tokio         = { version = "1.37.0", features = [ "full" ] }
-tokio-util    = "0.7.10"
+anyhow       = "1.0.82"
+better-panic = "0.3.0"
+crossterm    = { version = "0.27.0", features = [ "event-stream" ] }
+fdlimit      = "0.3.0"
+futures      = "0.3.30"
+mlua         = { version = "0.9.7", features = [ "lua54", "vendored" ] }
+ratatui      = "0.26.1"
+scopeguard   = "1.2.0"
+syntect      = { version = "5.2.0", default-features = false, features = [ "parsing", "plist-load", "regex-onig" ] }
+tokio        = { version = "1.37.0", features = [ "full" ] }
+tokio-util   = "0.7.10"
 
 # Logging
 tracing            = { version = "0.1.40", features = [ "max_level_debug", "release_max_level_warn" ] }

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

@@ -13,7 +13,12 @@ impl App {
 		};
 
 		let kind = payload.body.kind().to_owned();
-		let map = if payload.receiver == 0 { REMOTE.read() } else { LOCAL.read() };
+		let map = if payload.receiver == 0 || payload.receiver != payload.sender {
+			REMOTE.read()
+		} else {
+			LOCAL.read()
+		};
+
 		let Some(map) = map.get(&kind).filter(|&m| !m.is_empty()) else {
 			return;
 		};

yazi-fm/src/main.rs

@@ -49,11 +49,12 @@ async fn main() -> anyhow::Result<()> {
 
 	yazi_proxy::init();
 
-	yazi_dds::serve();
+	yazi_dds::init();
 
 	yazi_plugin::init();
 
 	yazi_core::init();
 
+	yazi_dds::serve();
 	app::App::serve().await
 }

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 }

yazi-plugin/src/utils/user.rs

@@ -6,7 +6,7 @@ impl Utils {
 	#[cfg(unix)]
 	pub(super) fn user(lua: &Lua, ya: &Table) -> mlua::Result<()> {
 		use uzers::{Groups, Users};
-		use yazi_boot::USERS_CACHE;
+		use yazi_dds::USERS_CACHE;
 		use yazi_shared::hostname;
 
 		use crate::utils::HOSTNAME_CACHE;