feat: add new debounce option to ya.input() API (#1025)

Cargo.lock

@@ -2857,6 +2857,7 @@ dependencies = [
  "shell-words",
  "syntect",
  "tokio",
+ "tokio-stream",
  "tokio-util",
  "tracing",
  "unicode-width",

yazi-plugin/Cargo.toml

@@ -36,6 +36,7 @@ shell-escape  = "0.1.5"
 shell-words   = "1.1.0"
 syntect       = { version = "5.2.0", default-features = false, features = [ "parsing", "plist-load", "regex-onig" ] }
 tokio         = { version = "1.37.0", features = [ "full" ] }
+tokio-stream = "0.1.15"
 tokio-util    = "0.7.11"
 unicode-width = "0.1.12"
 yazi-prebuild = "0.1.2"

yazi-plugin/src/bindings/input.rs

@@ -1,15 +1,23 @@
+use std::pin::Pin;
+
 use mlua::{prelude::LuaUserDataMethods, UserData};
-use tokio::sync::mpsc::UnboundedReceiver;
+use tokio::pin;
+use tokio_stream::StreamExt;
 use yazi_shared::InputError;
 
-pub struct InputRx {
-	inner: UnboundedReceiver<Result<String, InputError>>,
+pub struct InputRx<T: StreamExt<Item = Result<String, InputError>>> {
+	inner: T,
 }
 
-impl InputRx {
-	pub fn new(inner: UnboundedReceiver<Result<String, InputError>>) -> Self { Self { inner } }
+impl<T: StreamExt<Item = Result<String, InputError>>> InputRx<T> {
+	pub fn new(inner: T) -> Self { Self { inner } }
 
-	pub fn parse(res: Result<String, InputError>) -> (Option<String>, u8) {
+	pub async fn consume(inner: T) -> (Option<String>, u8) {
+		pin!(inner);
+		inner.next().await.map(Self::parse).unwrap_or((None, 0))
+	}
+
+	fn parse(res: Result<String, InputError>) -> (Option<String>, u8) {
 		match res {
 			Ok(s) => (Some(s), 1),
 			Err(InputError::Canceled(s)) => (Some(s), 2),
@@ -19,14 +27,11 @@ impl InputRx {
 	}
 }
 
-impl UserData for InputRx {
+impl<T: StreamExt<Item = Result<String, InputError>> + 'static> UserData for InputRx<T> {
 	fn add_methods<'lua, M: LuaUserDataMethods<'lua, Self>>(methods: &mut M) {
 		methods.add_async_method_mut("recv", |_, me, ()| async move {
-			let Some(res) = me.inner.recv().await else {
-				return Ok((None, 0));
-			};
-
-			Ok(Self::parse(res))
+			let mut inner = unsafe { Pin::new_unchecked(&mut me.inner) };
+			Ok(inner.next().await.map(Self::parse).unwrap_or((None, 0)))
 		});
 	}
 }

yazi-plugin/src/utils/layer.rs

@@ -1,10 +1,11 @@
-use std::str::FromStr;
+use std::{str::FromStr, time::Duration};
 
 use mlua::{ExternalError, ExternalResult, IntoLuaMulti, Lua, Table, Value};
 use tokio::sync::mpsc;
+use tokio_stream::wrappers::UnboundedReceiverStream;
 use yazi_config::{keymap::{Control, Key}, popup::InputCfg};
 use yazi_proxy::{AppProxy, InputProxy};
-use yazi_shared::{emit, event::Cmd, Layer};
+use yazi_shared::{emit, event::Cmd, Debounce, Layer};
 
 use super::Utils;
 use crate::bindings::{InputRx, Position};
@@ -59,7 +60,7 @@ impl Utils {
 			"input",
 			lua.create_async_function(|lua, t: Table| async move {
 				let realtime = t.raw_get("realtime").unwrap_or_default();
-				let mut rx = InputProxy::show(InputCfg {
+				let rx = UnboundedReceiverStream::new(InputProxy::show(InputCfg {
 					title: t.raw_get("title")?,
 					value: t.raw_get("value").unwrap_or_default(),
 					cursor: None, // TODO
@@ -67,14 +68,20 @@ impl Utils {
 					realtime,
 					completion: false,
 					highlight: false,
-				});
+				}));
 
-				if realtime {
+				if !realtime {
+					return InputRx::consume(rx).await.into_lua_multi(lua);
+				}
+
+				let debounce = t.raw_get::<_, f64>("debounce").unwrap_or_default();
+				if debounce < 0.0 {
+					Err("negative debounce duration".into_lua_err())
+				} else if debounce == 0.0 {
 					(InputRx::new(rx), Value::Nil).into_lua_multi(lua)
-				} else if let Some(res) = rx.recv().await {
-					InputRx::parse(res).into_lua_multi(lua)
 				} else {
-					(Value::Nil, 0).into_lua_multi(lua)
+					(InputRx::new(Debounce::new(rx, Duration::from_secs_f64(debounce))), Value::Nil)
+						.into_lua_multi(lua)
 				}
 			})?,
 		)?;