feat: terminal response detection under async stdin (#2347)

Cargo.lock

@@ -726,9 +726,9 @@ checksum = "c7f84e12ccf0a7ddc17a6c41c93326024c42920d7ee630d04950e6926645c0fe"
 
 [[package]]
 name = "equivalent"
-version = "1.0.1"
+version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5443807d6dff69373d433ab9ef5378ad8df50ca6298caf15de6e52e24aaf54d5"
+checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
 
 [[package]]
 name = "erased-serde"
@@ -2390,9 +2390,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.13.2"
+version = "1.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67"
+checksum = "7fcf8323ef1faaee30a44a340193b1ac6814fd9b7b4e88e9d4519a3e4abe1cfd"
 
 [[package]]
 name = "smawk"
@@ -3324,10 +3324,12 @@ dependencies = [
  "crossterm",
  "futures",
  "image",
+ "libc",
  "ratatui",
  "scopeguard",
  "tokio",
  "tracing",
+ "windows-sys 0.59.0",
  "yazi-config",
  "yazi-macro",
  "yazi-shared",

yazi-adapter/Cargo.toml

@@ -26,5 +26,11 @@ scopeguard  = { workspace = true }
 tokio       = { workspace = true }
 tracing     = { workspace = true }
 
+[target."cfg(unix)".dependencies]
+libc  = { workspace = true }
+
+[target.'cfg(windows)'.dependencies]
+windows-sys = { version = "0.59.0", features = [ "Win32_UI_Shell" ] }
+
 [target.'cfg(target_os = "macos")'.dependencies]
 crossterm = { workspace = true, features = [ "use-dev-tty", "libc" ] }

yazi-adapter/src/emulator.rs

@@ -1,13 +1,13 @@
 use std::{io::{LineWriter, stderr}, time::Duration};
 
-use anyhow::{Result, bail};
+use anyhow::Result;
 use crossterm::{cursor::{RestorePosition, SavePosition}, execute, style::Print, terminal::{disable_raw_mode, enable_raw_mode}};
 use scopeguard::defer;
-use tokio::{io::{AsyncReadExt, BufReader}, time::{sleep, timeout}};
+use tokio::time::sleep;
 use tracing::{debug, error, warn};
 use yazi_shared::Either;
 
-use crate::{Adapter, Brand, Mux, TMUX, Unknown};
+use crate::{Adapter, AsyncStdin, Brand, Mux, TMUX, Unknown};
 
 #[derive(Clone, Copy, Debug)]
 pub struct Emulator {
@@ -43,7 +43,7 @@ impl Emulator {
 			RestorePosition
 		)?;
 
-		let resp = futures::executor::block_on(Self::read_until_da1());
+		let resp = Self::read_until_da1();
 		Mux::tmux_drain()?;
 
 		let kind = if let Some(b) = Brand::from_csi(&resp).or(resort) {
@@ -105,69 +105,40 @@ impl Emulator {
 		result
 	}
 
-	pub async fn read_until_da1() -> String {
+	pub fn read_until_da1() -> String {
-		let mut buf: Vec<u8> = Vec::with_capacity(200);
+		let h = tokio::spawn(Self::error_to_user());
-		let read = async {
+		let (buf, result) = AsyncStdin::default().read_until(Duration::from_millis(300), |b, buf| {
-			let mut stdin = BufReader::new(tokio::io::stdin());
+			b == b'c'
-			loop {
+				&& buf.contains(&0x1b)
-				let mut c = [0; 1];
+				&& buf.rsplitn(2, |&b| b == 0x1b).next().is_some_and(|s| s.starts_with(b"[?"))
-				if stdin.read(&mut c).await? == 0 {
-					bail!("unexpected EOF");
-				}
-				buf.push(c[0]);
-				if c[0] != b'c' || !buf.contains(&0x1b) {
-					continue;
-				}
-				if buf.rsplitn(2, |&b| b == 0x1b).next().is_some_and(|s| s.starts_with(b"[?")) {
-					break;
-				}
-			}
-			Ok(())
-		};
-
-		let h = tokio::spawn(async move {
-			sleep(Duration::from_millis(300)).await;
-			Self::error_to_user().ok();
 		});
 
-		match timeout(Duration::from_secs(2), read).await {
-			Ok(Ok(())) => debug!("read_until_da1: {buf:?}"),
-			Err(e) => error!("read_until_da1 timed out: {buf:?}, error: {e:?}"),
-			Ok(Err(e)) => error!("read_until_da1 failed: {buf:?}, error: {e:?}"),
-		}
-
 		h.abort();
+		match result {
+			Ok(()) => debug!("read_until_da1: {buf:?}"),
+			Err(e) => error!("read_until_da1 failed: {buf:?}, error: {e:?}"),
+		}
+
 		String::from_utf8_lossy(&buf).into_owned()
 	}
 
-	pub async fn read_until_dsr() -> String {
+	pub fn read_until_dsr() -> String {
-		let mut buf: Vec<u8> = Vec::with_capacity(200);
+		let (buf, result) = AsyncStdin::default().read_until(Duration::from_millis(100), |b, buf| {
-		let read = async {
+			b == b'n' && (buf.ends_with(b"\x1b[0n") || buf.ends_with(b"\x1b[3n"))
-			let mut stdin = BufReader::new(tokio::io::stdin());
+		});
-			loop {
-				let mut c = [0; 1];
-				if stdin.read(&mut c).await? == 0 {
-					bail!("unexpected EOF");
-				}
-				buf.push(c[0]);
-				if c[0] == b'n' && (buf.ends_with(b"\x1b[0n") || buf.ends_with(b"\x1b[3n")) {
-					break;
-				}
-			}
-			Ok(())
-		};
 
-		match timeout(Duration::from_millis(500), read).await {
+		match result {
-			Ok(Ok(())) => debug!("read_until_dsr: {buf:?}"),
+			Ok(()) => debug!("read_until_dsr: {buf:?}"),
-			Err(e) => error!("read_until_dsr timed out: {buf:?}, error: {e:?}"),
+			Err(e) => error!("read_until_dsr failed: {buf:?}, error: {e:?}"),
-			Ok(Err(e)) => error!("read_until_dsr failed: {buf:?}, error: {e:?}"),
 		}
 		String::from_utf8_lossy(&buf).into_owned()
 	}
 
-	fn error_to_user() -> Result<(), std::io::Error> {
+	async fn error_to_user() {
 		use crossterm::style::{Attribute, Color, Print, ResetColor, SetAttributes, SetForegroundColor};
-		crossterm::execute!(
+
+		sleep(Duration::from_millis(200)).await;
+		_ = crossterm::execute!(
 			std::io::stderr(),
 			SetForegroundColor(Color::Red),
 			SetAttributes(Attribute::Bold.into()),
@@ -179,7 +150,7 @@ impl Emulator {
 			Print(
 				"Please check your terminal environment as per: https://yazi-rs.github.io/docs/faq#trt\r\n"
 			),
-		)
+		);
 	}
 
 	fn cell_size(resp: &str) -> Option<(u16, u16)> {

yazi-adapter/src/lib.rs

@@ -1,8 +1,8 @@
-#![allow(clippy::unit_arg)]
+#![allow(clippy::unit_arg, clippy::option_map_unit_fn)]
 
 yazi_macro::mod_pub!(drivers);
 
-yazi_macro::mod_flat!(adapter brand dimension emulator image info mux unknown);
+yazi_macro::mod_flat!(adapter brand dimension emulator image info mux stdin unknown);
 
 use yazi_shared::{SyncCell, in_wsl};
 

yazi-adapter/src/mux.rs

@@ -43,7 +43,7 @@ impl Mux {
 	pub fn tmux_drain() -> Result<()> {
 		if TMUX.get() {
 			crossterm::execute!(std::io::stderr(), crossterm::style::Print(Mux::csi("\x1b[5n")))?;
-			_ = futures::executor::block_on(Emulator::read_until_dsr());
+			_ = Emulator::read_until_dsr();
 		}
 		Ok(())
 	}

yazi-adapter/src/stdin.rs

@@ -0,0 +1,132 @@
+use std::{io::{Error, ErrorKind}, time::{Duration, Instant}};
+
+pub struct AsyncStdin {
+	#[cfg(unix)]
+	fds: libc::fd_set,
+}
+
+impl AsyncStdin {
+	pub fn read_until<P>(&mut self, timeout: Duration, predicate: P) -> (Vec<u8>, std::io::Result<()>)
+	where
+		P: Fn(u8, &[u8]) -> bool,
+	{
+		let mut buf: Vec<u8> = Vec::with_capacity(200);
+		let now = Instant::now();
+
+		let mut read = || {
+			loop {
+				if now.elapsed() > timeout {
+					return Err(Error::new(ErrorKind::TimedOut, "timed out"));
+				} else if !self.poll(Duration::from_millis(30))? {
+					continue;
+				}
+
+				let b = Self::read_u8()?;
+				buf.push(b);
+
+				if predicate(b, &buf) {
+					break;
+				}
+			}
+			Ok(())
+		};
+
+		let result = read();
+		(buf, result)
+	}
+}
+
+#[cfg(unix)]
+impl Default for AsyncStdin {
+	fn default() -> Self {
+		let mut me = Self { fds: unsafe { std::mem::MaybeUninit::zeroed().assume_init() } };
+		me.reset();
+		me
+	}
+}
+
+#[cfg(unix)]
+impl AsyncStdin {
+	pub fn poll(&mut self, timeout: Duration) -> std::io::Result<bool> {
+		let mut tv = libc::timeval {
+			tv_sec:  timeout.as_secs() as libc::time_t,
+			tv_usec: timeout.subsec_micros() as libc::suseconds_t,
+		};
+
+		let result = unsafe {
+			libc::select(
+				libc::STDIN_FILENO + 1,
+				&mut self.fds,
+				std::ptr::null_mut(),
+				std::ptr::null_mut(),
+				&mut tv,
+			)
+		};
+
+		match result {
+			-1 => Err(Error::last_os_error()),
+			0 => Ok(false),
+			_ => {
+				self.reset();
+				Ok(true)
+			}
+		}
+	}
+
+	pub fn read_u8() -> std::io::Result<u8> {
+		let mut b = 0;
+		match unsafe { libc::read(libc::STDIN_FILENO, &mut b as *mut _ as *mut _, 1) } {
+			-1 => Err(Error::last_os_error()),
+			0 => Err(Error::new(ErrorKind::UnexpectedEof, "unexpected EOF")),
+			_ => Ok(b),
+		}
+	}
+
+	fn reset(&mut self) {
+		unsafe {
+			libc::FD_ZERO(&mut self.fds);
+			libc::FD_SET(libc::STDIN_FILENO, &mut self.fds);
+		}
+	}
+}
+
+#[cfg(windows)]
+impl Default for AsyncStdin {
+	fn default() -> Self { Self {} }
+}
+
+#[cfg(windows)]
+impl AsyncStdin {
+	pub fn poll(&mut self, timeout: Duration) -> std::io::Result<bool> {
+		use std::os::windows::io::AsRawHandle;
+
+		use windows_sys::Win32::{Foundation::{WAIT_FAILED, WAIT_OBJECT_0}, System::Threading::WaitForSingleObject};
+
+		let handle = std::io::stdin().as_raw_handle();
+		let millis = timeout.as_millis();
+		match unsafe { WaitForSingleObject(handle, millis as u32) } {
+			WAIT_FAILED => Err(Error::last_os_error()),
+			WAIT_OBJECT_0 => Ok(true),
+			_ => Ok(false),
+		}
+	}
+
+	pub fn read_u8() -> std::io::Result<u8> {
+		use std::os::windows::io::AsRawHandle;
+
+		use windows_sys::Win32::Storage::FileSystem::ReadFile;
+
+		let mut buf = 0;
+		let mut bytes = 0;
+		let success = unsafe {
+			ReadFile(std::io::stdin().as_raw_handle(), &mut buf, 1, &mut bytes, std::ptr::null_mut())
+		};
+
+		if success == 0 {
+			return Err(Error::last_os_error());
+		} else if bytes == 0 {
+			return Err(Error::new(ErrorKind::UnexpectedEof, "unexpected EOF"));
+		}
+		Ok(buf)
+	}
+}

yazi-fm/src/term.rs

@@ -42,13 +42,14 @@ impl Term {
 			mouse::SetMouse(true),
 		)?;
 
-		let da = futures::executor::block_on(Emulator::read_until_da1());
+		let resp = Emulator::read_until_da1();
 		Mux::tmux_drain()?;
 
-		CSI_U.store(da.contains("\x1b[?0u"), Ordering::Relaxed);
+		CSI_U.store(resp.contains("\x1b[?0u"), Ordering::Relaxed);
-		BLINK.store(da.contains("\x1b[?12;1$y"), Ordering::Relaxed);
+		BLINK.store(resp.contains("\x1b[?12;1$y"), Ordering::Relaxed);
 		SHAPE.store(
-			da.split_once("\x1bP1$r")
+			resp
+				.split_once("\x1bP1$r")
 				.and_then(|(_, s)| s.bytes().next())
 				.filter(|&b| matches!(b, b'0'..=b'6'))
 				.map_or(u8::MAX, |b| b - b'0'),