feat(cli): allow printing selected DDS messages to the console

The motivations for this change are:
- easy debugging of DDS messages for yazi and plugin developers
- allow external applications to monitor yazi events.

In neovim specifically, there is a limitation that the stdout and stderr
streams cannot be monitored when displaying an embedded terminal
application. A second yazi instance could theoretically be started, but
the ui does currently not work when there is no screen to draw on.

Cargo.lock

@@ -2742,6 +2742,7 @@ dependencies = [
  "tokio",
  "toml_edit",
  "vergen",
+ "yazi-boot",
  "yazi-dds",
  "yazi-shared",
 ]

yazi-boot/src/boot.rs

@@ -155,6 +155,27 @@ impl Boot {
 			);
 		}
 	}
+
+	/// Allows for initialization without forcing the reading of command line
+	/// arguments.
+	pub fn init_with(local_events: HashSet<String>, remote_events: HashSet<String>) -> Boot {
+		let config_dir = Xdg::config_dir();
+
+		let (cwd, file) = Self::parse_entry(None);
+
+		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(),
+		}
+	}
 }
 
 impl Default for Boot {

yazi-cli/Cargo.toml

@@ -9,6 +9,7 @@ homepage    = "https://yazi-rs.github.io"
 repository  = "https://github.com/sxyazi/yazi"
 
 [dependencies]
+yazi-boot   = { path = "../yazi-boot", version = "0.2.5" }
 yazi-dds    = { path = "../yazi-dds", version = "0.2.5" }
 yazi-shared = { path = "../yazi-shared", version = "0.2.5" }
 

yazi-cli/src/args.rs

@@ -22,6 +22,10 @@ pub(super) enum Command {
 	PubStatic(CommandPubStatic),
 	/// Manage packages.
 	Pack(CommandPack),
+	/// Subscribe to messages from remote instance(s).
+	Sub(CommandSub),
+	/// Subscribe to messages from all remote instance(s).
+	SubStatic(CommandSubStatic),
 }
 
 #[derive(clap::Args)]
@@ -109,3 +113,21 @@ pub(super) struct CommandPack {
 	#[arg(short = 'u', long)]
 	pub(super) upgrade: bool,
 }
+
+#[derive(clap::Args)]
+pub(super) struct CommandSub {
+	/// The sender ID whose messages we want to monitor.
+	#[arg(index = 1)]
+	pub(super) sender: u64,
+
+	/// The kind of messages we are interested in.
+	#[arg(index = 2)]
+	pub(super) kinds: String,
+}
+
+#[derive(clap::Args)]
+pub(super) struct CommandSubStatic {
+	/// The kind of messages we are interested in.
+	#[arg(index = 1)]
+	pub(super) kinds: String,
+}

yazi-cli/src/main.rs

@@ -1,8 +1,11 @@
 mod args;
 mod package;
 
+use std::collections::HashSet;
+
 use args::*;
 use clap::Parser;
+use yazi_dds::dds_peer::DDSPeer;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -46,6 +49,26 @@ async fn main() -> anyhow::Result<()> {
 				package::Package::add_to_config(repo).await?;
 			}
 		}
+
+		Command::Sub(cmd) => {
+			yazi_dds::init();
+			let kinds = cmd.kinds.split(',').map(|s| s.to_owned()).collect::<HashSet<_>>();
+
+			yazi_boot::BOOT.init(yazi_boot::Boot::init_with(kinds.clone(), kinds.clone()));
+			yazi_dds::Client::echo_events_to_stdout(DDSPeer::from(cmd.sender), kinds);
+
+			tokio::signal::ctrl_c().await?;
+		}
+
+		Command::SubStatic(cmd) => {
+			yazi_dds::init();
+			let kinds = cmd.kinds.split(',').map(|s| s.to_owned()).collect::<HashSet<_>>();
+
+			yazi_boot::BOOT.init(yazi_boot::Boot::init_with(kinds.clone(), kinds.clone()));
+			yazi_dds::Client::echo_events_to_stdout(DDSPeer::All, kinds);
+
+			tokio::signal::ctrl_c().await?;
+		}
 	}
 
 	Ok(())

yazi-dds/src/client.rs

@@ -6,7 +6,9 @@ use serde::{Deserialize, Serialize};
 use tokio::{io::AsyncWriteExt, select, sync::mpsc, task::JoinHandle, time};
 use yazi_shared::RoCell;
 
-use crate::{body::{Body, BodyBye, BodyHi}, ClientReader, ClientWriter, Payload, Pubsub, Server, Stream};
+use crate::{body::{Body, BodyBye, BodyHi}, dds_peer::DDSPeer, ClientReader, ClientWriter, Payload, Pubsub, Server, Stream};
+
+pub mod dds_peer;
 
 pub(super) static ID: RoCell<u64> = RoCell::new();
 pub(super) static PEERS: RoCell<RwLock<HashMap<u64, Peer>>> = RoCell::new();
@@ -52,7 +54,7 @@ impl Client {
 						if line.is_empty() {
 							continue;
 						} else if line.starts_with("hey,") {
-							Self::handle_hey(line);
+							Self::handle_hey(&line);
 						} else {
 							Payload::from_str(&line).map(|p| p.emit()).ok();
 						}
@@ -62,6 +64,53 @@ impl Client {
 		});
 	}
 
+	pub fn echo_events_to_stdout(sender: DDSPeer, kinds: HashSet<String>) {
+		let mut rx = QUEUE_RX.drop();
+		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.is_empty() {
+							continue;
+						}
+
+						let payload = Payload::from_str(&line).unwrap();
+						if line.starts_with("hey,") {
+							Self::handle_hey(&line);
+							if sender.matches(payload.sender) {
+								println!("{}", line);
+							}
+						} else {
+							if ! sender.matches(payload.sender) {
+								continue;
+							}
+
+							if kinds.contains(payload.body.kind()) {
+								println!("{}", &line);
+							}
+						}
+					}
+				}
+			}
+		});
+	}
+
 	pub async fn shot(kind: &str, receiver: u64, severity: Option<u16>, body: &str) -> Result<()> {
 		Body::validate(kind)?;
 
@@ -136,8 +185,8 @@ impl Client {
 		Self::connect(server).await
 	}
 
-	fn handle_hey(s: String) {
-		if let Ok(Body::Hey(mut hey)) = Payload::from_str(&s).map(|p| p.body) {
+	fn handle_hey(s: &str) {
+		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;
 		}

yazi-dds/src/client/dds_peer.rs

@@ -0,0 +1,25 @@
+/// The id of a peer in the DDS system.
+#[derive(Debug, PartialEq)]
+pub enum DDSPeer {
+	/// Internally, `0` is used to represent all peers.
+	All,
+	One(u64),
+}
+
+impl DDSPeer {
+	pub fn matches(&self, peer_id: u64) -> bool {
+		match self {
+			Self::All => true,
+			Self::One(id) => *id == peer_id,
+		}
+	}
+}
+
+impl From<u64> for DDSPeer {
+	fn from(value: u64) -> Self {
+		match value {
+			0 => Self::All,
+			_ => Self::One(value),
+		}
+	}
+}