From f694b1a6ddd79b744ad5c41443c0524220b1d49e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=89=E5=92=B2=E9=9B=85=20=C2=B7=20Misaki=20Masa?= Date: Wed, 23 Aug 2023 10:55:39 +0800 Subject: [PATCH] feat: update mime-type after file changes detected (#78) --- Cargo.lock | 13 ++++ core/Cargo.toml | 1 + core/src/external/fd.rs | 12 ++-- core/src/external/rg.rs | 12 ++-- core/src/manager/manager.rs | 7 +- core/src/manager/tab.rs | 6 +- core/src/manager/watcher.rs | 136 ++++++++++++++++++++++-------------- shared/Cargo.toml | 1 + shared/src/buffer.rs | 38 ---------- shared/src/lib.rs | 8 ++- shared/src/stream.rs | 69 ++++++++++++++++++ shared/src/throttle.rs | 11 ++- shared/src/time.rs | 5 ++ 13 files changed, 204 insertions(+), 115 deletions(-) delete mode 100644 shared/src/buffer.rs create mode 100644 shared/src/stream.rs create mode 100644 shared/src/time.rs diff --git a/Cargo.lock b/Cargo.lock index c55379bf..375fcd70 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -389,6 +389,7 @@ dependencies = [ "shared", "syntect", "tokio", + "tokio-stream", "tracing", "trash", "unicode-width", @@ -1623,6 +1624,7 @@ version = "0.1.0" dependencies = [ "anyhow", "crossterm 0.27.0", + "futures", "libc", "parking_lot", "ratatui", @@ -1886,6 +1888,17 @@ dependencies = [ "syn 2.0.29", ] +[[package]] +name = "tokio-stream" +version = "0.1.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "397c988d37662c7dda6d2208364a706264bf3d6138b11d436cbac0ad38832842" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml" version = "0.7.6" diff --git a/core/Cargo.toml b/core/Cargo.toml index f7633328..3645e372 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -21,6 +21,7 @@ serde = "^1" serde_json = "^1" syntect = "^5" tokio = { version = "^1", features = [ "parking_lot", "macros", "rt-multi-thread", "sync", "time", "fs", "process", "io-std", "io-util" ] } +tokio-stream = "^0" tracing = "^0" trash = "^3" unicode-width = "^0" diff --git a/core/src/external/fd.rs b/core/src/external/fd.rs index 91b8106b..b93baaf6 100644 --- a/core/src/external/fd.rs +++ b/core/src/external/fd.rs @@ -1,8 +1,9 @@ use std::{path::PathBuf, process::Stdio, time::Duration}; use anyhow::Result; -use shared::DelayedBuffer; -use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, sync::mpsc::UnboundedReceiver}; +use shared::StreamBuf; +use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, sync::mpsc}; +use tokio_stream::wrappers::UnboundedReceiverStream; pub struct FdOpt { pub cwd: PathBuf, @@ -11,7 +12,7 @@ pub struct FdOpt { pub subject: String, } -pub fn fd(opt: FdOpt) -> Result>> { +pub fn fd(opt: FdOpt) -> Result>> { let mut child = Command::new("fd") .arg("--base-directory") .arg(&opt.cwd) @@ -26,11 +27,12 @@ pub fn fd(opt: FdOpt) -> Result>> { drop(child.stderr.take()); let mut it = BufReader::new(child.stdout.take().unwrap()).lines(); - let (mut buf, rx) = DelayedBuffer::new(Duration::from_millis(100)); + let (tx, rx) = mpsc::unbounded_channel(); + let rx = StreamBuf::new(UnboundedReceiverStream::new(rx), Duration::from_millis(300)); tokio::spawn(async move { while let Ok(Some(line)) = it.next_line().await { - buf.push(opt.cwd.join(line)); + tx.send(opt.cwd.join(line)).ok(); } child.wait().await.ok(); }); diff --git a/core/src/external/rg.rs b/core/src/external/rg.rs index c63aa322..41bd4ca2 100644 --- a/core/src/external/rg.rs +++ b/core/src/external/rg.rs @@ -1,8 +1,9 @@ use std::{path::PathBuf, process::Stdio, time::Duration}; use anyhow::Result; -use shared::DelayedBuffer; -use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, sync::mpsc::UnboundedReceiver}; +use shared::StreamBuf; +use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, sync::mpsc}; +use tokio_stream::wrappers::UnboundedReceiverStream; pub struct RgOpt { pub cwd: PathBuf, @@ -10,7 +11,7 @@ pub struct RgOpt { pub subject: String, } -pub fn rg(opt: RgOpt) -> Result>> { +pub fn rg(opt: RgOpt) -> Result>> { let mut child = Command::new("rg") .current_dir(&opt.cwd) .args(["--color=never", "--files-with-matches", "--smart-case"]) @@ -24,11 +25,12 @@ pub fn rg(opt: RgOpt) -> Result>> { drop(child.stderr.take()); let mut it = BufReader::new(child.stdout.take().unwrap()).lines(); - let (mut buf, rx) = DelayedBuffer::new(Duration::from_millis(100)); + let (tx, rx) = mpsc::unbounded_channel(); + let rx = StreamBuf::new(UnboundedReceiverStream::new(rx), Duration::from_millis(300)); tokio::spawn(async move { while let Ok(Some(line)) = it.next_line().await { - buf.push(opt.cwd.join(line)); + tx.send(opt.cwd.join(line)).ok(); } child.wait().await.ok(); }); diff --git a/core/src/manager/manager.rs b/core/src/manager/manager.rs index 9865b19c..821e2f4f 100644 --- a/core/src/manager/manager.rs +++ b/core/src/manager/manager.rs @@ -30,9 +30,10 @@ impl Manager { pub fn refresh(&mut self) { env::set_current_dir(self.cwd()).ok(); - self.watcher.trigger(self.cwd()); - if let Some(p) = self.parent() { - self.watcher.trigger(&p.cwd); + if let Some(f) = self.parent() { + self.watcher.trigger_dirs(&[self.cwd(), &f.cwd]); + } else { + self.watcher.trigger_dirs(&[self.cwd()]); } emit!(Hover); diff --git a/core/src/manager/tab.rs b/core/src/manager/tab.rs index 20ff703b..b059a0b9 100644 --- a/core/src/manager/tab.rs +++ b/core/src/manager/tab.rs @@ -2,6 +2,7 @@ use std::{collections::{BTreeMap, BTreeSet}, ffi::{OsStr, OsString}, mem, path:: use anyhow::{Error, Result}; use config::open::Opener; +use futures::StreamExt; use shared::Defer; use tokio::task::JoinHandle; @@ -241,10 +242,7 @@ impl Tab { }?; emit!(Files(FilesOp::search_empty(&cwd))); - while let Some(chunk) = rx.recv().await { - if chunk.is_empty() { - break; - } + while let Some(chunk) = rx.next().await { emit!(Files(FilesOp::Search(cwd.clone(), Files::read(chunk).await))); } Ok(()) diff --git a/core/src/manager/watcher.rs b/core/src/manager/watcher.rs index f8be2d86..f1ff33ba 100644 --- a/core/src/manager/watcher.rs +++ b/core/src/manager/watcher.rs @@ -1,22 +1,25 @@ -use std::{collections::{BTreeMap, BTreeSet}, path::{Path, PathBuf}, sync::Arc}; +use std::{collections::{BTreeMap, BTreeSet}, path::{Path, PathBuf}, sync::Arc, time::Duration}; +use futures::StreamExt; use indexmap::IndexMap; use notify::{event::{MetadataKind, ModifyKind}, EventKind, RecommendedWatcher, RecursiveMode, Watcher as _Watcher}; use parking_lot::RwLock; -use tokio::{fs, sync::mpsc::{self, Receiver, Sender}}; +use shared::StreamBuf; +use tokio::{fs, sync::mpsc}; +use tokio_stream::wrappers::UnboundedReceiverStream; -use crate::{emit, files::{Files, FilesOp}}; +use crate::{emit, external, files::{Files, FilesOp}}; pub struct Watcher { - tx: Sender, - watcher: RecommendedWatcher, watched: Arc>>>, } impl Watcher { pub(super) fn start() -> Self { - let (tx, rx) = mpsc::channel(50); + let (tx, rx) = mpsc::unbounded_channel(); + let rx = StreamBuf::new(UnboundedReceiverStream::new(rx), Duration::from_millis(300)); + let watcher = RecommendedWatcher::new( { let tx = tx.clone(); @@ -32,10 +35,11 @@ impl Watcher { let parent = path.parent().unwrap_or(&path).to_path_buf(); match event.kind { EventKind::Create(_) => { - tx.blocking_send(parent).ok(); + tx.send(parent).ok(); } EventKind::Modify(kind) => { match kind { + ModifyKind::Data(_) => {} ModifyKind::Metadata(kind) => match kind { MetadataKind::Permissions => {} MetadataKind::Ownership => {} @@ -46,12 +50,12 @@ impl Watcher { _ => return, }; - tx.blocking_send(path).ok(); - tx.blocking_send(parent).ok(); + tx.send(path).ok(); + tx.send(parent).ok(); } EventKind::Remove(_) => { - tx.blocking_send(path).ok(); - tx.blocking_send(parent).ok(); + tx.send(path).ok(); + tx.send(parent).ok(); } _ => (), } @@ -60,47 +64,11 @@ impl Watcher { Default::default(), ); - let instance = Self { tx, watcher: watcher.unwrap(), watched: Default::default() }; + let instance = Self { watcher: watcher.unwrap(), watched: Default::default() }; tokio::spawn(Self::changed(rx, instance.watched.clone())); instance } - async fn changed( - mut rx: Receiver, - watched: Arc>>>, - ) { - while let Some(path) = rx.recv().await { - let linked = watched - .read() - .iter() - .map_while(|(k, v)| v.as_ref().and_then(|v| path.strip_prefix(v).ok()).map(|v| k.join(v))) - .collect::>(); - - let result = Files::read_dir(&path).await; - if linked.is_empty() { - emit!(Files(match result { - Ok(items) => FilesOp::Read(path, items), - Err(_) => FilesOp::IOErr(path), - })); - continue; - } - - for ori in linked { - emit!(Files(match &result { - Ok(items) => { - let files = BTreeMap::from_iter(items.iter().map(|(p, f)| { - let p = ori.join(p.strip_prefix(&path).unwrap()); - let f = f.clone().set_path(&p); - (p, f) - })); - FilesOp::Read(ori, files) - } - Err(_) => FilesOp::IOErr(ori), - })); - } - } - } - pub(super) fn watch(&mut self, mut to_watch: BTreeSet) { let keys = self.watched.read().keys().cloned().collect::>(); for p in keys.difference(&to_watch) { @@ -142,11 +110,75 @@ impl Watcher { }); } - pub(super) fn trigger(&self, path: &Path) { - let tx = self.tx.clone(); - let path = path.to_path_buf(); + pub(super) fn trigger_dirs(&self, dirs: &[&Path]) { + let watched = self.watched.clone(); + let dirs = dirs.iter().map(|p| p.to_path_buf()).collect::>(); tokio::spawn(async move { - tx.send(path).await.ok(); + for dir in dirs { + Self::dir_changed(&dir, watched.clone()).await; + } }); } + + async fn changed( + mut rx: StreamBuf>, + watched: Arc>>>, + ) { + while let Some(paths) = rx.next().await { + let (mut files, mut dirs): (Vec<_>, Vec<_>) = Default::default(); + for path in paths.into_iter().collect::>() { + if fs::symlink_metadata(&path).await.map(|m| !m.is_dir()).unwrap_or(false) { + files.push(path); + } else { + dirs.push(path); + } + } + + Self::file_changed(files.iter().map(AsRef::as_ref).collect()).await; + for file in files { + emit!(Files(FilesOp::IOErr(file))); + } + + for dir in dirs { + Self::dir_changed(&dir, watched.clone()).await; + } + } + } + + async fn file_changed(paths: Vec<&Path>) { + if let Ok(mimes) = external::file(&paths).await { + emit!(Mimetype(mimes)); + } + } + + async fn dir_changed(path: &Path, watched: Arc>>>) { + let linked = watched + .read() + .iter() + .map_while(|(k, v)| v.as_ref().and_then(|v| path.strip_prefix(v).ok()).map(|v| k.join(v))) + .collect::>(); + + let result = Files::read_dir(path).await; + if linked.is_empty() { + emit!(Files(match result { + Ok(items) => FilesOp::Read(path.into(), items), + Err(_) => FilesOp::IOErr(path.into()), + })); + return; + } + + for ori in linked { + emit!(Files(match &result { + Ok(items) => { + let files = BTreeMap::from_iter(items.iter().map(|(p, f)| { + let p = ori.join(p.strip_prefix(path).unwrap()); + let f = f.clone().set_path(&p); + (p, f) + })); + FilesOp::Read(ori, files) + } + Err(_) => FilesOp::IOErr(ori), + })); + } + } } diff --git a/shared/Cargo.toml b/shared/Cargo.toml index 06c9fa4d..953218e9 100644 --- a/shared/Cargo.toml +++ b/shared/Cargo.toml @@ -6,6 +6,7 @@ edition = "2021" [dependencies] anyhow = "^1" crossterm = "^0" +futures = "^0" libc = "^0" parking_lot = "^0" ratatui = { version = "^0" } diff --git a/shared/src/buffer.rs b/shared/src/buffer.rs deleted file mode 100644 index 5b8674f9..00000000 --- a/shared/src/buffer.rs +++ /dev/null @@ -1,38 +0,0 @@ -use std::time::Duration; - -use tokio::{sync::mpsc::{self, UnboundedReceiver, UnboundedSender}, time::Instant}; - -pub struct DelayedBuffer { - buf: Vec, - tx: UnboundedSender>, - last: Instant, - interval: Duration, -} - -impl DelayedBuffer { - pub fn new(interval: Duration) -> (Self, UnboundedReceiver>) { - let (tx, rx) = mpsc::unbounded_channel(); - (Self { buf: Vec::new(), tx, last: Instant::now() - interval, interval }, rx) - } - - pub fn push(&mut self, item: T) { - self.buf.push(item); - if self.last.elapsed() >= self.interval { - self.last = Instant::now(); - self.tx.send(self.buf.drain(..).collect()).ok(); - } - } - - pub fn flush(&mut self) { - if !self.buf.is_empty() { - self.tx.send(self.buf.drain(..).collect()).ok(); - } - } -} - -impl Drop for DelayedBuffer { - fn drop(&mut self) { - self.flush(); - self.tx.send(vec![]).ok(); - } -} diff --git a/shared/src/lib.rs b/shared/src/lib.rs index 84a13d97..326423e1 100644 --- a/shared/src/lib.rs +++ b/shared/src/lib.rs @@ -1,19 +1,23 @@ -mod buffer; +#![allow(clippy::option_map_unit_fn)] + mod chars; mod defer; mod fns; mod fs; mod mime; mod ro_cell; +mod stream; mod term; mod throttle; +mod time; -pub use buffer::*; pub use chars::*; pub use defer::*; pub use fns::*; pub use fs::*; pub use mime::*; pub use ro_cell::*; +pub use stream::*; pub use term::*; pub use throttle::*; +pub use time::*; diff --git a/shared/src/stream.rs b/shared/src/stream.rs new file mode 100644 index 00000000..1c72dd9b --- /dev/null +++ b/shared/src/stream.rs @@ -0,0 +1,69 @@ +use std::{mem, pin::Pin, task::{Context, Poll}, time::Duration}; + +use futures::{FutureExt, Stream, StreamExt}; +use tokio::time::{sleep, Instant, Sleep}; + +pub struct StreamBuf +where + S: Stream, +{ + stream: S, + interval: Duration, + + sleep: Sleep, + pending: Vec, +} + +impl Unpin for StreamBuf where S: Stream {} + +impl StreamBuf +where + S: Stream + Unpin, +{ + pub fn new(stream: S, interval: Duration) -> StreamBuf { + Self { stream, interval, sleep: sleep(Duration::ZERO), pending: Default::default() } + } + + pub fn flush(&mut self) { + let sleep = unsafe { Pin::new_unchecked(&mut self.sleep) }; + sleep.reset(Instant::now()); + } +} + +impl Stream for StreamBuf +where + S: Stream + Unpin, +{ + type Item = Vec; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let (mut stream, interval, mut sleep, pending) = unsafe { + let me = self.get_unchecked_mut(); + (Pin::new(&mut me.stream), me.interval, Pin::new_unchecked(&mut me.sleep), &mut me.pending) + }; + + while let Poll::Ready(next) = stream.poll_next_unpin(cx) { + match next { + Some(next) => pending.push(next), + None => { + if pending.is_empty() { + return Poll::Ready(None); + } + break; + } + } + } + + if pending.is_empty() { + return Poll::Pending; + } + + match sleep.poll_unpin(cx) { + Poll::Ready(_) => { + sleep.reset(Instant::now() + interval); + Poll::Ready(Some(mem::take(pending))) + } + Poll::Pending => Poll::Pending, + } + } +} diff --git a/shared/src/throttle.rs b/shared/src/throttle.rs index 248eaa69..8ffb29af 100644 --- a/shared/src/throttle.rs +++ b/shared/src/throttle.rs @@ -1,7 +1,9 @@ -use std::{fmt::Debug, mem, sync::atomic::{AtomicU64, AtomicUsize, Ordering}, time::{self, Duration, SystemTime}}; +use std::{fmt::Debug, mem, sync::atomic::{AtomicU64, AtomicUsize, Ordering}, time::Duration}; use parking_lot::Mutex; +use crate::timestamp_ms; + #[derive(Debug)] pub struct Throttle { total: AtomicUsize, @@ -15,10 +17,7 @@ impl Throttle { Self { total: AtomicUsize::new(total), interval, - last: AtomicU64::new( - SystemTime::now().duration_since(time::UNIX_EPOCH).unwrap().as_millis() as u64 - - interval.as_millis() as u64, - ), + last: AtomicU64::new(timestamp_ms() - interval.as_millis() as u64), buf: Default::default(), } } @@ -33,7 +32,7 @@ impl Throttle { } let last = self.last.load(Ordering::Relaxed); - let now = SystemTime::now().duration_since(time::UNIX_EPOCH).unwrap().as_millis() as u64; + let now = timestamp_ms(); if now > self.interval.as_millis() as u64 + last { self.last.store(now, Ordering::Relaxed); return self.flush(data, f); diff --git a/shared/src/time.rs b/shared/src/time.rs new file mode 100644 index 00000000..580605e2 --- /dev/null +++ b/shared/src/time.rs @@ -0,0 +1,5 @@ +use std::time::{self, SystemTime}; + +pub fn timestamp_ms() -> u64 { + SystemTime::now().duration_since(time::UNIX_EPOCH).unwrap().as_millis() as u64 +}