feat: reconnect after a connection is lost

If the connection cannot be made when starting, the client will fail and
not reconnect. However, if the connection is lost after the client has
started, it will forever attempt to reconnect with a 1s delay.

Also clean up the implementation by removing previous mistakes

Cargo.lock

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

yazi-boot/src/boot.rs

@@ -155,27 +155,6 @@ 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,6 @@ 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

@@ -23,7 +23,7 @@ pub(super) enum Command {
 	/// Manage packages.
 	Pack(CommandPack),
 	/// Subscribe to messages from all remote instance(s).
-	SubStatic(CommandSubStatic),
+	Sub(CommandSub),
 }
 
 #[derive(clap::Args)]
@@ -114,17 +114,6 @@ pub(super) struct CommandPack {
 
 #[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

@@ -49,7 +49,7 @@ async fn main() -> anyhow::Result<()> {
 			}
 		}
 
-		Command::SubStatic(cmd) => {
+		Command::Sub(cmd) => {
 			yazi_dds::init();
 			let kinds = cmd.kinds.split(',').map(|s| s.to_owned()).collect::<HashSet<_>>();
 

yazi-dds/src/client.rs

@@ -65,21 +65,51 @@ 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.
+	/// 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, 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?;
+		let mut lines = Self::connect_listener(&kinds).await?;
+		let mut msg_counter = 0;
 
-		while let Ok(Some(s)) = lines.next_line().await {
+		loop {
+			println!("Waiting for messages...");
+			match lines.next_line().await {
+				Ok(Some(s)) => {
+					println!("Received: message {} '{}'", msg_counter, s);
+					msg_counter += 1;
 					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;
+							println!("Reconnected");
+							break;
+						}
+						Err(_) => {
+							println!("Reconnecting...");
+							time::sleep(time::Duration::from_secs(1)).await;
+						}
+					};
+				},
+				Err(e) => {
+					// could not establish initial connection
+					return Err(e.into());
+				}
+			}
+		}
+	}
 
-		Ok(())
+	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.