From 8b9ed5e39ebe4b2357b4705f8f001e4d3711e4a9 Mon Sep 17 00:00:00 2001
From: Mika Vilpas <mika.vilpas@gmail.com>
Date: Sun, 12 May 2024 16:12:29 +0300
Subject: [PATCH] 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             |  1 -
 yazi-boot/src/boot.rs  | 21 -----------------
 yazi-cli/Cargo.toml    |  1 -
 yazi-cli/src/args.rs   | 13 +----------
 yazi-cli/src/main.rs   |  2 +-
 yazi-dds/src/client.rs | 52 +++++++++++++++++++++++++++++++++---------
 6 files changed, 43 insertions(+), 47 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index dfc885c4..c3450634 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2742,7 +2742,6 @@ dependencies = [
  "tokio",
  "toml_edit",
  "vergen",
- "yazi-boot",
  "yazi-dds",
  "yazi-shared",
 ]
diff --git a/yazi-boot/src/boot.rs b/yazi-boot/src/boot.rs
index aeab2837..9f412674 100644
--- a/yazi-boot/src/boot.rs
+++ b/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 {
diff --git a/yazi-cli/Cargo.toml b/yazi-cli/Cargo.toml
index 7d7477f4..4b706ed0 100644
--- a/yazi-cli/Cargo.toml
+++ b/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" }
 
diff --git a/yazi-cli/src/args.rs b/yazi-cli/src/args.rs
index 4cd0eb14..2998a6ee 100644
--- a/yazi-cli/src/args.rs
+++ b/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,
diff --git a/yazi-cli/src/main.rs b/yazi-cli/src/main.rs
index ef95a9dc..26dc2b60 100644
--- a/yazi-cli/src/main.rs
+++ b/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<_>>();
 
diff --git a/yazi-dds/src/client.rs b/yazi-dds/src/client.rs
index c8fbefc8..fc4517b1 100644
--- a/yazi-dds/src/client.rs
+++ b/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 mut lines = Self::connect_listener(&kinds).await?;
+		let mut msg_counter = 0;
+
+		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());
+				}
+			}
+		}
+	}
+
+	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?;
-
-		while let Ok(Some(s)) = lines.next_line().await {
-			let kind = s.split(',').next();
-			if matches!(kind, Some(kind) if kinds.contains(kind)) {
-				println!("{}", s);
-			}
-		}
-
-		Ok(())
+		Ok(lines)
 	}
 
 	/// Connect to an existing server to send a single message.