feat: DDS (Data Distribution Service) (#826)

Cargo.lock

@@ -2759,12 +2759,28 @@ dependencies = [
  "yazi-adaptor",
  "yazi-boot",
  "yazi-config",
+ "yazi-dds",
  "yazi-plugin",
  "yazi-proxy",
  "yazi-scheduler",
  "yazi-shared",
 ]
 
+[[package]]
+name = "yazi-dds"
+version = "0.2.4"
+dependencies = [
+ "anyhow",
+ "mlua",
+ "parking_lot",
+ "serde",
+ "serde_json",
+ "tokio",
+ "tracing",
+ "yazi-boot",
+ "yazi-shared",
+]
+
 [[package]]
 name = "yazi-fm"
 version = "0.2.4"
@@ -2790,6 +2806,7 @@ dependencies = [
  "yazi-boot",
  "yazi-config",
  "yazi-core",
+ "yazi-dds",
  "yazi-plugin",
  "yazi-proxy",
  "yazi-scheduler",
@@ -2802,6 +2819,7 @@ version = "0.2.4"
 dependencies = [
  "ansi-to-tui",
  "anyhow",
+ "arc-swap",
  "crossterm",
  "futures",
  "md-5",
@@ -2821,6 +2839,7 @@ dependencies = [
  "yazi-adaptor",
  "yazi-boot",
  "yazi-config",
+ "yazi-dds",
  "yazi-prebuild",
  "yazi-proxy",
  "yazi-shared",
@@ -2861,6 +2880,7 @@ dependencies = [
  "trash",
  "yazi-adaptor",
  "yazi-config",
+ "yazi-dds",
  "yazi-plugin",
  "yazi-proxy",
  "yazi-shared",

README.md

@@ -13,8 +13,9 @@ Yazi (means "duck") is a terminal file manager written in Rust, based on non-blo
 - 🖼️ **Built-in Support for Multiple Image Protocols**: Also integrated with Überzug++, covering almost all terminals.
 - 🌟 **Built-in Code Highlighting and Image Decoding**: Combined with the pre-loading mechanism, greatly accelerates image and normal file loading.
 - 🔌 **Concurrent Plugin System**: UI plugins (rewriting most of the UI), functional plugins, custom previewer, and custom preloader; Just some pieces of Lua.
+- 📡 **Data Distribution Service**: Built on a client-server architecture (no additional server process required), integrated with a Lua-based publish-subscribe model, achieving cross-instance communication and state persistence.
 - 🧰 Integration with fd, rg, fzf, zoxide
-- 💫 Vim-like input/select/notify component, auto-completion for cd paths
+- 💫 Vim-like input/select/which/notify component, auto-completion for cd paths
 - 🏷️ Multi-Tab Support, Cross-directory selection, Scrollable Preview (for videos, PDFs, archives, directories, code, etc.)
 - 🔄 Bulk Renaming, Visual Mode, File Chooser
 - 🎨 Theme System, Custom Layouts, Trash Bin, CSI u

cspell.json

@@ -1 +1 @@
-{"version":"0.2","flagWords":[],"language":"en","words":["Punct","KEYMAP","splitn","crossterm","YAZI","unar","peekable","ratatui","syntect","pbpaste","pbcopy","ffmpegthumbnailer","oneshot","Posix","Lsar","XADDOS","zoxide","cands","Deque","precache","imageops","IFBLK","IFCHR","IFDIR","IFIFO","IFLNK","IFMT","IFSOCK","IRGRP","IROTH","IRUSR","ISGID","ISUID","ISVTX","IWGRP","IWOTH","IWUSR","IXGRP","IXOTH","IXUSR","libc","winsize","TIOCGWINSZ","xpixel","ypixel","ioerr","appender","Catppuccin","macchiato","gitmodules","Dotfiles","bashprofile","vimrc","flac","webp","exiftool","mediainfo","ripgrep","nvim","indexmap","indexmap","unwatch","canonicalize","serde","fsevent","Ueberzug","iterm","wezterm","sixel","chafa","ueberzugpp","️ Überzug","️ Überzug","Konsole","Alacritty","Überzug","pkgs","paru","unarchiver","pdftoppm","poppler","prebuild","singlefile","jpegopt","EXIF","rustfmt","mktemp","nanos","xclip","xsel","natord","Mintty","nixos","nixpkgs","SIGTSTP","SIGCONT","SIGCONT","mlua","nonstatic","userdata","metatable","natsort","backstack","luajit","Succ","Succ","cand","fileencoding","foldmethod","lightgreen","darkgray","lightred","lightyellow","lightcyan","nushell","msvc","aarch","linemode","sxyazi","rsplit","ZELLIJ","bitflags","bitflags","USERPROFILE","Neovim","vergen","gitcl","Renderable","preloaders","prec","imagesize","Upserting","prio","Ghostty","Catmull","Lanczos","cmds","unyank","scrolloff","headsup"]}
+{"version":"0.2","language":"en","flagWords":[],"words":["Punct","KEYMAP","splitn","crossterm","YAZI","unar","peekable","ratatui","syntect","pbpaste","pbcopy","ffmpegthumbnailer","oneshot","Posix","Lsar","XADDOS","zoxide","cands","Deque","precache","imageops","IFBLK","IFCHR","IFDIR","IFIFO","IFLNK","IFMT","IFSOCK","IRGRP","IROTH","IRUSR","ISGID","ISUID","ISVTX","IWGRP","IWOTH","IWUSR","IXGRP","IXOTH","IXUSR","libc","winsize","TIOCGWINSZ","xpixel","ypixel","ioerr","appender","Catppuccin","macchiato","gitmodules","Dotfiles","bashprofile","vimrc","flac","webp","exiftool","mediainfo","ripgrep","nvim","indexmap","indexmap","unwatch","canonicalize","serde","fsevent","Ueberzug","iterm","wezterm","sixel","chafa","ueberzugpp","️ Überzug","️ Überzug","Konsole","Alacritty","Überzug","pkgs","paru","unarchiver","pdftoppm","poppler","prebuild","singlefile","jpegopt","EXIF","rustfmt","mktemp","nanos","xclip","xsel","natord","Mintty","nixos","nixpkgs","SIGTSTP","SIGCONT","SIGCONT","mlua","nonstatic","userdata","metatable","natsort","backstack","luajit","Succ","Succ","cand","fileencoding","foldmethod","lightgreen","darkgray","lightred","lightyellow","lightcyan","nushell","msvc","aarch","linemode","sxyazi","rsplit","ZELLIJ","bitflags","bitflags","USERPROFILE","Neovim","vergen","gitcl","Renderable","preloaders","prec","imagesize","Upserting","prio","Ghostty","Catmull","Lanczos","cmds","unyank","scrolloff","headsup","unsub"]}

yazi-adaptor/Cargo.toml

@@ -22,7 +22,7 @@ image        = "0.24.9"
 imagesize    = "0.12.0"
 kamadak-exif = "0.5.5"
 ratatui      = "0.26.1"
-tokio        = { version = "1.36.0", features = [ "parking_lot", "io-util", "process" ] }
+tokio        = { version = "1.36.0", features = [ "full" ] }
 
 # Logging
 tracing = { version = "0.1.40", features = [ "max_level_debug", "release_max_level_warn" ] }

yazi-adaptor/src/kitty.rs

@@ -393,7 +393,7 @@ impl Kitty {
 		for y in 0..rect.height {
 			write!(buf, "\x1b[{};{}H\x1b[38;5;1m", rect.y + y + 1, rect.x + 1)?;
 			for x in 0..rect.width {
-				write!(buf, "{}", '\u{10EEEE}')?;
+				write!(buf, "\u{10EEEE}")?;
 				write!(buf, "{}", *DIACRITICS.get(y as usize).unwrap_or(&DIACRITICS[0]))?;
 				write!(buf, "{}", *DIACRITICS.get(x as usize).unwrap_or(&DIACRITICS[0]))?;
 			}

yazi-boot/src/args.rs

@@ -12,7 +12,7 @@ pub struct Args {
 	/// Write the cwd on exit to this file
 	#[arg(long)]
 	pub cwd_file:     Option<PathBuf>,
-	/// Write the selected files on open emitted by the chooser mode
+	/// Write the selected files to this file on open fired
 	#[arg(long)]
 	pub chooser_file: Option<PathBuf>,
 
@@ -20,6 +20,13 @@ pub struct Args {
 	#[arg(long, action)]
 	pub clear_cache: bool,
 
+	/// Report the specified local events to stdout
+	#[arg(long, action)]
+	pub local_events:  Option<String>,
+	/// Report the specified remote events to stdout
+	#[arg(long, action)]
+	pub remote_events: Option<String>,
+
 	/// Print debug information
 	#[arg(long, action)]
 	pub debug: bool,

yazi-boot/src/boot.rs

@@ -1,4 +1,4 @@
-use std::{env, ffi::OsString, fmt::Write, path::{Path, PathBuf}, process};
+use std::{collections::HashSet, env, ffi::OsString, fmt::Write, path::{Path, PathBuf}, process};
 
 use clap::Parser;
 use serde::Serialize;
@@ -13,9 +13,13 @@ pub struct Boot {
 	pub cwd:  PathBuf,
 	pub file: Option<OsString>,
 
+	pub local_events:  HashSet<String>,
+	pub remote_events: HashSet<String>,
+
 	pub config_dir: PathBuf,
 	pub flavor_dir: PathBuf,
 	pub plugin_dir: PathBuf,
+	pub state_dir:  PathBuf,
 }
 
 impl Boot {
@@ -84,7 +88,7 @@ impl Boot {
 		writeln!(
 			s,
 			"    Version: {:?}",
-			Command::new(env::var_os("YAZI_FILE_TWO").unwrap_or("file".into())).arg("--version").output()
+			Command::new(env::var_os("YAZI_FILE_ONE").unwrap_or("file".into())).arg("--version").output()
 		)?;
 
 		writeln!(s, "\nText Opener")?;
@@ -129,13 +133,28 @@ impl Default for Boot {
 		let config_dir = Xdg::config_dir();
 		let (cwd, file) = Self::parse_entry(ARGS.entry.as_deref());
 
+		let local_events = ARGS
+			.local_events
+			.as_ref()
+			.map(|s| s.split(',').map(|s| s.to_owned()).collect())
+			.unwrap_or_default();
+		let remote_events = ARGS
+			.remote_events
+			.as_ref()
+			.map(|s| s.split(',').map(|s| s.to_owned()).collect())
+			.unwrap_or_default();
+
 		let boot = Self {
 			cwd,
 			file,
 
+			local_events,
+			remote_events,
+
 			flavor_dir: config_dir.join("flavors"),
 			plugin_dir: config_dir.join("plugins"),
 			config_dir,
+			state_dir: Xdg::state_dir(),
 		};
 
 		std::fs::create_dir_all(&boot.flavor_dir).expect("Failed to create flavor directory");

yazi-config/src/plugin/plugin.rs

@@ -73,8 +73,8 @@ impl Plugin {
 			.iter()
 			.filter(|&rule| {
 				rule.cond.as_ref().and_then(|c| c.eval(f)) != Some(false)
-					&& (rule.name.as_ref().is_some_and(|n| n.match_path(path, is_folder))
+					&& (rule.mime.as_ref().zip(mime).map_or(false, |(m, s)| m.matches(s))
-						|| rule.mime.as_ref().zip(mime).map_or(false, |(m, s)| m.matches(s)))
+						|| rule.name.as_ref().is_some_and(|n| n.match_path(path, is_folder)))
 			})
 			.collect()
 	}

yazi-config/src/plugin/rule.rs

@@ -5,6 +5,7 @@ use yazi_shared::{event::Cmd, Condition};
 
 use crate::{Pattern, Priority, DEPRECATED_EXEC};
 
+#[derive(Debug)]
 pub struct PluginRule {
 	pub id:    u8,
 	pub cond:  Option<Condition>,

yazi-core/Cargo.toml

@@ -12,6 +12,7 @@ repository  = "https://github.com/sxyazi/yazi"
 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-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-scheduler = { path = "../yazi-scheduler", version = "0.2.4" }
@@ -29,7 +30,7 @@ parking_lot   = "0.12.1"
 ratatui       = "0.26.1"
 regex         = "1.10.3"
 serde         = "1.0.197"
-tokio         = { version = "1.36.0", features = [ "parking_lot", "macros", "rt-multi-thread", "sync", "time", "fs", "process", "io-std", "io-util" ] }
+tokio         = { version = "1.36.0", features = [ "full" ] }
 tokio-stream  = "0.1.15"
 tokio-util    = "0.7.10"
 unicode-width = "0.1.11"

yazi-core/src/clipboard.rs

@@ -60,7 +60,7 @@ impl Clipboard {
 		use tokio::{io::AsyncWriteExt, process::Command};
 		use yazi_shared::in_ssh_connection;
 
-		*self.content.lock() = s.as_ref().to_owned();
+		s.as_ref().clone_into(&mut self.content.lock());
 		if in_ssh_connection() {
 			execute!(BufWriter::new(stderr()), osc52::SetClipboard::new(s.as_ref())).ok();
 		}

yazi-core/src/input/input.rs

@@ -86,7 +86,6 @@ impl Input {
 		true
 	}
 
-	#[inline]
 	pub(super) fn flush_value(&mut self) {
 		self.ticket = self.ticket.wrapping_add(1);
 

yazi-core/src/manager/commands/bulk_rename.rs

@@ -77,7 +77,7 @@ impl Manager {
 			return Ok(());
 		}
 
-		let _permit = WATCHER.acquire().await.unwrap();
+		let permit = WATCHER.acquire().await.unwrap();
 		let (mut failed, mut succeeded) = (Vec::new(), HashMap::with_capacity(todo.len()));
 		for (o, n) in todo {
 			let (old, new) = (root.join(&o), root.join(&n));
@@ -96,7 +96,7 @@ impl Manager {
 		if !succeeded.is_empty() {
 			FilesOp::Upserting(cwd, succeeded).emit();
 		}
-		drop(_permit);
+		drop(permit);
 
 		if !failed.is_empty() {
 			Self::output_failed(failed).await?;

yazi-core/src/manager/commands/close.rs

@@ -5,7 +5,7 @@ use crate::{manager::Manager, tasks::Tasks};
 impl Manager {
 	pub fn close(&mut self, _: Cmd, tasks: &Tasks) {
 		if self.tabs.len() > 1 {
-			return self.tabs.close(self.tabs.idx);
+			return self.tabs.close(self.tabs.cursor);
 		}
 		self.quit((), tasks);
 	}

yazi-core/src/manager/commands/rename.rs

@@ -3,6 +3,7 @@ use std::collections::HashMap;
 use anyhow::Result;
 use tokio::fs;
 use yazi_config::popup::InputCfg;
+use yazi_dds::Pubsub;
 use yazi_proxy::{InputProxy, ManagerProxy, WATCHER};
 use yazi_shared::{event::Cmd, fs::{accessible, File, FilesOp, Url}};
 
@@ -49,6 +50,7 @@ impl Manager {
 			_ => None,
 		};
 
+		let tab = self.tabs.cursor;
 		tokio::spawn(async move {
 			let mut result = InputProxy::show(InputCfg::rename().with_value(name).with_cursor(cursor));
 			let Some(Ok(name)) = result.recv().await else {
@@ -57,20 +59,20 @@ impl Manager {
 
 			let new = hovered.parent().unwrap().join(name);
 			if opt.force || !accessible(&new).await {
-				Self::rename_do(hovered, Url::from(new)).await.ok();
+				Self::rename_do(tab, hovered, Url::from(new)).await.ok();
 				return;
 			}
 
 			let mut result = InputProxy::show(InputCfg::overwrite());
 			if let Some(Ok(choice)) = result.recv().await {
 				if choice == "y" || choice == "Y" {
-					Self::rename_do(hovered, Url::from(new)).await.ok();
+					Self::rename_do(tab, hovered, Url::from(new)).await.ok();
 				}
 			};
 		});
 	}
 
-	async fn rename_do(old: Url, new: Url) -> Result<()> {
+	async fn rename_do(tab: usize, old: Url, new: Url) -> Result<()> {
 		let _permit = WATCHER.acquire().await.unwrap();
 
 		fs::rename(&old, &new).await?;
@@ -79,6 +81,8 @@ impl Manager {
 		}
 
 		let file = File::from(new.clone()).await?;
+		Pubsub::pub_from_rename(tab, &old, &new);
+
 		FilesOp::Deleting(file.parent().unwrap(), vec![new.clone()]).emit();
 		FilesOp::Upserting(file.parent().unwrap(), HashMap::from_iter([(old, file)])).emit();
 		Ok(ManagerProxy::hover(Some(new)))

yazi-core/src/manager/commands/suspend.rs

@@ -1,4 +1,3 @@
-use yazi_proxy::AppProxy;
 use yazi_shared::event::Cmd;
 
 use crate::manager::Manager;
@@ -7,7 +6,7 @@ impl Manager {
 	pub fn suspend(&mut self, _: Cmd) {
 		#[cfg(unix)]
 		tokio::spawn(async move {
-			AppProxy::stop().await;
+			yazi_proxy::AppProxy::stop().await;
 			unsafe { libc::raise(libc::SIGTSTP) };
 		});
 	}

yazi-core/src/manager/commands/tab_close.rs

@@ -26,10 +26,11 @@ impl Tabs {
 		}
 
 		self.items.remove(opt.idx).shutdown();
-		if opt.idx <= self.idx {
+		if opt.idx <= self.cursor {
 			self.set_idx(self.absolute(1));
 		}
 
+		self.reorder();
 		render!();
 	}
 }

yazi-core/src/manager/commands/tab_create.rs

@@ -33,8 +33,9 @@ impl Tabs {
 		tab.conf = self.active().conf.clone();
 		tab.apply_files_attrs();
 
-		self.items.insert(self.idx + 1, tab);
+		self.items.insert(self.cursor + 1, tab);
-		self.set_idx(self.idx + 1);
+		self.set_idx(self.cursor + 1);
+		self.reorder();
 		render!();
 	}
 }

yazi-core/src/manager/commands/tab_swap.rs

@@ -15,12 +15,13 @@ impl From<Cmd> for Opt {
 impl Tabs {
 	pub fn swap(&mut self, opt: impl Into<Opt>) {
 		let idx = self.absolute(opt.into().step);
-		if idx == self.idx {
+		if idx == self.cursor {
 			return;
 		}
 
-		self.items.swap(self.idx, idx);
+		self.items.swap(self.cursor, idx);
 		self.set_idx(idx);
+		self.reorder();
 		render!();
 	}
 }

yazi-core/src/manager/commands/tab_switch.rs

@@ -20,12 +20,12 @@ impl Tabs {
 	pub fn switch(&mut self, opt: impl Into<Opt>) {
 		let opt = opt.into() as Opt;
 		let idx = if opt.relative {
-			(self.idx as isize + opt.step).rem_euclid(self.items.len() as isize) as usize
+			(self.cursor as isize + opt.step).rem_euclid(self.items.len() as isize) as usize
 		} else {
 			opt.step as usize
 		};
 
-		if idx == self.idx || idx >= self.items.len() {
+		if idx == self.cursor || idx >= self.items.len() {
 			return;
 		}
 

yazi-core/src/manager/commands/update_files.rs

@@ -27,7 +27,7 @@ impl Manager {
 		}
 
 		for op in ops {
-			let idx = self.tabs.idx;
+			let idx = self.tabs.cursor;
 			self.yanked.apply_op(&op);
 
 			for (_, tab) in self.tabs.iter_mut().enumerate().filter(|(i, _)| *i != idx) {

yazi-core/src/manager/commands/update_mimetype.rs

@@ -1,6 +1,6 @@
 use std::collections::HashMap;
 
-use yazi_plugin::ValueSendable;
+use yazi_dds::ValueSendable;
 use yazi_shared::{event::Cmd, fs::Url, render};
 
 use crate::{manager::Manager, tasks::Tasks};

yazi-core/src/manager/commands/yank.rs

@@ -1,5 +1,4 @@
-use std::collections::HashSet;
+use yazi_dds::Pubsub;
-
 use yazi_shared::{event::Cmd, render};
 
 use crate::manager::{Manager, Yanked};
@@ -18,13 +17,14 @@ impl Manager {
 			return;
 		}
 
-		let selected: HashSet<_> = self.selected_or_hovered(false).into_iter().cloned().collect();
+		self.yanked = Yanked {
-		if selected.is_empty() {
+			cut:  opt.into().cut,
-			return;
+			urls: self.selected_or_hovered(false).into_iter().cloned().collect(),
-		}
+		};
 
-		self.yanked = Yanked { cut: opt.into().cut, urls: selected };
 		self.active_mut().escape_select();
+		Pubsub::pub_from_yank(self.yanked.cut, &self.yanked.urls);
+
 		render!();
 	}
 }

yazi-core/src/manager/tabs.rs

@@ -7,13 +7,13 @@ use yazi_shared::fs::Url;
 use crate::tab::Tab;
 
 pub struct Tabs {
-	pub idx:          usize,
+	pub cursor:       usize,
 	pub(super) items: Vec<Tab>,
 }
 
 impl Tabs {
 	pub fn make() -> Self {
-		let mut tabs = Self { idx: 0, items: vec![Tab::from(Url::from(&BOOT.cwd))] };
+		let mut tabs = Self { cursor: 0, items: vec![Tab::from(Url::from(&BOOT.cwd))] };
 		if let Some(file) = &BOOT.file {
 			tabs.items[0].reveal(Url::from(BOOT.cwd.join(file)));
 		}
@@ -25,24 +25,28 @@ impl Tabs {
 	#[inline]
 	pub(super) fn absolute(&self, rel: isize) -> usize {
 		if rel > 0 {
-			(self.idx + rel as usize).min(self.items.len() - 1)
+			(self.cursor + rel as usize).min(self.items.len() - 1)
 		} else {
-			self.idx.saturating_sub(rel.unsigned_abs())
+			self.cursor.saturating_sub(rel.unsigned_abs())
 		}
 	}
 
 	#[inline]
+	pub(super) fn reorder(&mut self) {
+		self.items.iter_mut().enumerate().for_each(|(i, tab)| tab.idx = i);
+	}
+
 	pub(super) fn set_idx(&mut self, idx: usize) {
-		if self.idx == idx {
+		if self.cursor == idx {
 			return;
 		}
 
 		// Reset the preview of the previous active tab
-		if let Some(active) = self.items.get_mut(self.idx) {
+		if let Some(active) = self.items.get_mut(self.cursor) {
 			active.preview.reset_image();
 		}
 
-		self.idx = idx;
+		self.cursor = idx;
 		ManagerProxy::refresh();
 		ManagerProxy::peek(true);
 	}
@@ -50,10 +54,10 @@ impl Tabs {
 
 impl Tabs {
 	#[inline]
-	pub fn active(&self) -> &Tab { &self.items[self.idx] }
+	pub fn active(&self) -> &Tab { &self.items[self.cursor] }
 
 	#[inline]
-	pub(super) fn active_mut(&mut self) -> &mut Tab { &mut self.items[self.idx] }
+	pub(super) fn active_mut(&mut self) -> &mut Tab { &mut self.items[self.cursor] }
 }
 
 impl Deref for Tabs {

yazi-core/src/tab/commands/arrow.rs

@@ -1,3 +1,4 @@
+use yazi_dds::Pubsub;
 use yazi_proxy::ManagerProxy;
 use yazi_shared::{event::Cmd, render};
 
@@ -37,6 +38,7 @@ impl Tab {
 			}
 		}
 
+		Pubsub::pub_from_hover(self.idx, self.current.hovered().map(|h| &h.url));
 		ManagerProxy::hover(None);
 		render!();
 	}

yazi-core/src/tab/commands/cd.rs

@@ -3,6 +3,7 @@ use std::{mem, time::Duration};
 use tokio::{fs, pin};
 use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt};
 use yazi_config::popup::InputCfg;
+use yazi_dds::Pubsub;
 use yazi_proxy::{CompletionProxy, InputProxy, ManagerProxy, TabProxy};
 use yazi_shared::{event::Cmd, fs::{expand_path, Url}, render, Debounce, InputError};
 
@@ -64,6 +65,7 @@ impl Tab {
 			self.backstack.push(opt.target.clone());
 		}
 
+		Pubsub::pub_from_cd(self.idx, &self.current.cwd);
 		ManagerProxy::refresh();
 		render!();
 	}

yazi-core/src/tab/tab.rs

@@ -8,6 +8,7 @@ use super::{Backstack, Config, Finder, Mode, Preview};
 use crate::{folder::{Folder, FolderStage}, tab::Selected};
 
 pub struct Tab {
+	pub idx:     usize,
 	pub mode:    Mode,
 	pub conf:    Config,
 	pub current: Folder,
@@ -27,6 +28,7 @@ impl From<Url> for Tab {
 		let parent = url.parent_url().map(Folder::from);
 
 		Self {
+			idx: 0,
 			mode: Default::default(),
 			current: Folder::from(url.clone()),
 			parent,

yazi-core/src/tasks/plugin.rs

@@ -1,4 +1,4 @@
-use yazi_plugin::ValueSendable;
+use yazi_dds::ValueSendable;
 
 use super::Tasks;
 

yazi-dds/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name        = "yazi-dds"
+version     = "0.2.4"
+edition     = "2021"
+license     = "MIT"
+authors     = [ "sxyazi <sxyazi@gmail.com>" ]
+description = "Yazi data distribution service"
+homepage    = "https://yazi-rs.github.io"
+repository  = "https://github.com/sxyazi/yazi"
+
+[dependencies]
+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.114"
+tokio       = { version = "1.36.0", features = [ "full" ] }
+
+# Logging
+tracing = { version = "0.1.40", features = [ "max_level_debug", "release_max_level_warn" ] }

yazi-dds/src/body/body.rs

@@ -0,0 +1,106 @@
+use anyhow::Result;
+use mlua::{ExternalResult, IntoLua, Lua, Value};
+use serde::{Deserialize, Serialize};
+
+use super::{BodyBulk, BodyCd, BodyCustom, BodyHey, BodyHi, BodyHover, BodyRename, BodyTabs, BodyYank};
+use crate::Payload;
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum Body<'a> {
+	Hi(BodyHi),
+	Hey(BodyHey),
+	Tabs(BodyTabs<'a>),
+	Cd(BodyCd<'a>),
+	Hover(BodyHover<'a>),
+	Rename(BodyRename<'a>),
+	Bulk(BodyBulk<'a>),
+	Yank(BodyYank<'a>),
+	Custom(BodyCustom),
+}
+
+impl<'a> Body<'a> {
+	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)?),
+			"tabs" => Body::Tabs(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)?),
+			_ => BodyCustom::from_str(kind, body)?,
+		})
+	}
+
+	pub fn from_lua(kind: &str, value: Value) -> Result<Self> {
+		Ok(match kind {
+			"hi" | "hey" | "tabs" | "cd" | "hover" | "rename" | "bulk" | "yank" => {
+				Err("Cannot construct system event from Lua").into_lua_err()?
+			}
+			_ => BodyCustom::from_lua(kind, value)?,
+		})
+	}
+
+	#[inline]
+	pub fn kind(&self) -> &str {
+		match self {
+			Self::Hi(_) => "hi",
+			Self::Hey(_) => "hey",
+			Self::Tabs(_) => "tabs",
+			Self::Cd(_) => "cd",
+			Self::Hover(_) => "hover",
+			Self::Rename(_) => "rename",
+			Self::Bulk(_) => "bulk",
+			Self::Yank(_) => "yank",
+			Self::Custom(b) => b.kind.as_str(),
+		}
+	}
+
+	pub fn tab(kind: &str, body: &str) -> usize {
+		match kind {
+			"cd" | "hover" | "bulk" | "rename" => {}
+			_ => return 0,
+		}
+
+		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,
+			_ => 0,
+		}
+	}
+
+	pub fn upgrade(self) -> Payload<'a> {
+		let severity = match self {
+			Body::Hi(_) => 0,
+			Body::Hey(_) => 0,
+			Body::Tabs(_) => 10,
+			Body::Cd(_) => 20,
+			Body::Hover(_) => 30,
+			Body::Rename(_) => 0,
+			Body::Bulk(_) => 0,
+			Body::Yank(_) => 40,
+			Body::Custom(_) => 0,
+		};
+		Payload::new(self).with_severity(severity)
+	}
+}
+
+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::Tabs(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::Custom(b) => b.into_lua(lua),
+		}
+	}
+}

yazi-dds/src/body/bulk.rs

@@ -0,0 +1,70 @@
+use std::{borrow::Cow, collections::{hash_map, HashMap}};
+
+use mlua::{AnyUserData, IntoLua, IntoLuaMulti, Lua, MetaMethod, UserData, UserDataRefMut, Value};
+use serde::{Deserialize, Serialize};
+use yazi_shared::fs::Url;
+
+use super::Body;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BodyBulk<'a> {
+	pub tab:     usize,
+	pub changes: Cow<'a, HashMap<Url, Url>>,
+}
+
+impl<'a> BodyBulk<'a> {
+	#[inline]
+	pub fn borrowed(tab: usize, changes: &'a HashMap<Url, Url>) -> Body<'a> {
+		Self { tab, changes: Cow::Borrowed(changes) }.into()
+	}
+}
+
+impl BodyBulk<'static> {
+	#[inline]
+	pub fn owned(tab: usize, changes: &HashMap<Url, Url>) -> Body<'static> {
+		Self { tab, changes: Cow::Owned(changes.clone()) }.into()
+	}
+}
+
+impl<'a> From<BodyBulk<'a>> for Body<'a> {
+	fn from(value: BodyBulk<'a>) -> Self { Self::Bulk(value) }
+}
+
+impl IntoLua<'_> for BodyBulk<'static> {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> {
+		lua
+			.create_any_userdata(BodyBulkIter {
+				tab:   self.tab,
+				inner: self.changes.into_owned().into_iter(),
+			})?
+			.into_lua(lua)
+	}
+}
+
+// --- Iterator
+pub struct BodyBulkIter {
+	pub tab:   usize,
+	pub inner: hash_map::IntoIter<Url, Url>,
+}
+
+impl UserData for BodyBulkIter {
+	fn add_fields<'a, F: mlua::UserDataFields<'a, Self>>(fields: &mut F) {
+		fields.add_field_method_get("tab", |_, me| Ok(me.tab));
+	}
+
+	fn add_methods<'a, M: mlua::UserDataMethods<'a, Self>>(methods: &mut M) {
+		methods.add_meta_method(MetaMethod::Len, |_, me, ()| Ok(me.inner.len()));
+
+		methods.add_meta_function(MetaMethod::Pairs, |lua, me: AnyUserData| {
+			let iter = lua.create_function(|lua, mut me: UserDataRefMut<Self>| {
+				if let Some((from, to)) = me.inner.next() {
+					(lua.create_any_userdata(from)?, lua.create_any_userdata(to)?).into_lua_multi(lua)
+				} else {
+					().into_lua_multi(lua)
+				}
+			})?;
+
+			Ok((iter, me))
+		});
+	}
+}

yazi-dds/src/body/cd.rs

@@ -0,0 +1,46 @@
+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 BodyCd<'a> {
+	pub owned: bool,
+	pub tab:   usize,
+	pub url:   Cow<'a, Url>,
+}
+
+impl<'a> BodyCd<'a> {
+	#[inline]
+	pub fn borrowed(tab: usize, url: &'a Url) -> Body<'a> {
+		Self { owned: false, tab, url: Cow::Borrowed(url) }.into()
+	}
+}
+
+impl BodyCd<'static> {
+	#[inline]
+	pub fn owned(tab: usize) -> Body<'static> {
+		Self { owned: false, tab, url: Default::default() }.into()
+	}
+}
+
+impl<'a> From<BodyCd<'a>> for Body<'a> {
+	fn from(value: BodyCd<'a>) -> Self { Self::Cd(value) }
+}
+
+impl IntoLua<'_> for BodyCd<'static> {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> {
+		if self.owned {
+			lua.create_table_from([
+				("tab", self.tab.into_lua(lua)?),
+				("url", lua.create_any_userdata(self.url.into_owned())?.into_lua(lua)?),
+			])?
+		} else {
+			lua.create_table_from([("tab", self.tab)])?
+		}
+		.into_lua(lua)
+	}
+}

yazi-dds/src/body/custom.rs

@@ -0,0 +1,35 @@
+use mlua::{IntoLua, Lua, Value};
+use serde::{Deserialize, Serialize};
+
+use super::Body;
+use crate::ValueSendable;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BodyCustom {
+	#[serde(skip)]
+	pub kind:  String,
+	#[serde(flatten)]
+	pub value: ValueSendable,
+}
+
+impl BodyCustom {
+	#[inline]
+	pub fn from_str(kind: &str, value: &str) -> anyhow::Result<Body<'static>> {
+		let mut me = serde_json::from_str::<Self>(value)?;
+		kind.clone_into(&mut me.kind);
+		Ok(me.into())
+	}
+
+	#[inline]
+	pub fn from_lua(kind: &str, value: Value) -> mlua::Result<Body<'static>> {
+		Ok(Self { kind: kind.to_owned(), value: value.try_into()? }.into())
+	}
+}
+
+impl From<BodyCustom> for Body<'_> {
+	fn from(value: BodyCustom) -> Self { Self::Custom(value) }
+}
+
+impl IntoLua<'_> for BodyCustom {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> { self.value.into_lua(lua) }
+}

yazi-dds/src/body/hey.rs

@@ -0,0 +1,22 @@
+use std::collections::HashMap;
+
+use mlua::{ExternalResult, IntoLua, Lua, Value};
+use serde::{Deserialize, Serialize};
+
+use super::Body;
+use crate::Peer;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BodyHey {
+	pub peers: HashMap<u64, Peer>,
+}
+
+impl From<BodyHey> for Body<'_> {
+	fn from(value: BodyHey) -> Self { Self::Hey(value) }
+}
+
+impl IntoLua<'_> for BodyHey {
+	fn into_lua(self, _: &Lua) -> mlua::Result<Value<'_>> {
+		Err("BodyHey cannot be converted to Lua").into_lua_err()
+	}
+}

yazi-dds/src/body/hi.rs

@@ -0,0 +1,22 @@
+use std::collections::HashSet;
+
+use mlua::{ExternalResult, IntoLua, Lua, Value};
+use serde::{Deserialize, Serialize};
+
+use super::Body;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BodyHi {
+	pub id:        u64,
+	pub abilities: HashSet<String>,
+}
+
+impl From<BodyHi> for Body<'_> {
+	fn from(value: BodyHi) -> Self { Self::Hi(value) }
+}
+
+impl IntoLua<'_> for BodyHi {
+	fn into_lua(self, _: &Lua) -> mlua::Result<Value<'_>> {
+		Err("BodyHi cannot be converted to Lua").into_lua_err()
+	}
+}

yazi-dds/src/body/hover.rs

@@ -0,0 +1,44 @@
+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 BodyHover<'a> {
+	pub owned: bool,
+	pub tab:   usize,
+	pub url:   Option<Cow<'a, Url>>,
+}
+
+impl<'a> BodyHover<'a> {
+	#[inline]
+	pub fn borrowed(tab: usize, url: Option<&'a Url>) -> Body<'a> {
+		Self { owned: false, tab, url: url.map(Cow::Borrowed) }.into()
+	}
+}
+
+impl BodyHover<'static> {
+	#[inline]
+	pub fn owned(tab: usize) -> Body<'static> { Self { owned: false, tab, url: None }.into() }
+}
+
+impl<'a> From<BodyHover<'a>> for Body<'a> {
+	fn from(value: BodyHover<'a>) -> Self { Self::Hover(value) }
+}
+
+impl IntoLua<'_> for BodyHover<'static> {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> {
+		if self.owned && self.url.is_some() {
+			lua.create_table_from([
+				("tab", self.tab.into_lua(lua)?),
+				("url", lua.create_any_userdata(self.url.unwrap().into_owned())?.into_lua(lua)?),
+			])?
+		} else {
+			lua.create_table_from([("tab", self.tab)])?
+		}
+		.into_lua(lua)
+	}
+}

yazi-dds/src/body/mod.rs

@@ -0,0 +1,23 @@
+#![allow(clippy::module_inception)]
+
+mod body;
+mod bulk;
+mod cd;
+mod custom;
+mod hey;
+mod hi;
+mod hover;
+mod rename;
+mod tabs;
+mod yank;
+
+pub use body::*;
+pub use bulk::*;
+pub use cd::*;
+pub use custom::*;
+pub use hey::*;
+pub use hi::*;
+pub use hover::*;
+pub use rename::*;
+pub use tabs::*;
+pub use yank::*;

yazi-dds/src/body/rename.rs

@@ -0,0 +1,44 @@
+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 BodyRename<'a> {
+	pub tab:  usize,
+	pub from: Cow<'a, Url>,
+	pub to:   Cow<'a, Url>,
+}
+
+impl<'a> BodyRename<'a> {
+	#[inline]
+	pub fn borrowed(tab: usize, from: &'a Url, to: &'a Url) -> Body<'a> {
+		Self { tab, from: Cow::Borrowed(from), to: Cow::Borrowed(to) }.into()
+	}
+}
+
+impl BodyRename<'static> {
+	#[inline]
+	pub fn owned(tab: usize, from: &Url, to: &Url) -> Body<'static> {
+		Self { tab, from: Cow::Owned(from.clone()), to: Cow::Owned(to.clone()) }.into()
+	}
+}
+
+impl<'a> From<BodyRename<'a>> for Body<'a> {
+	fn from(value: BodyRename<'a>) -> Self { Self::Rename(value) }
+}
+
+impl IntoLua<'_> for BodyRename<'static> {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> {
+		lua
+			.create_table_from([
+				("tab", self.tab.into_lua(lua)?),
+				("from", lua.create_any_userdata(self.from.into_owned())?.into_lua(lua)?),
+				("to", lua.create_any_userdata(self.to.into_owned())?.into_lua(lua)?),
+			])?
+			.into_lua(lua)
+	}
+}

yazi-dds/src/body/tabs.rs

@@ -0,0 +1,83 @@
+use std::borrow::Cow;
+
+use mlua::{IntoLua, Lua, MetaMethod, UserData, Value};
+use serde::{Deserialize, Serialize};
+use yazi_shared::fs::Url;
+
+use super::Body;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BodyTabs<'a> {
+	pub owned:  bool,
+	pub cursor: usize,
+	pub items:  Vec<BodyTabsItem<'a>>,
+}
+
+impl<'a> BodyTabs<'a> {
+	#[inline]
+	pub fn borrowed(cursor: usize, urls: &[&'a Url]) -> Body<'a> {
+		Self {
+			owned: false,
+			cursor,
+			items: urls.iter().map(|&u| BodyTabsItem { url: Cow::Borrowed(u) }).collect(),
+		}
+		.into()
+	}
+}
+
+impl BodyTabs<'static> {
+	#[inline]
+	pub fn owned(cursor: usize) -> Body<'static> {
+		Self { owned: false, cursor, items: Default::default() }.into()
+	}
+}
+
+impl<'a> From<BodyTabs<'a>> for Body<'a> {
+	fn from(value: BodyTabs<'a>) -> Self { Self::Tabs(value) }
+}
+
+impl IntoLua<'_> for BodyTabs<'static> {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value<'_>> {
+		if self.owned {
+			BodyTabsIter::from(self).into_lua(lua)
+		} else {
+			lua.create_table_from([("cursor", self.cursor)])?.into_lua(lua)
+		}
+	}
+}
+
+// --- Item
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct BodyTabsItem<'a> {
+	pub url: Cow<'a, Url>,
+}
+
+impl UserData for BodyTabsItem<'static> {
+	fn add_fields<'a, F: mlua::UserDataFields<'a, Self>>(fields: &mut F) {
+		fields.add_field_method_get("url", |lua, me| lua.create_any_userdata(me.url.clone()));
+	}
+}
+
+// --- Iterator
+pub struct BodyTabsIter {
+	pub cursor: usize,
+	pub items:  Vec<BodyTabsItem<'static>>,
+}
+
+impl From<BodyTabs<'static>> for BodyTabsIter {
+	fn from(value: BodyTabs<'static>) -> Self { Self { cursor: value.cursor, items: value.items } }
+}
+
+impl UserData for BodyTabsIter {
+	fn add_fields<'a, F: mlua::UserDataFields<'a, Self>>(fields: &mut F) {
+		fields.add_field_method_get("cursor", |_, me| Ok(me.cursor));
+	}
+
+	fn add_methods<'a, M: mlua::UserDataMethods<'a, Self>>(methods: &mut M) {
+		methods.add_meta_method(MetaMethod::Len, |_, me, ()| Ok(me.items.len()));
+
+		methods.add_meta_method(MetaMethod::Index, |_, me, idx: usize| {
+			if idx > me.items.len() || idx == 0 { Ok(None) } else { Ok(Some(me.items[idx - 1].clone())) }
+		});
+	}
+}

yazi-dds/src/body/yank.rs

@@ -0,0 +1,72 @@
+use std::{borrow::Cow, collections::HashSet};
+
+use mlua::{IntoLua, Lua, MetaMethod, UserData, Value};
+use serde::{Deserialize, Serialize};
+use yazi_shared::fs::Url;
+
+use super::Body;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BodyYank<'a> {
+	pub owned: bool,
+	pub cut:   bool,
+	pub urls:  Cow<'a, HashSet<Url>>,
+}
+
+impl<'a> BodyYank<'a> {
+	#[inline]
+	pub fn borrowed(cut: bool, urls: &'a HashSet<Url>) -> Body<'a> {
+		Self { owned: false, cut, urls: Cow::Borrowed(urls) }.into()
+	}
+}
+
+impl BodyYank<'static> {
+	#[inline]
+	pub fn owned(cut: bool) -> Body<'static> {
+		Self { owned: false, cut, urls: Default::default() }.into()
+	}
+}
+
+impl<'a> From<BodyYank<'a>> for Body<'a> {
+	fn from(value: BodyYank<'a>) -> Self { Self::Yank(value) }
+}
+
+impl IntoLua<'_> for BodyYank<'static> {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value<'_>> {
+		if self.owned {
+			BodyYankIter::from(self).into_lua(lua)
+		} else {
+			lua.create_table_from([("cut", self.cut)])?.into_lua(lua)
+		}
+	}
+}
+
+// --- Iterator
+pub struct BodyYankIter {
+	pub cut:  bool,
+	pub urls: Vec<Url>,
+}
+
+impl From<BodyYank<'static>> for BodyYankIter {
+	fn from(value: BodyYank) -> Self {
+		Self { cut: value.cut, urls: value.urls.into_owned().into_iter().collect() }
+	}
+}
+
+impl UserData for BodyYankIter {
+	fn add_fields<'a, F: mlua::UserDataFields<'a, Self>>(fields: &mut F) {
+		fields.add_field_method_get("is_cut", |_, me| Ok(me.cut));
+	}
+
+	fn add_methods<'a, M: mlua::UserDataMethods<'a, Self>>(methods: &mut M) {
+		methods.add_meta_method(MetaMethod::Len, |_, me, ()| Ok(me.urls.len()));
+
+		methods.add_meta_method(MetaMethod::Index, |lua, me, idx: usize| {
+			if idx > me.urls.len() || idx == 0 {
+				Ok(None)
+			} else {
+				Some(lua.create_any_userdata(me.urls[idx - 1].clone())).transpose()
+			}
+		});
+	}
+}

yazi-dds/src/client.rs

@@ -0,0 +1,151 @@
+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 yazi_shared::RoCell;
+
+use crate::{body::Body, 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,
+	pub(super) tx:        mpsc::UnboundedSender<String>,
+	pub(super) abilities: HashSet<String>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Peer {
+	pub(super) abilities: HashSet<String>,
+}
+
+impl Client {
+	pub(super) fn serve(mut rx: mpsc::UnboundedReceiver<String>) {
+		while rx.try_recv().is_ok() {}
+
+		tokio::spawn(async move {
+			let mut server = None;
+			let (mut lines, mut writer) = Self::connect(&mut server).await;
+
+			loop {
+				select! {
+					Some(payload) = rx.recv() => {
+						if writer.write_all(payload.as_bytes()).await.is_err() {
+							(lines, writer) = Self::reconnect(&mut server).await;
+							writer.write_all(payload.as_bytes()).await.ok(); // Retry once
+						}
+					}
+					Ok(next) = lines.next_line() => {
+						let Some(line) = next else {
+							(lines, writer) = Self::reconnect(&mut server).await;
+							continue;
+						};
+
+						if line.starts_with("hey,") {
+							Self::handle_hey(line);
+						} else {
+							Payload::from_str(&line).map(|p| p.flush(false).emit()).ok();
+						}
+					}
+				}
+			}
+		});
+	}
+
+	#[inline]
+	pub(super) fn push(payload: Payload) { QUEUE.send(format!("{}\n", payload)).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>) {
+		let mut first = true;
+		loop {
+			if let Ok(stream) = UnixStream::connect("/tmp/yazi.sock").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 server.is_some() {
+				super::STATE.load().await.ok();
+			}
+
+			if mem::replace(&mut first, false) && server.is_some() {
+				continue;
+			}
+			time::sleep(time::Duration::from_secs(1)).await;
+		}
+	}
+
+	#[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>) {
+		PEERS.write().clear();
+
+		time::sleep(time::Duration::from_millis(500)).await;
+		Self::connect(server).await
+	}
+
+	fn handle_hey(s: String) {
+		if let Ok(Body::Hey(mut hey)) = Payload::from_str(&s).map(|p| p.body) {
+			hey.peers.retain(|&id, _| id != *ID);
+			*PEERS.write() = hey.peers;
+		}
+	}
+}
+
+impl Peer {
+	#[inline]
+	pub(super) fn new(abilities: &HashSet<String>) -> Self { Self { abilities: abilities.clone() } }
+
+	#[inline]
+	pub(super) fn able(&self, ability: &str) -> bool { self.abilities.contains(ability) }
+}

yazi-dds/src/lib.rs

@@ -0,0 +1,34 @@
+#![allow(clippy::option_map_unit_fn)]
+pub mod body;
+mod client;
+mod payload;
+mod pubsub;
+mod sendable;
+mod server;
+mod state;
+
+pub use client::*;
+pub use payload::*;
+pub use pubsub::*;
+pub use sendable::*;
+use server::*;
+pub use state::*;
+
+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);
+
+	// Server
+	CLIENTS.with(Default::default);
+	STATE.with(Default::default);
+
+	// Pubsub
+	LOCAL.with(Default::default);
+	REMOTE.with(Default::default);
+
+	Client::serve(rx);
+}

yazi-dds/src/payload.rs

@@ -0,0 +1,95 @@
+use std::{fmt::Display, io::Write, str::FromStr};
+
+use anyhow::{anyhow, Result};
+use yazi_boot::BOOT;
+use yazi_shared::{emit, event::Cmd, Layer};
+
+use crate::body::Body;
+
+#[derive(Debug)]
+pub struct Payload<'a> {
+	pub receiver: u64,
+	pub severity: u8,
+	pub body:     Body<'a>,
+}
+
+impl<'a> Payload<'a> {
+	#[inline]
+	pub(super) fn new(body: Body<'a>) -> Self { Self { receiver: 0, severity: 0, body } }
+
+	#[inline]
+	pub(super) fn with_receiver(mut self, receiver: u64) -> Self {
+		self.receiver = receiver;
+		self
+	}
+
+	#[inline]
+	pub(super) fn with_severity(mut self, severity: u8) -> Self {
+		self.severity = severity;
+		self
+	}
+
+	pub(super) fn flush(self, force: bool) -> Self {
+		let b = force
+			|| if self.receiver == 0 {
+				BOOT.remote_events.contains(self.body.kind())
+			} else if let Body::Custom(b) = &self.body {
+				BOOT.local_events.contains(&b.kind)
+			} else {
+				false
+			};
+
+		if b {
+			writeln!(std::io::stdout(), "{self}").ok();
+		}
+		self
+	}
+}
+
+impl Payload<'static> {
+	pub(super) fn emit(self) {
+		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> {
+		let mut parts = s.splitn(4, ',');
+
+		let kind = parts.next().ok_or_else(|| anyhow!("empty kind"))?;
+
+		let receiver =
+			parts.next().and_then(|s| s.parse().ok()).ok_or_else(|| anyhow!("invalid receiver"))?;
+
+		let severity =
+			parts.next().and_then(|s| s.parse().ok()).ok_or_else(|| anyhow!("invalid severity"))?;
+
+		let body = parts.next().ok_or_else(|| anyhow!("empty body"))?;
+
+		Ok(Self { receiver, severity, body: Body::from_str(kind, body)? })
+	}
+}
+
+impl Display for Payload<'_> {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		let result = match &self.body {
+			Body::Hi(b) => serde_json::to_string(b),
+			Body::Hey(b) => serde_json::to_string(b),
+			Body::Tabs(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),
+			Body::Bulk(b) => serde_json::to_string(b),
+			Body::Yank(b) => serde_json::to_string(b),
+			Body::Custom(b) => serde_json::to_string(b),
+		};
+
+		if let Ok(s) = result {
+			write!(f, "{},{},{},{s}", self.body.kind(), self.receiver, self.severity)
+		} else {
+			Err(std::fmt::Error)
+		}
+	}
+}

yazi-dds/src/pubsub.rs

@@ -0,0 +1,156 @@
+use std::collections::{HashMap, HashSet};
+
+use mlua::Function;
+use parking_lot::RwLock;
+use yazi_boot::BOOT;
+use yazi_shared::{fs::Url, RoCell};
+
+use crate::{body::{Body, BodyCd, BodyHi, BodyHover, BodyRename, BodyTabs, BodyYank}, Client, Payload, ID, PEERS};
+
+pub static LOCAL: RoCell<RwLock<HashMap<String, HashMap<String, Function<'static>>>>> =
+	RoCell::new();
+
+pub static REMOTE: RoCell<RwLock<HashMap<String, HashMap<String, Function<'static>>>>> =
+	RoCell::new();
+
+macro_rules! sub {
+	($var:ident) => {
+		|plugin: &str, kind: &str, f: Function<'static>| {
+			let mut var = $var.write();
+			let Some(map) = var.get_mut(kind) else {
+				var.insert(kind.to_owned(), HashMap::from_iter([(plugin.to_owned(), f)]));
+				return true;
+			};
+
+			if !map.contains_key(plugin) {
+				map.insert(plugin.to_owned(), f);
+				return true;
+			}
+			false
+		}
+	};
+}
+
+macro_rules! unsub {
+	($var:ident) => {
+		|plugin: &str, kind: &str| {
+			let mut var = $var.write();
+			let Some(map) = var.get_mut(kind) else { return false };
+
+			if map.remove(plugin).is_none() {
+				return false;
+			}
+			if map.is_empty() {
+				var.remove(kind);
+			}
+			true
+		}
+	};
+}
+
+pub struct Pubsub;
+
+impl Pubsub {
+	pub fn sub(plugin: &str, kind: &str, f: Function<'static>) -> bool {
+		sub!(LOCAL)(plugin, kind, f)
+	}
+
+	pub fn sub_remote(plugin: &str, kind: &str, f: Function<'static>) -> bool {
+		sub!(REMOTE)(plugin, kind, f) && Self::pub_from_hi()
+	}
+
+	pub fn unsub(plugin: &str, kind: &str) -> bool { unsub!(LOCAL)(plugin, kind) }
+
+	pub fn unsub_remote(plugin: &str, kind: &str) -> bool {
+		unsub!(REMOTE)(plugin, kind) && Self::pub_from_hi()
+	}
+
+	pub fn pub_(body: Body<'static>) { body.upgrade().with_receiver(*ID).flush(false).emit(); }
+
+	pub fn pub_to(receiver: u64, body: Body<'static>) {
+		if receiver == *ID {
+			return Self::pub_(body);
+		}
+
+		let (kind, peers) = (body.kind(), PEERS.read());
+		if receiver == 0 && peers.values().any(|c| c.able(kind)) {
+			Client::push(body.upgrade());
+		} else if peers.get(&receiver).is_some_and(|c| c.able(kind)) {
+			Client::push(body.upgrade().with_receiver(receiver));
+		}
+	}
+
+	pub fn pub_static(severity: u8, body: Body) {
+		let (kind, peers) = (body.kind(), PEERS.read());
+		if peers.values().any(|c| c.able(kind)) {
+			Client::push(body.upgrade().with_severity(severity));
+		}
+	}
+
+	pub fn pub_from_hi() -> bool {
+		Client::push(Payload::new(
+			BodyHi { id: *ID, abilities: REMOTE.read().keys().cloned().collect() }.into(),
+		));
+		true
+	}
+
+	pub fn pub_from_tabs(tab: usize, urls: &[&Url]) {
+		if LOCAL.read().contains_key("tabs") {
+			Self::pub_(BodyTabs::owned(tab));
+		}
+		if PEERS.read().values().any(|p| p.able("tabs")) {
+			Client::push(BodyTabs::borrowed(tab, urls).upgrade());
+		}
+		if BOOT.local_events.contains("tabs") {
+			BodyTabs::borrowed(tab, urls).upgrade().flush(true);
+		}
+	}
+
+	pub fn pub_from_cd(tab: usize, url: &Url) {
+		if LOCAL.read().contains_key("cd") {
+			Self::pub_(BodyCd::owned(tab));
+		}
+		if PEERS.read().values().any(|p| p.able("cd")) {
+			Client::push(BodyCd::borrowed(tab, url).upgrade());
+		}
+		if BOOT.local_events.contains("cd") {
+			BodyCd::borrowed(tab, url).upgrade().flush(true);
+		}
+	}
+
+	pub fn pub_from_hover(tab: usize, url: Option<&Url>) {
+		if LOCAL.read().contains_key("hover") {
+			Self::pub_(BodyHover::owned(tab));
+		}
+		if PEERS.read().values().any(|p| p.able("hover")) {
+			Client::push(BodyHover::borrowed(tab, url).upgrade());
+		}
+		if BOOT.local_events.contains("hover") {
+			BodyHover::borrowed(tab, url).upgrade().flush(true);
+		}
+	}
+
+	pub fn pub_from_rename(tab: usize, from: &Url, to: &Url) {
+		if LOCAL.read().contains_key("rename") {
+			Self::pub_(BodyRename::owned(tab, from, to));
+		}
+		if PEERS.read().values().any(|p| p.able("rename")) {
+			Client::push(BodyRename::borrowed(tab, from, to).upgrade());
+		}
+		if BOOT.local_events.contains("rename") {
+			BodyRename::borrowed(tab, from, to).upgrade().flush(true);
+		}
+	}
+
+	pub fn pub_from_yank(cut: bool, urls: &HashSet<Url>) {
+		if LOCAL.read().contains_key("yank") {
+			Self::pub_(BodyYank::owned(cut));
+		}
+		if PEERS.read().values().any(|p| p.able("yank")) {
+			Client::push(BodyYank::borrowed(cut, urls).upgrade());
+		}
+		if BOOT.local_events.contains("yank") {
+			BodyYank::borrowed(cut, urls).upgrade().flush(true);
+		}
+	}
+}

yazi-dds/src/sendable.rs

@@ -0,0 +1,132 @@
+use std::collections::HashMap;
+
+use mlua::{ExternalError, IntoLua, Lua, Value, Variadic};
+use serde::{Deserialize, Serialize};
+use yazi_shared::OrderedFloat;
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum ValueSendable {
+	Nil,
+	Boolean(bool),
+	Integer(i64),
+	Number(f64),
+	String(Vec<u8>),
+	Table(HashMap<ValueSendableKey, ValueSendable>),
+}
+
+impl ValueSendable {
+	pub fn try_from_variadic(values: Variadic<Value>) -> mlua::Result<Vec<Self>> {
+		let mut vec = Vec::with_capacity(values.len());
+		for value in values {
+			vec.push(Self::try_from(value)?);
+		}
+		Ok(vec)
+	}
+
+	pub fn into_table_string(self) -> HashMap<String, String> {
+		let Self::Table(table) = self else {
+			return Default::default();
+		};
+
+		let mut map = HashMap::with_capacity(table.len());
+		for pair in table {
+			let (ValueSendableKey::String(k), Self::String(v)) = pair else {
+				continue;
+			};
+			if let (Ok(k), Ok(v)) = (String::from_utf8(k), String::from_utf8(v)) {
+				map.insert(k, v);
+			}
+		}
+		map
+	}
+}
+
+impl<'a> TryFrom<Value<'a>> for ValueSendable {
+	type Error = mlua::Error;
+
+	fn try_from(value: Value) -> Result<Self, Self::Error> {
+		Ok(match value {
+			Value::Nil => Self::Nil,
+			Value::Boolean(b) => Self::Boolean(b),
+			Value::LightUserData(_) => Err("light userdata is not supported".into_lua_err())?,
+			Value::Integer(n) => Self::Integer(n),
+			Value::Number(n) => Self::Number(n),
+			Value::String(s) => Self::String(s.as_bytes().to_vec()),
+			Value::Table(t) => {
+				let mut map = HashMap::with_capacity(t.len().map(|l| l as usize)?);
+				for result in t.pairs::<Value, Value>() {
+					let (k, v) = result?;
+					map.insert(Self::try_from(k)?.try_into()?, v.try_into()?);
+				}
+				Self::Table(map)
+			}
+			Value::Function(_) => Err("function is not supported".into_lua_err())?,
+			Value::Thread(_) => Err("thread is not supported".into_lua_err())?,
+			Value::UserData(_) => Err("userdata is not supported".into_lua_err())?,
+			Value::Error(_) => Err("error is not supported".into_lua_err())?,
+		})
+	}
+}
+
+impl IntoLua<'_> for ValueSendable {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> {
+		match self {
+			Self::Nil => Ok(Value::Nil),
+			Self::Boolean(v) => Ok(Value::Boolean(v)),
+			Self::Integer(v) => Ok(Value::Integer(v)),
+			Self::Number(v) => Ok(Value::Number(v)),
+			Self::String(v) => Ok(Value::String(lua.create_string(v)?)),
+			Self::Table(v) => {
+				let seq_len = v.keys().filter(|&k| !k.is_numeric()).count();
+				let table = lua.create_table_with_capacity(seq_len, v.len() - seq_len)?;
+				for (k, v) in v {
+					table.raw_set(k, v)?;
+				}
+				Ok(Value::Table(table))
+			}
+		}
+	}
+}
+
+#[derive(Debug, Hash, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum ValueSendableKey {
+	Nil,
+	Boolean(bool),
+	Integer(i64),
+	Number(OrderedFloat),
+	String(Vec<u8>),
+}
+
+impl ValueSendableKey {
+	#[inline]
+	fn is_numeric(&self) -> bool { matches!(self, Self::Integer(_) | Self::Number(_)) }
+}
+
+impl TryInto<ValueSendableKey> for ValueSendable {
+	type Error = mlua::Error;
+
+	fn try_into(self) -> Result<ValueSendableKey, Self::Error> {
+		Ok(match self {
+			Self::Nil => ValueSendableKey::Nil,
+			Self::Boolean(v) => ValueSendableKey::Boolean(v),
+			Self::Integer(v) => ValueSendableKey::Integer(v),
+			Self::Number(v) => ValueSendableKey::Number(OrderedFloat::new(v)),
+			Self::String(v) => ValueSendableKey::String(v),
+			Self::Table(_) => Err("table is not supported".into_lua_err())?,
+		})
+	}
+}
+
+impl IntoLua<'_> for ValueSendableKey {
+	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> {
+		match self {
+			Self::Nil => Ok(Value::Nil),
+			Self::Boolean(k) => Ok(Value::Boolean(k)),
+			Self::Integer(k) => Ok(Value::Integer(k)),
+			Self::Number(k) => Ok(Value::Number(k.get())),
+			Self::String(k) => Ok(Value::String(lua.create_string(k)?)),
+		}
+	}
+}

yazi-dds/src/server.rs

@@ -0,0 +1,125 @@
+use std::{collections::HashMap, str::FromStr, time::Duration};
+
+use anyhow::Result;
+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};
+
+pub(super) static CLIENTS: RoCell<RwLock<HashMap<u64, Client>>> = RoCell::new();
+
+pub(super) struct Server;
+
+impl Server {
+	pub(super) async fn make() -> Result<JoinHandle<()>> {
+		CLIENTS.write().clear();
+		let listener = Self::bind().await?;
+
+		Ok(tokio::spawn(async move {
+			while let Ok((stream, _)) = listener.accept().await {
+				let (tx, mut rx) = mpsc::unbounded_channel::<String>();
+				let (reader, mut writer) = tokio::io::split(stream);
+
+				tokio::spawn(async move {
+					let mut id = None;
+					let mut lines = BufReader::new(reader).lines();
+					loop {
+						select! {
+							Some(payload) = rx.recv() => {
+								if writer.write_all(payload.as_bytes()).await.is_err() {
+									break;
+								}
+							}
+							_ = time::sleep(Duration::from_secs(5)) => {
+								if writer.write_u8(b'\n').await.is_err() {
+									break;
+								}
+							}
+							Ok(Some(mut line)) = lines.next_line() => {
+								if line.starts_with("hi,") {
+									Self::handle_hi(line, &mut id, tx.clone());
+									continue;
+								}
+
+								let mut parts = line.splitn(4, ',');
+								let Some(id) = id else { continue };
+								let Some(kind) = parts.next() else { continue };
+								let Some(receiver) = parts.next().and_then(|s| s.parse().ok()) else { continue };
+								let Some(severity) = parts.next().and_then(|s| s.parse::<u8>().ok()) else { continue };
+
+								let clients = CLIENTS.read();
+								let clients: Vec<_> = if receiver == 0 {
+									clients.values().filter(|c| c.id != id && c.able(kind)).collect()
+								} else if let Some(c) = clients.get(&receiver).filter(|c| c.id != id && c.able(kind)) {
+									vec![c]
+								} else {
+									vec![]
+								};
+
+								if clients.is_empty() {
+									continue;
+								}
+
+								if receiver == 0 && severity > 0 {
+									let Some(body) = parts.next() else { continue };
+									STATE.add(format!("{}_{severity}_{kind}", Body::tab(kind, body)), &line);
+								}
+
+								line.push('\n');
+								clients.into_iter().for_each(|c| _ = c.tx.send(line.clone()));
+							}
+							else => break
+						}
+					}
+					Self::handle_bye(id);
+				});
+			}
+		}))
+	}
+
+	#[cfg(unix)]
+	#[inline]
+	async fn bind() -> Result<tokio::net::UnixListener> {
+		tokio::fs::remove_file("/tmp/yazi.sock").await.ok();
+		Ok(tokio::net::UnixListener::bind("/tmp/yazi.sock")?)
+	}
+
+	#[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(Body::Hi(hi)) = Payload::from_str(&s).map(|p| p.body) else { return };
+
+		let mut clients = CLIENTS.write();
+		id.replace(hi.id).and_then(|id| clients.remove(&id));
+
+		if let Some(ref state) = *STATE.read() {
+			state.values().for_each(|s| _ = tx.send(format!("{s}\n")));
+		}
+
+		clients.insert(hi.id, Client { id: hi.id, tx, abilities: hi.abilities });
+		Self::handle_hey(&clients);
+	}
+
+	fn handle_hey(clients: &HashMap<u64, Client>) {
+		let payload = format!(
+			"{}\n",
+			Payload::new(
+				BodyHey { peers: clients.values().map(|c| (c.id, Peer::new(&c.abilities))).collect() }
+					.into()
+			)
+		);
+		clients.values().for_each(|c| _ = c.tx.send(payload.clone()));
+	}
+
+	fn handle_bye(id: Option<u64>) {
+		let mut clients = CLIENTS.write();
+		if id.and_then(|id| clients.remove(&id)).is_some() {
+			Self::handle_hey(&clients);
+		}
+	}
+}

yazi-dds/src/state.rs

@@ -0,0 +1,87 @@
+use std::{collections::HashMap, mem, ops::Deref, sync::atomic::{AtomicU64, Ordering}, time::UNIX_EPOCH};
+
+use anyhow::Result;
+use parking_lot::RwLock;
+use tokio::{fs::{self, File, OpenOptions}, io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter}};
+use yazi_boot::BOOT;
+use yazi_shared::{timestamp_us, RoCell};
+
+use crate::{body::Body, CLIENTS};
+
+pub static STATE: RoCell<State> = RoCell::new();
+
+#[derive(Default)]
+pub struct State {
+	inner: RwLock<Option<HashMap<String, String>>>,
+	last:  AtomicU64,
+}
+
+impl Deref for State {
+	type Target = RwLock<Option<HashMap<String, String>>>;
+
+	fn deref(&self) -> &Self::Target { &self.inner }
+}
+
+impl State {
+	pub fn add(&self, key: String, value: &str) {
+		if let Some(ref mut inner) = *self.inner.write() {
+			inner.insert(key, value.to_owned());
+			self.last.store(timestamp_us(), Ordering::Relaxed);
+		}
+	}
+
+	pub async fn load(&self) -> Result<()> {
+		let mut buf = BufReader::new(File::open(BOOT.state_dir.join(".dds")).await?);
+		let mut line = String::new();
+
+		let mut inner = HashMap::new();
+		while buf.read_line(&mut line).await? > 0 {
+			let mut parts = line.splitn(4, ',');
+			let Some(kind) = parts.next() else { continue };
+			let Some(_) = parts.next() else { continue };
+			let Some(severity) = parts.next().and_then(|s| s.parse::<u8>().ok()) else { continue };
+			let Some(body) = parts.next() else { continue };
+			inner.insert(format!("{}_{severity}_{kind}", Body::tab(kind, body)), mem::take(&mut line));
+		}
+
+		let clients = CLIENTS.read();
+		for payload in inner.values() {
+			clients.values().for_each(|c| _ = c.tx.send(format!("{payload}\n")));
+		}
+
+		self.inner.write().replace(inner);
+		self.last.store(timestamp_us(), Ordering::Relaxed);
+		Ok(())
+	}
+
+	pub async fn drain(&self) -> Result<()> {
+		let Some(inner) = self.inner.write().take() else { return Ok(()) };
+		if self.skip().await.unwrap_or(false) {
+			return Ok(());
+		}
+
+		let mut buf = BufWriter::new(
+			OpenOptions::new()
+				.write(true)
+				.create(true)
+				.truncate(true)
+				.open(BOOT.state_dir.join(".dds"))
+				.await?,
+		);
+
+		let mut state = inner.into_iter().collect::<Vec<_>>();
+		state.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));
+		for (_, v) in state {
+			buf.write_all(v.as_bytes()).await?;
+			buf.write_u8(b'\n').await?;
+		}
+
+		Ok(())
+	}
+
+	async fn skip(&self) -> Result<bool> {
+		let meta = fs::symlink_metadata(BOOT.state_dir.join(".dds")).await?;
+		let modified = meta.modified()?.duration_since(UNIX_EPOCH)?.as_micros();
+		Ok(modified >= self.last.load(Ordering::Relaxed) as u128)
+	}
+}

yazi-fm/Cargo.toml

@@ -13,6 +13,7 @@ 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-scheduler = { path = "../yazi-scheduler", version = "0.2.4" }
@@ -27,7 +28,7 @@ futures       = "0.3.30"
 mlua          = { version = "0.9.6", features = [ "lua54", "vendored" ] }
 ratatui       = "0.26.1"
 syntect       = { version = "5.2.0", default-features = false, features = [ "parsing", "plist-load", "regex-onig" ] }
-tokio         = { version = "1.36.0", features = [ "parking_lot" ] }
+tokio         = { version = "1.36.0", features = [ "full" ] }
 tokio-util    = "0.7.10"
 unicode-width = "0.1.11"
 

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

@@ -0,0 +1,31 @@
+use mlua::IntoLua;
+use tracing::error;
+use yazi_dds::{Payload, LOCAL, REMOTE};
+use yazi_plugin::LUA;
+use yazi_shared::event::Cmd;
+
+use crate::{app::App, lives::Lives};
+
+impl App {
+	pub(crate) fn accept_payload(&mut self, mut cmd: Cmd) {
+		let Some(payload) = cmd.take_data::<Payload>() else {
+			return;
+		};
+
+		let kind = payload.body.kind().to_owned();
+		let map = if payload.receiver == 0 { REMOTE.read() } else { LOCAL.read() };
+		let Some(map) = map.get(&kind).filter(|&m| !m.is_empty()) else {
+			return;
+		};
+
+		_ = Lives::scope(&self.cx, |_| {
+			let body = payload.body.into_lua(&LUA)?;
+			for f in map.values() {
+				if let Err(e) = f.call::<_, ()>(body.clone()) {
+					error!("Failed to call `{kind}` handler: {e}");
+				}
+			}
+			Ok(())
+		});
+	}
+}

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

@@ -1,3 +1,4 @@
+mod accept_payload;
 mod notify;
 mod plugin;
 mod quit;

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

@@ -2,8 +2,8 @@ use std::fmt::Display;
 
 use mlua::TableExt;
 use tracing::warn;
-use yazi_plugin::{OptData, LOADED, LUA};
+use yazi_plugin::{OptData, LOADER, LUA};
-use yazi_shared::{emit, event::Cmd, Layer};
+use yazi_shared::{emit, event::Cmd, Defer, Layer};
 
 use crate::{app::App, lives::Lives};
 
@@ -18,12 +18,12 @@ impl App {
 			return self.cx.tasks.plugin_micro(opt.name, opt.data.args);
 		}
 
-		if LOADED.read().contains_key(&opt.name) {
+		if LOADER.read().contains_key(&opt.name) {
 			return self.plugin_do(opt);
 		}
 
 		tokio::spawn(async move {
-			if LOADED.ensure(&opt.name).await.is_ok() {
+			if LOADER.ensure(&opt.name).await.is_ok() {
 				Self::_plugin_do(opt.name, opt.data);
 			}
 		});
@@ -40,7 +40,10 @@ impl App {
 			Err(e) => return warn!("{e}"),
 		};
 
-		let plugin = match LOADED.load(&opt.name) {
+		LOADER.set_running(Some(&opt.name));
+		let _defer = Defer::new(|| LOADER.set_running(None));
+
+		let plugin = match LOADER.load(&opt.name) {
 			Ok(plugin) => plugin,
 			Err(e) => return warn!("{e}"),
 		};

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::STATE.drain()).ok();
 
 		if !opt.no_cwd_file {
 			self.cwd_to_file();

yazi-fm/src/executor.rs

@@ -34,6 +34,7 @@ impl<'a> Executor<'a> {
 			};
 		}
 
+		on!(accept_payload);
 		on!(notify);
 		on!(plugin);
 		on!(plugin_do);

yazi-fm/src/lives/tabs.rs

@@ -22,7 +22,7 @@ impl Tabs {
 
 	pub(super) fn register(lua: &Lua) -> mlua::Result<()> {
 		lua.register_userdata_type::<Self>(|reg| {
-			reg.add_field_method_get("idx", |_, me| Ok(me.idx + 1));
+			reg.add_field_method_get("idx", |_, me| Ok(me.cursor + 1));
 
 			reg.add_meta_method(MetaMethod::Len, |_, me, ()| Ok(me.len()));
 

yazi-fm/src/main.rs

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

yazi-plugin/Cargo.toml

@@ -12,12 +12,14 @@ repository  = "https://github.com/sxyazi/yazi"
 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-dds     = { path = "../yazi-dds", version = "0.2.4" }
 yazi-proxy   = { path = "../yazi-proxy", version = "0.2.4" }
 yazi-shared  = { path = "../yazi-shared", version = "0.2.4" }
 
 # External dependencies
 ansi-to-tui   = "3.1.0"
 anyhow        = "1.0.81"
+arc-swap      = "1.7.0"
 crossterm     = "0.27.0"
 futures       = "0.3.30"
 md-5          = "0.10.6"
@@ -29,7 +31,7 @@ serde_json    = "1.0.114"
 shell-escape  = "0.1.5"
 shell-words   = "1.1.0"
 syntect       = { version = "5.2.0", default-features = false, features = [ "parsing", "plist-load", "regex-onig" ] }
-tokio         = { version = "1.36.0", features = [ "parking_lot", "rt-multi-thread" ] }
+tokio         = { version = "1.36.0", features = [ "full" ] }
 tokio-util    = "0.7.10"
 tracing       = { version = "0.1.40", features = [ "max_level_debug", "release_max_level_warn" ] }
 unicode-width = "0.1.11"

yazi-plugin/src/bindings/position.rs

@@ -32,8 +32,8 @@ impl<'a> TryFrom<mlua::Table<'a>> for Position {
 	}
 }
 
-impl<'lua> IntoLua<'lua> for Position {
+impl IntoLua<'_> for Position {
-	fn into_lua(self, lua: &'lua mlua::Lua) -> mlua::Result<mlua::Value> {
+	fn into_lua(self, lua: &mlua::Lua) -> mlua::Result<mlua::Value> {
 		lua
 			.create_table_from([
 				(1.into_lua(lua)?, self.origin.to_string().into_lua(lua)?),

yazi-plugin/src/cast.rs

@@ -1,7 +1,4 @@
-use std::collections::HashMap;
+use mlua::AnyUserData;
-
-use mlua::{AnyUserData, ExternalError, IntoLua, Lua, Value, Variadic};
-use yazi_shared::OrderedFloat;
 
 use crate::elements::Renderable;
 
@@ -22,128 +19,3 @@ pub fn cast_to_renderable(ud: AnyUserData) -> Option<Box<dyn Renderable + Send>>
 		None
 	}
 }
-
-#[derive(Debug)]
-pub enum ValueSendable {
-	Nil,
-	Boolean(bool),
-	Integer(i64),
-	Number(f64),
-	String(Vec<u8>),
-	Table(HashMap<ValueSendableKey, ValueSendable>),
-}
-
-impl<'a> TryFrom<Value<'a>> for ValueSendable {
-	type Error = mlua::Error;
-
-	fn try_from(value: Value) -> Result<Self, Self::Error> {
-		Ok(match value {
-			Value::Nil => ValueSendable::Nil,
-			Value::Boolean(b) => ValueSendable::Boolean(b),
-			Value::LightUserData(_) => Err("light userdata is not supported".into_lua_err())?,
-			Value::Integer(n) => ValueSendable::Integer(n),
-			Value::Number(n) => ValueSendable::Number(n),
-			Value::String(s) => ValueSendable::String(s.as_bytes().to_vec()),
-			Value::Table(t) => {
-				let mut map = HashMap::with_capacity(t.len().map(|l| l as usize)?);
-				for result in t.pairs::<Value, Value>() {
-					let (k, v) = result?;
-					map.insert(ValueSendable::try_from(k)?.try_into()?, v.try_into()?);
-				}
-				ValueSendable::Table(map)
-			}
-			Value::Function(_) => Err("function is not supported".into_lua_err())?,
-			Value::Thread(_) => Err("thread is not supported".into_lua_err())?,
-			Value::UserData(_) => Err("userdata is not supported".into_lua_err())?,
-			Value::Error(_) => Err("error is not supported".into_lua_err())?,
-		})
-	}
-}
-
-impl<'lua> IntoLua<'lua> for ValueSendable {
-	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> {
-		match self {
-			ValueSendable::Nil => Ok(Value::Nil),
-			ValueSendable::Boolean(b) => Ok(Value::Boolean(b)),
-			ValueSendable::Integer(n) => Ok(Value::Integer(n)),
-			ValueSendable::Number(n) => Ok(Value::Number(n)),
-			ValueSendable::String(s) => Ok(Value::String(lua.create_string(s)?)),
-			ValueSendable::Table(t) => {
-				let seq_len = t.keys().filter(|&k| !k.is_numeric()).count();
-				let table = lua.create_table_with_capacity(seq_len, t.len() - seq_len)?;
-				for (k, v) in t {
-					table.raw_set(k, v)?;
-				}
-				Ok(Value::Table(table))
-			}
-		}
-	}
-}
-
-impl ValueSendable {
-	pub fn try_from_variadic(values: Variadic<Value>) -> mlua::Result<Vec<ValueSendable>> {
-		let mut vec = Vec::with_capacity(values.len());
-		for value in values {
-			vec.push(ValueSendable::try_from(value)?);
-		}
-		Ok(vec)
-	}
-
-	pub fn into_table_string(self) -> HashMap<String, String> {
-		let ValueSendable::Table(table) = self else {
-			return Default::default();
-		};
-
-		let mut map = HashMap::with_capacity(table.len());
-		for pair in table {
-			let (ValueSendableKey::String(k), ValueSendable::String(v)) = pair else {
-				continue;
-			};
-			if let (Ok(k), Ok(v)) = (String::from_utf8(k), String::from_utf8(v)) {
-				map.insert(k, v);
-			}
-		}
-		map
-	}
-}
-
-#[derive(Debug, Hash, PartialEq, Eq)]
-pub enum ValueSendableKey {
-	Nil,
-	Boolean(bool),
-	Integer(i64),
-	Number(OrderedFloat),
-	String(Vec<u8>),
-}
-
-impl ValueSendableKey {
-	#[inline]
-	fn is_numeric(&self) -> bool { matches!(self, Self::Integer(_) | Self::Number(_)) }
-}
-
-impl TryInto<ValueSendableKey> for ValueSendable {
-	type Error = mlua::Error;
-
-	fn try_into(self) -> Result<ValueSendableKey, Self::Error> {
-		Ok(match self {
-			ValueSendable::Nil => ValueSendableKey::Nil,
-			ValueSendable::Boolean(b) => ValueSendableKey::Boolean(b),
-			ValueSendable::Integer(n) => ValueSendableKey::Integer(n),
-			ValueSendable::Number(n) => ValueSendableKey::Number(OrderedFloat::new(n)),
-			ValueSendable::String(s) => ValueSendableKey::String(s),
-			ValueSendable::Table(_) => Err("table is not supported".into_lua_err())?,
-		})
-	}
-}
-
-impl<'lua> IntoLua<'lua> for ValueSendableKey {
-	fn into_lua(self, lua: &Lua) -> mlua::Result<Value> {
-		match self {
-			ValueSendableKey::Nil => Ok(Value::Nil),
-			ValueSendableKey::Boolean(b) => Ok(Value::Boolean(b)),
-			ValueSendableKey::Integer(n) => Ok(Value::Integer(n)),
-			ValueSendableKey::Number(n) => Ok(Value::Number(n.get())),
-			ValueSendableKey::String(s) => Ok(Value::String(lua.create_string(s)?)),
-		}
-	}
-}

yazi-plugin/src/elements/elements.rs

@@ -1,9 +1,9 @@
-use mlua::{AnyUserData, Lua, Table};
+use mlua::{AnyUserData, Lua};
 
 use crate::cast_to_renderable;
 
 pub fn pour(lua: &Lua) -> mlua::Result<()> {
-	let ui: Table = lua.create_table()?;
+	let ui = lua.create_table()?;
 
 	// Register
 	super::Padding::register(lua)?;

yazi-plugin/src/isolate/entry.rs

@@ -1,15 +1,16 @@
 use mlua::{ExternalError, ExternalResult, Table, TableExt};
 use tokio::runtime::Handle;
+use yazi_dds::ValueSendable;
 
 use super::slim_lua;
-use crate::{ValueSendable, LOADED};
+use crate::LOADER;
 
 pub async fn entry(name: String, args: Vec<ValueSendable>) -> mlua::Result<()> {
-	LOADED.ensure(&name).await.into_lua_err()?;
+	LOADER.ensure(&name).await.into_lua_err()?;
 
 	tokio::task::spawn_blocking(move || {
 		let lua = slim_lua(&name)?;
-		let plugin: Table = if let Some(b) = LOADED.read().get(&name) {
+		let plugin: Table = if let Some(b) = LOADER.read().get(&name) {
 			lua.load(b).call(())?
 		} else {
 			return Err("unloaded plugin".into_lua_err());

yazi-plugin/src/isolate/peek.rs

@@ -6,7 +6,7 @@ use yazi_config::LAYOUT;
 use yazi_shared::{emit, event::Cmd, Layer};
 
 use super::slim_lua;
-use crate::{bindings::{Cast, File, Window}, elements::Rect, OptData, LOADED, LUA};
+use crate::{bindings::{Cast, File, Window}, elements::Rect, OptData, LOADER, LUA};
 
 pub fn peek(cmd: &Cmd, file: yazi_shared::fs::File, skip: usize) -> CancellationToken {
 	let ct = CancellationToken::new();
@@ -15,7 +15,7 @@ pub fn peek(cmd: &Cmd, file: yazi_shared::fs::File, skip: usize) -> Cancellation
 	let (ct1, ct2) = (ct.clone(), ct.clone());
 	tokio::task::spawn_blocking(move || {
 		let future = async {
-			LOADED.ensure(&name).await.into_lua_err()?;
+			LOADER.ensure(&name).await.into_lua_err()?;
 
 			let lua = slim_lua(&name)?;
 			lua.set_hook(
@@ -25,7 +25,7 @@ pub fn peek(cmd: &Cmd, file: yazi_shared::fs::File, skip: usize) -> Cancellation
 				},
 			);
 
-			let plugin: Table = if let Some(b) = LOADED.read().get(&name) {
+			let plugin: Table = if let Some(b) = LOADER.read().get(&name) {
 				lua.load(b).call(())?
 			} else {
 				return Err("unloaded plugin".into_lua_err());

yazi-plugin/src/isolate/preload.rs

@@ -3,19 +3,19 @@ use tokio::runtime::Handle;
 use yazi_config::LAYOUT;
 
 use super::slim_lua;
-use crate::{bindings::{Cast, File}, elements::Rect, LOADED};
+use crate::{bindings::{Cast, File}, elements::Rect, LOADER};
 
 pub async fn preload(
 	name: &str,
 	files: Vec<yazi_shared::fs::File>,
 	multi: bool,
 ) -> mlua::Result<u8> {
-	LOADED.ensure(name).await.into_lua_err()?;
+	LOADER.ensure(name).await.into_lua_err()?;
 
 	let name = name.to_owned();
 	tokio::task::spawn_blocking(move || {
 		let lua = slim_lua(&name)?;
-		let plugin: Table = if let Some(b) = LOADED.read().get(&name) {
+		let plugin: Table = if let Some(b) = LOADER.read().get(&name) {
 			lua.load(b).call(())?
 		} else {
 			return Err("unloaded plugin".into_lua_err());

yazi-plugin/src/lib.rs

@@ -11,6 +11,7 @@ mod loader;
 mod lua;
 mod opt;
 pub mod process;
+pub mod pubsub;
 pub mod url;
 pub mod utils;
 

yazi-plugin/src/loader.rs

@@ -1,4 +1,4 @@
-use std::{borrow::Cow, collections::HashMap, ops::Deref};
+use std::{borrow::Cow, collections::HashMap, ops::Deref, sync::Arc};
 
 use anyhow::{bail, Result};
 use mlua::{ExternalError, Table};
@@ -9,19 +9,24 @@ use yazi_shared::RoCell;
 
 use crate::LUA;
 
-pub static LOADED: RoCell<Loader> = RoCell::new();
+pub static LOADER: RoCell<Loader> = RoCell::new();
+
+pub(super) static RUNNING: RoCell<arc_swap::ArcSwapOption<String>> = RoCell::new();
 
 #[derive(Default)]
 pub struct Loader {
-	loaded: RwLock<HashMap<String, Vec<u8>>>,
+	cache: RwLock<HashMap<String, Vec<u8>>>,
 }
 
 impl Loader {
 	#[inline]
-	pub(super) fn init() { LOADED.with(Default::default); }
+	pub(super) fn init() {
+		LOADER.with(Default::default);
+		RUNNING.with(Default::default);
+	}
 
 	pub async fn ensure(&self, name: &str) -> Result<()> {
-		if self.loaded.read().contains_key(name) {
+		if self.cache.read().contains_key(name) {
 			return Ok(());
 		}
 
@@ -42,7 +47,7 @@ impl Loader {
 			}))
 		})?;
 
-		self.loaded.write().insert(name.to_owned(), b.into_owned());
+		self.cache.write().insert(name.to_owned(), b.into_owned());
 		Ok(())
 	}
 
@@ -64,11 +69,19 @@ impl Loader {
 		loaded.raw_set(name, t.clone())?;
 		Ok(t)
 	}
+
+	pub fn set_running(&self, name: Option<&str>) {
+		if let Some(s) = name {
+			RUNNING.store(Some(Arc::new(s.to_owned())));
+		} else {
+			RUNNING.store(None);
+		}
+	}
 }
 
 impl Deref for Loader {
 	type Target = RwLock<HashMap<String, Vec<u8>>>;
 
 	#[inline]
-	fn deref(&self) -> &Self::Target { &self.loaded }
+	fn deref(&self) -> &Self::Target { &self.cache }
 }

yazi-plugin/src/lua.rs

@@ -6,13 +6,13 @@ use yazi_shared::RoCell;
 pub static LUA: RoCell<Lua> = RoCell::new();
 
 pub(super) fn init_lua() {
-	let lua = Lua::new();
+	LUA.init(Lua::new());
-	stage_1(&lua).expect("failed to initialize Lua");
+
-	stage_2(&lua);
+	stage_1(&LUA).expect("failed to initialize Lua");
-	LUA.init(lua);
+	stage_2(&LUA);
 }
 
-fn stage_1(lua: &Lua) -> Result<()> {
+fn stage_1(lua: &'static Lua) -> Result<()> {
 	crate::Loader::init();
 	crate::Config::new(lua).install_boot()?.install_manager()?.install_theme()?;
 	crate::utils::init();
@@ -24,6 +24,7 @@ fn stage_1(lua: &Lua) -> Result<()> {
 	crate::bindings::File::register(lua)?;
 	crate::bindings::Icon::register(lua)?;
 	crate::elements::pour(lua)?;
+	crate::pubsub::Pubsub::install(lua)?;
 	crate::url::pour(lua)?;
 
 	// Components
@@ -40,7 +41,7 @@ fn stage_1(lua: &Lua) -> Result<()> {
 	Ok(())
 }
 
-fn stage_2(lua: &Lua) {
+fn stage_2(lua: &'static Lua) {
 	lua.load(include_str!("../preset/setup.lua")).exec().unwrap();
 
 	if let Ok(b) = std::fs::read(BOOT.config_dir.join("init.lua")) {

yazi-plugin/src/opt.rs

@@ -1,9 +1,8 @@
 use anyhow::bail;
 use mlua::{Lua, Table};
+use yazi_dds::ValueSendable;
 use yazi_shared::event::Cmd;
 
-use crate::ValueSendable;
-
 pub struct Opt {
 	pub name: String,
 	pub sync: bool,

yazi-plugin/src/pubsub/mod.rs

@@ -0,0 +1,5 @@
+#![allow(clippy::module_inception)]
+
+mod pubsub;
+
+pub use pubsub::*;

yazi-plugin/src/pubsub/pubsub.rs

@@ -0,0 +1,93 @@
+use mlua::{ExternalResult, Function, Lua, Value};
+use yazi_dds::body::Body;
+
+use crate::RUNNING;
+
+pub struct Pubsub;
+
+impl Pubsub {
+	pub fn install(lua: &'static Lua) -> mlua::Result<()> {
+		let ps = lua.create_table()?;
+
+		ps.raw_set(
+			"pub",
+			lua.create_function(|_, (kind, value): (mlua::String, Value)| {
+				yazi_dds::Pubsub::pub_(Body::from_lua(kind.to_str()?, value).into_lua_err()?);
+				Ok(())
+			})?,
+		)?;
+
+		ps.raw_set(
+			"pub_to",
+			lua.create_function(|_, (receiver, kind, value): (u64, mlua::String, Value)| {
+				yazi_dds::Pubsub::pub_to(receiver, Body::from_lua(kind.to_str()?, value).into_lua_err()?);
+				Ok(())
+			})?,
+		)?;
+
+		ps.raw_set(
+			"pub_static",
+			lua.create_function(|_, (severity, kind, value): (u8, mlua::String, Value)| {
+				if severity < 1 {
+					return Err("Severity must be at least 1").into_lua_err();
+				}
+
+				yazi_dds::Pubsub::pub_static(
+					severity,
+					Body::from_lua(kind.to_str()?, value).into_lua_err()?,
+				);
+				Ok(())
+			})?,
+		)?;
+
+		ps.raw_set(
+			"sub",
+			lua.create_function(|_, (kind, f): (mlua::String, Function)| {
+				let Some(name) = &*RUNNING.load() else {
+					return Err("`sub()` must be called in a sync plugin").into_lua_err();
+				};
+				if !yazi_dds::Pubsub::sub(name, kind.to_str()?, f) {
+					return Err("`sub()` called twice").into_lua_err();
+				}
+				Ok(())
+			})?,
+		)?;
+
+		ps.raw_set(
+			"sub_remote",
+			lua.create_function(|_, (kind, f): (mlua::String, Function)| {
+				let Some(name) = &*RUNNING.load() else {
+					return Err("`sub_remote()` must be called in a sync plugin").into_lua_err();
+				};
+				if !yazi_dds::Pubsub::sub_remote(name, kind.to_str()?, f) {
+					return Err("`sub_remote()` called twice").into_lua_err();
+				}
+				Ok(())
+			})?,
+		)?;
+
+		ps.raw_set(
+			"unsub",
+			lua.create_function(|_, kind: mlua::String| {
+				if let Some(name) = &*RUNNING.load() {
+					Ok(yazi_dds::Pubsub::unsub(name, kind.to_str()?))
+				} else {
+					Err("`unsub()` must be called in a sync plugin").into_lua_err()
+				}
+			})?,
+		)?;
+
+		ps.raw_set(
+			"unsub_remote",
+			lua.create_function(|_, kind: mlua::String| {
+				if let Some(name) = &*RUNNING.load() {
+					Ok(yazi_dds::Pubsub::unsub_remote(name, kind.to_str()?))
+				} else {
+					Err("`unsub_remote()` must be called in a sync plugin").into_lua_err()
+				}
+			})?,
+		)?;
+
+		lua.globals().raw_set("ps", ps)
+	}
+}

yazi-plugin/src/utils/call.rs

@@ -1,10 +1,10 @@
 use std::collections::HashMap;
 
 use mlua::{ExternalError, Lua, Table, Value};
+use yazi_dds::ValueSendable;
 use yazi_shared::{emit, event::Cmd, render, Layer};
 
 use super::Utils;
-use crate::ValueSendable;
 
 impl Utils {
 	fn parse_args(t: Table) -> mlua::Result<(Vec<String>, HashMap<String, String>)> {

yazi-plugin/src/utils/plugin.rs

@@ -1,9 +1,10 @@
 use mlua::{ExternalError, Function, IntoLua, Lua, Table, Value, Variadic};
 use tokio::sync::oneshot;
+use yazi_dds::ValueSendable;
 use yazi_shared::{emit, event::Cmd, Layer};
 
 use super::Utils;
-use crate::{OptData, ValueSendable};
+use crate::OptData;
 
 impl Utils {
 	pub(super) fn plugin(lua: &Lua, ya: &Table) -> mlua::Result<()> {

yazi-plugin/src/utils/utils.rs

@@ -1,12 +1,13 @@
 #[cfg(unix)]
 pub(super) static USERS_CACHE: yazi_shared::RoCell<uzers::UsersCache> = yazi_shared::RoCell::new();
 
+#[cfg(unix)]
 pub(super) static HOSTNAME_CACHE: std::sync::OnceLock<Option<String>> = std::sync::OnceLock::new();
 
 pub(super) struct Utils;
 
 pub fn install(lua: &mlua::Lua) -> mlua::Result<()> {
-	let ya: mlua::Table = lua.create_table()?;
+	let ya = lua.create_table()?;
 
 	Utils::app(lua, &ya)?;
 	Utils::cache(lua, &ya)?;

yazi-proxy/Cargo.toml

@@ -15,4 +15,4 @@ yazi-shared = { path = "../yazi-shared", version = "0.2.4" }
 # External dependencies
 anyhow = "1.0.81"
 mlua   = { version = "0.9.6", features = [ "lua54", "vendored" ] }
-tokio  = { version = "1.36.0", features = [ "parking_lot" ] }
+tokio  = { version = "1.36.0", features = [ "full" ] }

yazi-proxy/src/app.rs

@@ -20,6 +20,7 @@ impl AppProxy {
 		emit!(Call(Cmd::new("resume"), Layer::App));
 	}
 
+	#[inline]
 	pub fn notify(opt: NotifyOpt) {
 		emit!(Call(Cmd::new("notify").with_data(opt), Layer::App));
 	}

yazi-scheduler/Cargo.toml

@@ -11,6 +11,7 @@ repository  = "https://github.com/sxyazi/yazi"
 [dependencies]
 yazi-adaptor = { path = "../yazi-adaptor", version = "0.2.4" }
 yazi-config  = { path = "../yazi-config", 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" }
@@ -23,7 +24,7 @@ crossterm              = "0.27.0"
 futures                = "0.3.30"
 parking_lot            = "0.12.1"
 regex                  = "1.10.3"
-tokio                  = { version = "1.36.0", features = [ "parking_lot", "rt-multi-thread" ] }
+tokio                  = { version = "1.36.0", features = [ "full" ] }
 tokio-stream           = "0.1.15"
 
 # Logging

yazi-scheduler/src/plugin/op.rs

@@ -1,4 +1,4 @@
-use yazi_plugin::ValueSendable;
+use yazi_dds::ValueSendable;
 
 #[derive(Debug)]
 pub enum PluginOp {

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_plugin::ValueSendable;
+use yazi_dds::ValueSendable;
 use yazi_shared::{fs::{unique_path, Url}, Throttle};
 
 use super::{Ongoing, TaskProg, TaskStage};

yazi-shared/Cargo.toml

@@ -19,7 +19,7 @@ percent-encoding = "2.3.1"
 ratatui          = "0.26.1"
 regex            = "1.10.3"
 serde            = "1.0.197"
-tokio            = { version = "1.36.0", features = [ "parking_lot", "macros", "rt-multi-thread", "sync", "time", "fs" ] }
+tokio            = { version = "1.36.0", features = [ "full" ] }
 
 [target."cfg(unix)".dependencies]
 libc = "0.2.153"

yazi-shared/src/condition.rs

@@ -44,6 +44,7 @@ impl ConditionOp {
 	}
 }
 
+#[derive(Debug)]
 pub struct Condition {
 	ops: Vec<ConditionOp>,
 }
@@ -53,11 +54,11 @@ impl FromStr for Condition {
 
 	fn from_str(expr: &str) -> Result<Self, Self::Err> {
 		let cond = Self::build(expr);
-		if cond.eval(|_| true).is_some() {
+		if cond.eval(|_| true).is_none() {
-			Ok(cond)
+			bail!("Invalid condition: {expr}");
-		} else {
-			bail!("Invalid condition: {}", expr);
 		}
+
+		Ok(cond)
 	}
 }
 

yazi-shared/src/fs/url.rs

@@ -1,6 +1,7 @@
 use std::{ffi::{OsStr, OsString}, fmt::{Debug, Display, Formatter}, ops::{Deref, DerefMut}, path::{Path, PathBuf}};
 
 use percent_encoding::{percent_decode_str, percent_encode, AsciiSet, CONTROLS};
+use serde::{Deserialize, Serialize};
 
 const ENCODE_SET: &AsciiSet = &CONTROLS.add(b'#');
 
@@ -212,3 +213,19 @@ impl From<&str> for UrlScheme {
 		}
 	}
 }
+
+impl Serialize for Url {
+	fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+		serializer.collect_str(self)
+	}
+}
+
+impl<'de> Deserialize<'de> for Url {
+	fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+	where
+		D: serde::Deserializer<'de>,
+	{
+		let s = String::deserialize(deserializer)?;
+		Ok(Url::from(s))
+	}
+}

yazi-shared/src/number.rs

@@ -1,6 +1,9 @@
 use std::hash::{Hash, Hasher};
 
-#[derive(Clone, Copy, Debug)]
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+#[serde(transparent)]
 pub struct OrderedFloat(f64);
 
 impl OrderedFloat {