refactor: simplify listening implementation

- don't start a new server. Instead, crash if the server is not running.
- remove support for listening for a specific peer. The protocol doesn't
  support it and it cannot be made to work.
- if the existing yazi closes, this cannot be detected, and the
  connection is never reestablished. Maybe this can be made to work in a
  future commit.

yazi-cli/src/args.rs

@@ -22,8 +22,6 @@ 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),
 }

yazi-cli/src/main.rs

@@ -5,7 +5,6 @@ use std::collections::HashSet;
 
 use args::*;
 use clap::Parser;
-use yazi_dds::dds_peer::DDSPeer;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
@@ -50,22 +49,11 @@ async fn main() -> anyhow::Result<()> {
 			}
 		}
 
-		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(kinds).await?;
-			yazi_dds::Client::echo_events_to_stdout(DDSPeer::All, kinds);
 
 			tokio::signal::ctrl_c().await?;
 		}

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

@@ -6,9 +6,7 @@ use serde::{Deserialize, Serialize};
 use tokio::{io::AsyncWriteExt, select, sync::mpsc, task::JoinHandle, time};
 use yazi_shared::RoCell;
 
-use crate::{body::{Body, BodyBye, BodyHi}, dds_peer::DDSPeer, ClientReader, ClientWriter, Payload, Pubsub, Server, Stream};
+use crate::{body::{Body, BodyBye, BodyHi}, 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();
@@ -29,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() {}
@@ -64,57 +63,26 @@ impl Client {
 		});
 	}
 
-	pub fn echo_events_to_stdout(sender: DDSPeer, kinds: HashSet<String>) {
+	/// Connect to an existing server and listen in on the messages that are being
-		let mut rx = QUEUE_RX.drop();
+	/// sent by other yazi instances.
-		while rx.try_recv().is_ok() {}
+	/// If no server is running, fail.
+	pub async fn echo_events_to_stdout(kinds: HashSet<String>) -> Result<()> {
+		let (mut 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?;
 
-		tokio::spawn(async move {
+		while let Ok(Some(s)) = lines.next_line().await {
-			let mut server = None;
+			let kind = s.split(',').next();
-			let (mut lines, mut writer) = Self::connect(&mut server).await;
+			if matches!(kind, Some(kind) if kinds.contains(kind)) {
-
+				println!("{}", s);
-			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) {
-								continue;
-							}
-
-							if kinds.contains(payload.body.kind()) {
-								println!("{}", &line);
-							}
-						} else {
-							if ! sender.matches(payload.sender) {
-								continue;
-							}
-
-							if kinds.contains(payload.body.kind()) {
-								println!("{}", &line);
-							}
-						}
-					}
-				}
 			}
-		});
+		}
+
+		Ok(())
 	}
 
+	/// 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)?;
 

yazi-dds/src/client/dds_peer.rs

@@ -1,25 +0,0 @@
-/// 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),
-		}
-	}
-}