diff --git a/Cargo.lock b/Cargo.lock index c3450634..dfc885c4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2742,6 +2742,7 @@ 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 9f412674..aeab2837 100644 --- a/yazi-boot/src/boot.rs +++ b/yazi-boot/src/boot.rs @@ -155,6 +155,27 @@ impl Boot { ); } } + + /// Allows for initialization without forcing the reading of command line + /// arguments. + pub fn init_with(local_events: HashSet, remote_events: HashSet) -> 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 4b706ed0..7d7477f4 100644 --- a/yazi-cli/Cargo.toml +++ b/yazi-cli/Cargo.toml @@ -9,6 +9,7 @@ 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 cb8f39bd..6da0242a 100644 --- a/yazi-cli/src/args.rs +++ b/yazi-cli/src/args.rs @@ -22,6 +22,10 @@ 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), } #[derive(clap::Args)] @@ -109,3 +113,21 @@ pub(super) struct CommandPack { #[arg(short = 'u', long)] pub(super) upgrade: bool, } + +#[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 e7aacaf0..652a9e2f 100644 --- a/yazi-cli/src/main.rs +++ b/yazi-cli/src/main.rs @@ -1,8 +1,11 @@ mod args; mod package; +use std::collections::HashSet; + use args::*; use clap::Parser; +use yazi_dds::dds_peer::DDSPeer; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -46,6 +49,26 @@ 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::>(); + + 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); + + tokio::signal::ctrl_c().await?; + } } Ok(()) diff --git a/yazi-dds/src/client.rs b/yazi-dds/src/client.rs index b9bf8d18..3a51bb4f 100644 --- a/yazi-dds/src/client.rs +++ b/yazi-dds/src/client.rs @@ -6,7 +6,9 @@ use serde::{Deserialize, Serialize}; use tokio::{io::AsyncWriteExt, select, sync::mpsc, task::JoinHandle, time}; use yazi_shared::RoCell; -use crate::{body::{Body, BodyBye, BodyHi}, ClientReader, ClientWriter, Payload, Pubsub, Server, Stream}; +use crate::{body::{Body, BodyBye, BodyHi}, dds_peer::DDSPeer, ClientReader, ClientWriter, Payload, Pubsub, Server, Stream}; + +pub mod dds_peer; pub(super) static ID: RoCell = RoCell::new(); pub(super) static PEERS: RoCell>> = RoCell::new(); @@ -52,7 +54,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 +64,53 @@ impl Client { }); } + pub fn echo_events_to_stdout(sender: DDSPeer, kinds: HashSet) { + let mut rx = QUEUE_RX.drop(); + while rx.try_recv().is_ok() {} + + 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) { + println!("{}", line); + } + } else { + if ! sender.matches(payload.sender) { + continue; + } + + if kinds.contains(payload.body.kind()) { + println!("{}", &line); + } + } + } + } + } + }); + } + pub async fn shot(kind: &str, receiver: u64, severity: Option, body: &str) -> Result<()> { Body::validate(kind)?; @@ -136,8 +185,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; } diff --git a/yazi-dds/src/client/dds_peer.rs b/yazi-dds/src/client/dds_peer.rs new file mode 100644 index 00000000..31e44e3d --- /dev/null +++ b/yazi-dds/src/client/dds_peer.rs @@ -0,0 +1,25 @@ +/// 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), + } + } +}