From 2961ecf5b26cf5eb7a56904f9bc4815f5091911f Mon Sep 17 00:00:00 2001 From: sxyazi Date: Fri, 29 Mar 2024 13:50:35 +0800 Subject: [PATCH] .. --- yazi-boot/src/boot.rs | 2 ++ yazi-dds/src/body/bulk.rs | 5 ---- yazi-dds/src/body/cd.rs | 7 +---- yazi-dds/src/body/hover.rs | 7 +---- yazi-dds/src/body/rename.rs | 3 -- yazi-dds/src/body/tabs.rs | 12 +------- yazi-dds/src/body/yank.rs | 7 +---- yazi-dds/src/client.rs | 12 ++++---- yazi-dds/src/lib.rs | 4 ++- yazi-dds/src/pubsub.rs | 32 ++++++++++---------- yazi-dds/src/server.rs | 23 +++++++------- yazi-dds/src/state.rs | 51 ++++++++++++++++++++++++++++++++ yazi-fm/src/app/commands/quit.rs | 1 + 13 files changed, 96 insertions(+), 70 deletions(-) create mode 100644 yazi-dds/src/state.rs diff --git a/yazi-boot/src/boot.rs b/yazi-boot/src/boot.rs index 3320f877..450e461d 100644 --- a/yazi-boot/src/boot.rs +++ b/yazi-boot/src/boot.rs @@ -19,6 +19,7 @@ pub struct Boot { pub config_dir: PathBuf, pub flavor_dir: PathBuf, pub plugin_dir: PathBuf, + pub state_dir: PathBuf, } impl Boot { @@ -153,6 +154,7 @@ impl Default for Boot { flavor_dir: config_dir.join("flavors"), plugin_dir: config_dir.join("plugins"), config_dir, + state_dir: Xdg::state_dir(), }; std::fs::create_dir_all(&boot.flavor_dir).expect("Failed to create flavor directory"); diff --git a/yazi-dds/src/body/bulk.rs b/yazi-dds/src/body/bulk.rs index 68a14944..c15ccd3d 100644 --- a/yazi-dds/src/body/bulk.rs +++ b/yazi-dds/src/body/bulk.rs @@ -24,11 +24,6 @@ impl BodyBulk<'static> { pub fn owned(tab: usize, changes: &HashMap) -> Body<'static> { Self { tab, changes: Cow::Owned(changes.clone()) }.into() } - - #[inline] - pub fn digest(tab: usize, changes: &HashMap) -> Body<'static> { - Self::owned(tab, changes) - } } impl<'a> From> for Body<'a> { diff --git a/yazi-dds/src/body/cd.rs b/yazi-dds/src/body/cd.rs index b9dbb98f..5c8fa761 100644 --- a/yazi-dds/src/body/cd.rs +++ b/yazi-dds/src/body/cd.rs @@ -22,12 +22,7 @@ impl<'a> BodyCd<'a> { impl BodyCd<'static> { #[inline] - pub fn owned(tab: usize, url: &Url) -> Body<'static> { - Self { owned: true, tab, url: Cow::Owned(url.clone()) }.into() - } - - #[inline] - pub fn digest(tab: usize) -> Body<'static> { + pub fn owned(tab: usize) -> Body<'static> { Self { owned: false, tab, url: Default::default() }.into() } } diff --git a/yazi-dds/src/body/hover.rs b/yazi-dds/src/body/hover.rs index 3753fe83..45211b62 100644 --- a/yazi-dds/src/body/hover.rs +++ b/yazi-dds/src/body/hover.rs @@ -22,12 +22,7 @@ impl<'a> BodyHover<'a> { impl BodyHover<'static> { #[inline] - pub fn owned(tab: usize, url: Option<&Url>) -> Body<'static> { - Self { owned: true, tab, url: url.map(|u| Cow::Owned(u.clone())) }.into() - } - - #[inline] - pub fn digest(tab: usize) -> Body<'static> { Self { owned: false, tab, url: None }.into() } + pub fn owned(tab: usize) -> Body<'static> { Self { owned: false, tab, url: None }.into() } } impl<'a> From> for Body<'a> { diff --git a/yazi-dds/src/body/rename.rs b/yazi-dds/src/body/rename.rs index 0842bb2d..98fac976 100644 --- a/yazi-dds/src/body/rename.rs +++ b/yazi-dds/src/body/rename.rs @@ -25,9 +25,6 @@ impl BodyRename<'static> { pub fn owned(tab: usize, from: &Url, to: &Url) -> Body<'static> { Self { tab, from: Cow::Owned(from.clone()), to: Cow::Owned(to.clone()) }.into() } - - #[inline] - pub fn digest(tab: usize, from: &Url, to: &Url) -> Body<'static> { Self::owned(tab, from, to) } } impl<'a> From> for Body<'a> { diff --git a/yazi-dds/src/body/tabs.rs b/yazi-dds/src/body/tabs.rs index d5d25b3f..e6d7a762 100644 --- a/yazi-dds/src/body/tabs.rs +++ b/yazi-dds/src/body/tabs.rs @@ -27,17 +27,7 @@ impl<'a> BodyTabs<'a> { impl BodyTabs<'static> { #[inline] - pub fn owned(cursor: usize, urls: &[&Url]) -> Body<'static> { - Self { - owned: true, - cursor, - items: urls.iter().map(|&u| BodyTabsItem { url: Cow::Owned(u.clone()) }).collect(), - } - .into() - } - - #[inline] - pub fn digest(cursor: usize) -> Body<'static> { + pub fn owned(cursor: usize) -> Body<'static> { Self { owned: false, cursor, items: Default::default() }.into() } } diff --git a/yazi-dds/src/body/yank.rs b/yazi-dds/src/body/yank.rs index cda76007..bacb6434 100644 --- a/yazi-dds/src/body/yank.rs +++ b/yazi-dds/src/body/yank.rs @@ -22,12 +22,7 @@ impl<'a> BodyYank<'a> { impl BodyYank<'static> { #[inline] - pub fn owned(cut: bool, urls: &HashSet) -> Body<'static> { - Self { owned: true, cut, urls: Cow::Owned(urls.clone()) }.into() - } - - #[inline] - pub fn digest(cut: bool) -> Body<'static> { + pub fn owned(cut: bool) -> Body<'static> { Self { owned: false, cut, urls: Default::default() }.into() } } diff --git a/yazi-dds/src/client.rs b/yazi-dds/src/client.rs index 94a3b8c9..3f0ceca3 100644 --- a/yazi-dds/src/client.rs +++ b/yazi-dds/src/client.rs @@ -9,7 +9,7 @@ use crate::{body::Body, Payload, Pubsub, Server}; pub(super) static ID: RoCell = RoCell::new(); pub(super) static PEERS: RoCell>> = RoCell::new(); -pub(super) static QUEUE: RoCell> = RoCell::new(); +pub(super) static QUEUE: RoCell> = RoCell::new(); #[cfg(not(unix))] use tokio::net::TcpStream; @@ -29,7 +29,7 @@ pub struct Peer { } impl Client { - pub(super) fn serve(mut rx: mpsc::UnboundedReceiver>) { + pub(super) fn serve(mut rx: mpsc::UnboundedReceiver) { while rx.try_recv().is_ok() {} tokio::spawn(async move { @@ -39,10 +39,9 @@ impl Client { loop { select! { Some(payload) = rx.recv() => { - let s = format!("{payload}\n"); - if writer.write_all(s.as_bytes()).await.is_err() { + if writer.write_all(payload.as_bytes()).await.is_err() { (lines, writer) = Self::reconnect(&mut server).await; - writer.write_all(s.as_bytes()).await.ok(); // Retry once + writer.write_all(payload.as_bytes()).await.ok(); // Retry once } } Ok(next) = lines.next_line() => { @@ -62,6 +61,9 @@ impl Client { }); } + #[inline] + pub(super) fn push(payload: Payload) { QUEUE.send(format!("{}\n", payload)).ok(); } + #[inline] pub(super) fn able(&self, ability: &str) -> bool { self.abilities.contains(ability) } diff --git a/yazi-dds/src/lib.rs b/yazi-dds/src/lib.rs index 2274d4f0..a148f92b 100644 --- a/yazi-dds/src/lib.rs +++ b/yazi-dds/src/lib.rs @@ -5,12 +5,14 @@ mod payload; mod pubsub; mod sendable; mod server; +mod state; pub use client::*; pub use payload::*; pub use pubsub::*; pub use sendable::*; use server::*; +pub use state::*; pub fn init() { let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); @@ -22,7 +24,7 @@ pub fn init() { // Server CLIENTS.with(Default::default); - STATES.with(Default::default); + STATE.with(Default::default); // Pubsub LOCAL.with(Default::default); diff --git a/yazi-dds/src/pubsub.rs b/yazi-dds/src/pubsub.rs index 69d46faf..fb72c2c8 100644 --- a/yazi-dds/src/pubsub.rs +++ b/yazi-dds/src/pubsub.rs @@ -5,7 +5,7 @@ use parking_lot::RwLock; use yazi_boot::BOOT; use yazi_shared::{fs::Url, RoCell}; -use crate::{body::{Body, BodyCd, BodyHi, BodyHover, BodyRename, BodyTabs, BodyYank}, Payload, ID, PEERS, QUEUE}; +use crate::{body::{Body, BodyCd, BodyHi, BodyHover, BodyRename, BodyTabs, BodyYank}, Client, Payload, ID, PEERS}; pub static LOCAL: RoCell>>>> = RoCell::new(); @@ -74,21 +74,21 @@ impl Pubsub { let (kind, peers) = (body.kind(), PEERS.read()); if receiver == 0 && peers.values().any(|c| c.able(kind)) { - QUEUE.send(body.upgrade()).ok(); + Client::push(body.upgrade()); } else if peers.get(&receiver).is_some_and(|c| c.able(kind)) { - QUEUE.send(body.upgrade().with_receiver(receiver)).ok(); + Client::push(body.upgrade().with_receiver(receiver)); } } - pub fn pub_static(severity: u8, body: Body<'static>) { + pub fn pub_static(severity: u8, body: Body) { let (kind, peers) = (body.kind(), PEERS.read()); if peers.values().any(|c| c.able(kind)) { - QUEUE.send(body.upgrade().with_severity(severity)).ok(); + Client::push(body.upgrade().with_severity(severity)); } } pub fn pub_from_hi() -> bool { - _ = QUEUE.send(Payload::new( + Client::push(Payload::new( BodyHi { id: *ID, abilities: REMOTE.read().keys().cloned().collect() }.into(), )); true @@ -96,10 +96,10 @@ impl Pubsub { pub fn pub_from_tabs(tab: usize, urls: &[&Url]) { if LOCAL.read().contains_key("tabs") { - Self::pub_(BodyTabs::digest(tab)); + Self::pub_(BodyTabs::owned(tab)); } if PEERS.read().values().any(|p| p.able("tabs")) { - QUEUE.send(BodyTabs::owned(tab, urls).upgrade()).ok(); + Client::push(BodyTabs::borrowed(tab, urls).upgrade()); } if BOOT.local_events.contains("tabs") { BodyTabs::borrowed(tab, urls).upgrade().flush(true); @@ -108,10 +108,10 @@ impl Pubsub { pub fn pub_from_cd(tab: usize, url: &Url) { if LOCAL.read().contains_key("cd") { - Self::pub_(BodyCd::digest(tab)); + Self::pub_(BodyCd::owned(tab)); } if PEERS.read().values().any(|p| p.able("cd")) { - QUEUE.send(BodyCd::owned(tab, url).upgrade()).ok(); + Client::push(BodyCd::borrowed(tab, url).upgrade()); } if BOOT.local_events.contains("cd") { BodyCd::borrowed(tab, url).upgrade().flush(true); @@ -120,10 +120,10 @@ impl Pubsub { pub fn pub_from_hover(tab: usize, url: Option<&Url>) { if LOCAL.read().contains_key("hover") { - Self::pub_(BodyHover::digest(tab)); + Self::pub_(BodyHover::owned(tab)); } if PEERS.read().values().any(|p| p.able("hover")) { - QUEUE.send(BodyHover::owned(tab, url).upgrade()).ok(); + Client::push(BodyHover::borrowed(tab, url).upgrade()); } if BOOT.local_events.contains("hover") { BodyHover::borrowed(tab, url).upgrade().flush(true); @@ -132,10 +132,10 @@ impl Pubsub { pub fn pub_from_rename(tab: usize, from: &Url, to: &Url) { if LOCAL.read().contains_key("rename") { - Self::pub_(BodyRename::digest(tab, from, to)); + Self::pub_(BodyRename::owned(tab, from, to)); } if PEERS.read().values().any(|p| p.able("rename")) { - QUEUE.send(BodyRename::owned(tab, from, to).upgrade()).ok(); + Client::push(BodyRename::borrowed(tab, from, to).upgrade()); } if BOOT.local_events.contains("rename") { BodyRename::borrowed(tab, from, to).upgrade().flush(true); @@ -144,10 +144,10 @@ impl Pubsub { pub fn pub_from_yank(cut: bool, urls: &HashSet) { if LOCAL.read().contains_key("yank") { - Self::pub_(BodyYank::digest(cut)); + Self::pub_(BodyYank::owned(cut)); } if PEERS.read().values().any(|p| p.able("yank")) { - QUEUE.send(BodyYank::owned(cut, urls).upgrade()).ok(); + Client::push(BodyYank::borrowed(cut, urls).upgrade()); } if BOOT.local_events.contains("yank") { BodyYank::borrowed(cut, urls).upgrade().flush(true); diff --git a/yazi-dds/src/server.rs b/yazi-dds/src/server.rs index 7c9be169..2cbd035c 100644 --- a/yazi-dds/src/server.rs +++ b/yazi-dds/src/server.rs @@ -1,14 +1,13 @@ use std::{collections::HashMap, str::FromStr, time::Duration}; use anyhow::Result; -use parking_lot::{Mutex, RwLock}; +use parking_lot::RwLock; use tokio::{io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, select, sync::mpsc, task::JoinHandle, time}; use yazi_shared::RoCell; -use crate::{body::{Body, BodyHey}, Client, Payload, Peer}; +use crate::{body::{Body, BodyHey}, Client, Payload, Peer, STATE}; pub(super) static CLIENTS: RoCell>> = RoCell::new(); -pub(super) static STATES: RoCell>> = RoCell::new(); pub(super) struct Server; @@ -48,11 +47,6 @@ impl Server { let Some(receiver) = parts.next().and_then(|s| s.parse().ok()) else { continue }; let Some(severity) = parts.next().and_then(|s| s.parse::().ok()) else { continue }; - if receiver == 0 && severity > 0 { - let Some(body) = parts.next() else { continue }; - STATES.lock().insert(format!("{}_{severity}_{kind}", Body::tab(kind, body)), line.clone()); - } - let clients = CLIENTS.read(); let clients: Vec<_> = if receiver == 0 { clients.values().filter(|c| c.id != id && c.able(kind)).collect() @@ -62,10 +56,17 @@ impl Server { vec![] }; - if !clients.is_empty() { - line.push('\n'); - clients.into_iter().for_each(|c| _ = c.tx.send(line.clone())); + if clients.is_empty() { + continue; } + + if receiver == 0 && severity > 0 { + let Some(body) = parts.next() else { continue }; + STATE.lock().add(format!("{}_{severity}_{kind}", Body::tab(kind, body)), &line); + } + + line.push('\n'); + clients.into_iter().for_each(|c| _ = c.tx.send(line.clone())); } else => break } diff --git a/yazi-dds/src/state.rs b/yazi-dds/src/state.rs new file mode 100644 index 00000000..ed2b3739 --- /dev/null +++ b/yazi-dds/src/state.rs @@ -0,0 +1,51 @@ +use std::{collections::HashMap, io::{BufRead, BufReader, BufWriter, Write}, mem}; + +use anyhow::Result; +use parking_lot::Mutex; +use yazi_boot::BOOT; +use yazi_shared::RoCell; + +use crate::{body::Body, QUEUE}; + +pub static STATE: RoCell> = RoCell::new(); + +#[derive(Default)] +pub struct State { + inner: HashMap, +} + +impl State { + pub fn add(&mut self, key: String, value: &str) { self.inner.insert(key, value.to_owned()); } + + pub fn load(&mut self) -> Result<()> { + let mut buf = BufReader::new(std::fs::File::open(BOOT.state_dir.join("state"))?); + let mut line = String::new(); + + while buf.read_line(&mut line)? > 0 { + let mut parts = line.splitn(4, ','); + let Some(kind) = parts.next() else { continue }; + let Some(_) = parts.next() else { continue }; + let Some(severity) = parts.next().and_then(|s| s.parse::().ok()) else { continue }; + let Some(body) = parts.next() else { continue }; + + self.inner.insert(format!("{}_{severity}_{kind}", Body::tab(kind, body)), line.clone()); + QUEUE.send(mem::take(&mut line)).ok(); + } + Ok(()) + } + + pub fn drain(&mut self) -> Result<()> { + let mut buf = BufWriter::new( + std::fs::OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .open(BOOT.state_dir.join("state"))?, + ); + + let mut state = mem::take(&mut self.inner).into_iter().collect::>(); + state.sort_unstable_by(|(a, _), (b, _)| a.cmp(b)); + state.into_iter().for_each(|(_, v)| _ = writeln!(buf, "{v}")); + Ok(()) + } +} diff --git a/yazi-fm/src/app/commands/quit.rs b/yazi-fm/src/app/commands/quit.rs index e3d2ac18..5004ea3b 100644 --- a/yazi-fm/src/app/commands/quit.rs +++ b/yazi-fm/src/app/commands/quit.rs @@ -10,6 +10,7 @@ impl App { self.cx.tasks.shutdown(); self.cx.manager.shutdown(); + yazi_dds::STATE.lock().drain().ok(); if !opt.no_cwd_file { self.cwd_to_file(); }