feat: update mime-type after file changes detected (#78)

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"

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"

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<UnboundedReceiver<Vec<PathBuf>>> {
+pub fn fd(opt: FdOpt) -> Result<StreamBuf<UnboundedReceiverStream<PathBuf>>> {
 	let mut child = Command::new("fd")
 		.arg("--base-directory")
 		.arg(&opt.cwd)
@@ -26,11 +27,12 @@ pub fn fd(opt: FdOpt) -> Result<UnboundedReceiver<Vec<PathBuf>>> {
 	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();
 	});

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<UnboundedReceiver<Vec<PathBuf>>> {
+pub fn rg(opt: RgOpt) -> Result<StreamBuf<UnboundedReceiverStream<PathBuf>>> {
 	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<UnboundedReceiver<Vec<PathBuf>>> {
 	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();
 	});

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);
 

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(())

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<PathBuf>,
-
 	watcher: RecommendedWatcher,
 	watched: Arc<RwLock<IndexMap<PathBuf, Option<PathBuf>>>>,
 }
 
 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<PathBuf>,
-		watched: Arc<RwLock<IndexMap<PathBuf, Option<PathBuf>>>>,
-	) {
-		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::<Vec<_>>();
-
-			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<PathBuf>) {
 		let keys = self.watched.read().keys().cloned().collect::<BTreeSet<_>>();
 		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::<Vec<_>>();
 		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<UnboundedReceiverStream<PathBuf>>,
+		watched: Arc<RwLock<IndexMap<PathBuf, Option<PathBuf>>>>,
+	) {
+		while let Some(paths) = rx.next().await {
+			let (mut files, mut dirs): (Vec<_>, Vec<_>) = Default::default();
+			for path in paths.into_iter().collect::<BTreeSet<_>>() {
+				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<RwLock<IndexMap<PathBuf, Option<PathBuf>>>>) {
+		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::<Vec<_>>();
+
+		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),
+			}));
+		}
+	}
 }

shared/Cargo.toml

@@ -6,6 +6,7 @@ edition = "2021"
 [dependencies]
 anyhow      = "^1"
 crossterm   = "^0"
+futures     = "^0"
 libc        = "^0"
 parking_lot = "^0"
 ratatui  		 = { version = "^0" }

shared/src/buffer.rs

@@ -1,38 +0,0 @@
-use std::time::Duration;
-
-use tokio::{sync::mpsc::{self, UnboundedReceiver, UnboundedSender}, time::Instant};
-
-pub struct DelayedBuffer<T> {
-	buf:      Vec<T>,
-	tx:       UnboundedSender<Vec<T>>,
-	last:     Instant,
-	interval: Duration,
-}
-
-impl<T> DelayedBuffer<T> {
-	pub fn new(interval: Duration) -> (Self, UnboundedReceiver<Vec<T>>) {
-		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<T> Drop for DelayedBuffer<T> {
-	fn drop(&mut self) {
-		self.flush();
-		self.tx.send(vec![]).ok();
-	}
-}

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::*;

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<S>
+where
+	S: Stream,
+{
+	stream:   S,
+	interval: Duration,
+
+	sleep:   Sleep,
+	pending: Vec<S::Item>,
+}
+
+impl<S> Unpin for StreamBuf<S> where S: Stream {}
+
+impl<S> StreamBuf<S>
+where
+	S: Stream + Unpin,
+{
+	pub fn new(stream: S, interval: Duration) -> StreamBuf<S> {
+		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<S> Stream for StreamBuf<S>
+where
+	S: Stream + Unpin,
+{
+	type Item = Vec<S::Item>;
+
+	fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+		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,
+		}
+	}
+}

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<T> {
 	total:    AtomicUsize,
@@ -15,10 +17,7 @@ impl<T> Throttle<T> {
 		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<T> Throttle<T> {
 		}
 
 		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);

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
+}