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.

yazi-boot/src/boot.rs

@@ -127,6 +127,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,7 +9,8 @@ homepage    = "https://yazi-rs.github.io"
 repository  = "https://github.com/sxyazi/yazi"
 
 [dependencies]
-yazi-dds = { path = "../yazi-dds", version = "0.2.5" }
+yazi-boot = { path = "../yazi-boot", version = "0.2.5" }
+yazi-dds  = { path = "../yazi-dds", version = "0.2.5" }
 
 # External dependencies
 anyhow     = "1.0.82"

yazi-cli/src/args.rs

@@ -17,6 +17,10 @@ pub(super) enum Command {
 	Pub(CommandPub),
 	/// Publish a static message to all remote instances.
 	PubStatic(CommandPubStatic),
+	/// Subscribe to messages from remote instance(s).
+	Sub(CommandSub),
+	/// Subscribe to messages from all remote instance(s).
+	SubStatic(CommandSubStatic),
 }
 
 #[derive(clap::Args)]
@@ -76,3 +80,21 @@ impl CommandPubStatic {
 		}
 	}
 }
+
+#[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,7 +1,10 @@
 mod args;
 
+use std::collections::HashSet;
+
 use args::*;
 use clap::Parser;
+use yazi_dds::dds_peer::DDSPeer;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -22,6 +25,26 @@ async fn main() -> anyhow::Result<()> {
 				std::process::exit(1);
 			}
 		}
+
+		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)?;
 
@@ -122,8 +171,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),
+		}
+	}
+}