feat: support ya sub subcommand for the Ya CLI

yazi-cli/src/args.rs

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

yazi-cli/src/main.rs

@@ -1,6 +1,8 @@
 mod args;
 mod package;
 
+use std::collections::HashSet;
+
 use args::*;
 use clap::Parser;
 
@@ -46,6 +48,15 @@ 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_dds::Client::echo_events_to_stdout(kinds).await?;
+
+			tokio::signal::ctrl_c().await?;
+		}
 	}
 
 	Ok(())

yazi-dds/src/body/hi.rs

@@ -5,8 +5,10 @@ use serde::{Deserialize, Serialize};
 
 use super::Body;
 
+/// The handshake message
 #[derive(Debug, Serialize, Deserialize)]
 pub struct BodyHi<'a> {
+	/// Specifies the kinds of events that the client can handle
 	pub abilities: HashSet<Cow<'a, String>>,
 	pub version:   String,
 }

yazi-dds/src/client.rs

@@ -27,6 +27,7 @@ pub struct Peer {
 }
 
 impl Client {
+	/// Connect to an existing server or start a new one.
 	pub(super) fn serve() {
 		let mut rx = QUEUE_RX.drop();
 		while rx.try_recv().is_ok() {}
@@ -52,7 +53,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 +63,50 @@ impl Client {
 		});
 	}
 
+	/// Connect to an existing server and listen in on the messages that are being
+	/// sent by other yazi instances.
+	/// If no server is running, fail right away.
+	/// If a server is closed, attempt to reconnect forever.
+	pub async fn echo_events_to_stdout(kinds: HashSet<String>) -> Result<()> {
+		let mut lines = Self::connect_listener(&kinds).await?;
+
+		loop {
+			match lines.next_line().await {
+				Ok(Some(s)) => {
+					let kind = s.split(',').next();
+					if matches!(kind, Some(kind) if kinds.contains(kind)) {
+						println!("{}", s);
+					}
+				}
+				Ok(None) => loop {
+					println!("Connection closed");
+					match Self::connect_listener(&kinds).await {
+						Ok(new_lines) => {
+							lines = new_lines;
+							break;
+						}
+						Err(_) => {
+							time::sleep(time::Duration::from_secs(1)).await;
+						}
+					};
+				},
+				Err(e) => {
+					// could not establish initial connection
+					return Err(e.into());
+				}
+			}
+		}
+	}
+
+	async fn connect_listener(kinds: &HashSet<String>) -> Result<ClientReader> {
+		let (lines, mut writer) = Stream::connect().await?;
+		let hi = Payload::new(BodyHi::borrowed(kinds.iter().collect()));
+		writer.write_all(format!("{}\n", hi).as_bytes()).await?;
+		writer.flush().await?;
+		Ok(lines)
+	}
+
+	/// Connect to an existing server to send a single message.
 	pub async fn shot(kind: &str, receiver: u64, severity: Option<u16>, body: &str) -> Result<()> {
 		Body::validate(kind)?;
 
@@ -136,8 +181,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;
 		}