..

yazi-dds/src/sendable.rs

@@ -24,53 +24,43 @@ impl Sendable {
 			}
 			Value::Function(_) => Err("function is not supported".into_lua_err())?,
 			Value::Thread(_) => Err("thread is not supported".into_lua_err())?,
-			Value::UserData(_) => Err("userdata is not supported".into_lua_err())?,
+			Value::UserData(ud) => {
-			Value::Error(_) => Err("error is not supported".into_lua_err())?,
+				if let Ok(t) = ud.take::<yazi_shared::fs::Url>() {
-		})
+					Data::Url(t)
-	}
+				} else if let Ok(t) = ud.take::<super::body::BodyYankIter>() {
-
+					Data::Any(Box::new(t))
-	pub fn value_to_key(value: Value) -> mlua::Result<DataKey> {
+				} else {
-		Ok(match value {
+					Err("unsupported userdata included".into_lua_err())?
-			Value::Nil => DataKey::Nil,
+				}
-			Value::Boolean(v) => DataKey::Boolean(v),
+			}
-			Value::LightUserData(_) => Err("light userdata is not supported".into_lua_err())?,
-			Value::Integer(v) => DataKey::Integer(v),
-			Value::Number(v) => DataKey::Number(OrderedFloat::new(v)),
-			Value::String(v) => DataKey::String(v.to_str()?.to_owned()),
-			Value::Table(_) => Err("table is not supported".into_lua_err())?,
-			Value::Function(_) => Err("function is not supported".into_lua_err())?,
-			Value::Thread(_) => Err("thread is not supported".into_lua_err())?,
-			Value::UserData(_) => Err("userdata is not supported".into_lua_err())?,
 			Value::Error(_) => Err("error is not supported".into_lua_err())?,
 		})
 	}
 
 	pub fn data_to_value(lua: &Lua, data: Data) -> mlua::Result<Value> {
-		match data {
+		Ok(match data {
-			Data::Nil => Ok(Value::Nil),
+			Data::Nil => Value::Nil,
-			Data::Boolean(v) => Ok(Value::Boolean(v)),
+			Data::Boolean(v) => Value::Boolean(v),
-			Data::Integer(v) => Ok(Value::Integer(v)),
+			Data::Integer(v) => Value::Integer(v),
-			Data::Number(v) => Ok(Value::Number(v)),
+			Data::Number(v) => Value::Number(v),
-			Data::String(v) => Ok(Value::String(lua.create_string(v)?)),
+			Data::String(v) => Value::String(lua.create_string(v)?),
-			Data::Table(v) => {
+			Data::Table(t) => {
-				let seq_len = v.keys().filter(|&k| !k.is_numeric()).count();
+				let seq_len = t.keys().filter(|&k| !k.is_numeric()).count();
-				let table = lua.create_table_with_capacity(seq_len, v.len() - seq_len)?;
+				let table = lua.create_table_with_capacity(seq_len, t.len() - seq_len)?;
-				for (k, v) in v {
+				for (k, v) in t {
 					table.raw_set(Self::key_to_value(lua, k)?, Self::data_to_value(lua, v)?)?;
 				}
-				Ok(Value::Table(table))
+				Value::Table(table)
 			}
-		}
+			Data::Url(v) => Value::UserData(lua.create_any_userdata(v)?),
-	}
+			Data::Any(v) => {
-
+				if let Ok(t) = v.downcast::<super::body::BodyYankIter>() {
-	pub fn key_to_value(lua: &Lua, key: DataKey) -> mlua::Result<Value> {
+					Value::UserData(lua.create_userdata(*t)?)
-		match key {
+				} else {
-			DataKey::Nil => Ok(Value::Nil),
+					Err("unsupported userdata included".into_lua_err())?
-			DataKey::Boolean(k) => Ok(Value::Boolean(k)),
+				}
-			DataKey::Integer(k) => Ok(Value::Integer(k)),
+			}
-			DataKey::Number(k) => Ok(Value::Number(k.get())),
+		})
-			DataKey::String(k) => Ok(Value::String(lua.create_string(k)?)),
-		}
 	}
 
 	pub fn vec_to_table(lua: &Lua, data: Vec<Data>) -> mlua::Result<Table> {
@@ -96,4 +86,30 @@ impl Sendable {
 		}
 		Ok(vec)
 	}
+
+	fn value_to_key(value: Value) -> mlua::Result<DataKey> {
+		Ok(match value {
+			Value::Nil => DataKey::Nil,
+			Value::Boolean(v) => DataKey::Boolean(v),
+			Value::LightUserData(_) => Err("light userdata is not supported".into_lua_err())?,
+			Value::Integer(v) => DataKey::Integer(v),
+			Value::Number(v) => DataKey::Number(OrderedFloat::new(v)),
+			Value::String(v) => DataKey::String(v.to_str()?.to_owned()),
+			Value::Table(_) => Err("table is not supported".into_lua_err())?,
+			Value::Function(_) => Err("function is not supported".into_lua_err())?,
+			Value::Thread(_) => Err("thread is not supported".into_lua_err())?,
+			Value::UserData(_) => Err("userdata is not supported".into_lua_err())?,
+			Value::Error(_) => Err("error is not supported".into_lua_err())?,
+		})
+	}
+
+	fn key_to_value(lua: &Lua, key: DataKey) -> mlua::Result<Value> {
+		Ok(match key {
+			DataKey::Nil => Value::Nil,
+			DataKey::Boolean(k) => Value::Boolean(k),
+			DataKey::Integer(k) => Value::Integer(k),
+			DataKey::Number(k) => Value::Number(k.get()),
+			DataKey::String(k) => Value::String(lua.create_string(k)?),
+		})
+	}
 }

yazi-scheduler/src/file/file.rs

@@ -1,6 +1,6 @@
 use std::{borrow::Cow, collections::VecDeque, fs::Metadata, path::{Path, PathBuf}};
 
-use anyhow::Result;
+use anyhow::{anyhow, Result};
 use futures::{future::BoxFuture, FutureExt};
 use tokio::{fs, io::{self, ErrorKind::{AlreadyExists, NotFound}}, sync::mpsc};
 use tracing::warn;
@@ -51,9 +51,10 @@ impl File {
 							if task.retry < TASKS.bizarre_retry
 								&& matches!(e.raw_os_error(), Some(1) | Some(93)) =>
 						{
-							self.log(task.id, format!("Paste task retry: {:?}", task))?;
 							task.retry += 1;
-							return Ok(self.macro_.send(FileOp::Paste(task).into(), LOW).await?);
+							self.log(task.id, format!("Paste task retry: {:?}", task))?;
+							self.queue(FileOp::Paste(task), LOW).await?;
+							return Ok(());
 						}
 						Err(e) => Err(e)?,
 					}
@@ -147,9 +148,9 @@ impl File {
 			self.prog.send(TaskProg::New(id, meta.len()))?;
 
 			if meta.is_file() {
-				self.macro_.send(FileOp::Paste(task).into(), LOW).await?;
+				self.queue(FileOp::Paste(task), LOW).await?;
 			} else if meta.is_symlink() {
-				self.macro_.send(FileOp::Link(task.to_link(meta)).into(), NORMAL).await?;
+				self.queue(FileOp::Link(task.to_link(meta)), NORMAL).await?;
 			}
 			return self.succ(id);
 		}
@@ -193,9 +194,9 @@ impl File {
 				self.prog.send(TaskProg::New(task.id, meta.len()))?;
 
 				if meta.is_file() {
-					self.macro_.send(FileOp::Paste(task.clone()).into(), LOW).await?;
+					self.queue(FileOp::Paste(task.clone()), LOW).await?;
 				} else if meta.is_symlink() {
-					self.macro_.send(FileOp::Link(task.to_link(meta)).into(), NORMAL).await?;
+					self.queue(FileOp::Link(task.to_link(meta)), NORMAL).await?;
 				}
 			}
 		}
@@ -209,7 +210,7 @@ impl File {
 		}
 
 		self.prog.send(TaskProg::New(id, task.meta.as_ref().unwrap().len()))?;
-		self.macro_.send(FileOp::Link(task).into(), NORMAL).await?;
+		self.queue(FileOp::Link(task), NORMAL).await?;
 		self.succ(id)
 	}
 
@@ -219,7 +220,7 @@ impl File {
 			let id = task.id;
 			task.length = meta.len();
 			self.prog.send(TaskProg::New(id, meta.len()))?;
-			self.macro_.send(FileOp::Delete(task).into(), NORMAL).await?;
+			self.queue(FileOp::Delete(task), NORMAL).await?;
 			return self.succ(id);
 		}
 
@@ -238,7 +239,7 @@ impl File {
 				task.target = Url::from(entry.path());
 				task.length = meta.len();
 				self.prog.send(TaskProg::New(task.id, meta.len()))?;
-				self.macro_.send(FileOp::Delete(task.clone()).into(), NORMAL).await?;
+				self.queue(FileOp::Delete(task.clone()), NORMAL).await?;
 			}
 		}
 		self.succ(task.id)
@@ -249,7 +250,7 @@ impl File {
 		task.length = calculate_size(&task.target).await;
 
 		self.prog.send(TaskProg::New(id, task.length))?;
-		self.macro_.send(FileOp::Trash(task).into(), LOW).await?;
+		self.queue(FileOp::Trash(task), LOW).await?;
 		self.succ(id)
 	}
 
@@ -296,6 +297,11 @@ impl File {
 	fn log(&self, id: usize, line: String) -> Result<()> {
 		Ok(self.prog.send(TaskProg::Log(id, line))?)
 	}
+
+	#[inline]
+	async fn queue(&self, op: impl Into<TaskOp>, priority: u8) -> Result<()> {
+		self.macro_.send(op.into(), priority).await.map_err(|_| anyhow!("Failed to send task"))
+	}
 }
 
 impl FileOpPaste {

yazi-scheduler/src/plugin/plugin.rs

@@ -1,4 +1,4 @@
-use anyhow::Result;
+use anyhow::{anyhow, Result};
 use tokio::sync::mpsc;
 use yazi_plugin::isolate;
 
@@ -43,7 +43,7 @@ impl Plugin {
 		let id = task.id;
 
 		self.prog.send(TaskProg::New(id, 0))?;
-		self.macro_.try_send(PluginOp::Entry(task).into(), HIGH)?;
+		self.queue(PluginOp::Entry(task), HIGH)?;
 		self.succ(id)
 	}
 }
@@ -56,4 +56,9 @@ impl Plugin {
 	fn fail(&self, id: usize, reason: String) -> Result<()> {
 		Ok(self.prog.send(TaskProg::Fail(id, reason))?)
 	}
+
+	#[inline]
+	fn queue(&self, op: impl Into<TaskOp>, priority: u8) -> Result<()> {
+		self.macro_.try_send(op.into(), priority).map_err(|_| anyhow!("Failed to send task"))
+	}
 }

yazi-scheduler/src/preload/preload.rs

@@ -1,6 +1,6 @@
 use std::collections::{HashMap, HashSet};
 
-use anyhow::Result;
+use anyhow::{anyhow, Result};
 use parking_lot::RwLock;
 use tokio::sync::mpsc;
 use tracing::error;
@@ -73,8 +73,8 @@ impl Preload {
 		self.prog.send(TaskProg::New(id, 0))?;
 
 		match task.plugin.prio {
-			Priority::Low => self.macro_.send(PreloadOp::Rule(task).into(), NORMAL).await?,
+			Priority::Low => self.queue(PreloadOp::Rule(task), NORMAL).await?,
-			Priority::Normal => self.macro_.send(PreloadOp::Rule(task).into(), HIGH).await?,
+			Priority::Normal => self.queue(PreloadOp::Rule(task), HIGH).await?,
 			Priority::High => self.work(PreloadOp::Rule(task)).await?,
 		}
 		self.succ(id)
@@ -97,4 +97,9 @@ impl Preload {
 	fn fail(&self, id: usize, reason: String) -> Result<()> {
 		Ok(self.prog.send(TaskProg::Fail(id, reason))?)
 	}
+
+	#[inline]
+	async fn queue(&self, op: impl Into<TaskOp>, priority: u8) -> Result<()> {
+		self.macro_.send(op.into(), priority).await.map_err(|_| anyhow!("Failed to send task"))
+	}
 }

yazi-shared/src/event/data.rs

@@ -1,8 +1,8 @@
-use std::collections::HashMap;
+use std::{any::Any, collections::HashMap};
 
 use serde::{Deserialize, Serialize};
 
-use crate::OrderedFloat;
+use crate::{fs::Url, OrderedFloat};
 
 // --- Arg
 #[derive(Debug, Serialize, Deserialize)]
@@ -14,6 +14,10 @@ pub enum Data {
 	Number(f64),
 	String(String),
 	Table(HashMap<DataKey, Data>),
+	#[serde(skip)]
+	Url(Url),
+	#[serde(skip)]
+	Any(Box<dyn Any + Send>),
 }
 
 impl Data {