diff --git a/yazi-cli/src/args.rs b/yazi-cli/src/args.rs index 6da0242a..4cd0eb14 100644 --- a/yazi-cli/src/args.rs +++ b/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), } diff --git a/yazi-cli/src/main.rs b/yazi-cli/src/main.rs index 652a9e2f..ef95a9dc 100644 --- a/yazi-cli/src/main.rs +++ b/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::>(); - - 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::>(); - yazi_boot::BOOT.init(yazi_boot::Boot::init_with(kinds.clone(), kinds.clone())); - yazi_dds::Client::echo_events_to_stdout(DDSPeer::All, kinds); + yazi_dds::Client::echo_events_to_stdout(kinds).await?; tokio::signal::ctrl_c().await?; } diff --git a/yazi-dds/src/body/hi.rs b/yazi-dds/src/body/hi.rs index c60a2370..a66c3eaf 100644 --- a/yazi-dds/src/body/hi.rs +++ b/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>, pub version: String, } diff --git a/yazi-dds/src/client.rs b/yazi-dds/src/client.rs index 00335591..c8fbefc8 100644 --- a/yazi-dds/src/client.rs +++ b/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}; - -pub mod dds_peer; +use crate::{body::{Body, BodyBye, BodyHi}, ClientReader, ClientWriter, Payload, Pubsub, Server, Stream}; pub(super) static ID: RoCell = RoCell::new(); pub(super) static PEERS: RoCell>> = 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) { - let mut rx = QUEUE_RX.drop(); - while rx.try_recv().is_ok() {} + /// 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. + pub async fn echo_events_to_stdout(kinds: HashSet) -> 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 { - 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) { - continue; - } - - if kinds.contains(payload.body.kind()) { - println!("{}", &line); - } - } else { - if ! sender.matches(payload.sender) { - continue; - } - - if kinds.contains(payload.body.kind()) { - println!("{}", &line); - } - } - } - } + 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(()) } + /// Connect to an existing server to send a single message. pub async fn shot(kind: &str, receiver: u64, severity: Option, body: &str) -> Result<()> { Body::validate(kind)?; diff --git a/yazi-dds/src/client/dds_peer.rs b/yazi-dds/src/client/dds_peer.rs deleted file mode 100644 index 31e44e3d..00000000 --- a/yazi-dds/src/client/dds_peer.rs +++ /dev/null @@ -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 for DDSPeer { - fn from(value: u64) -> Self { - match value { - 0 => Self::All, - _ => Self::One(value), - } - } -}