This commit is contained in:
sxyazi 2023-12-29 02:02:54 +08:00
parent a9e8c9aa92
commit 2fad9bcd16
No known key found for this signature in database
4 changed files with 46 additions and 61 deletions

View file

@ -8,29 +8,16 @@ use yazi_config::TASKS;
use yazi_shared::fs::{calculate_size, copy_with_progress, path_relative_to, Url}; use yazi_shared::fs::{calculate_size, copy_with_progress, path_relative_to, Url};
use super::{FileOp, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash}; use super::{FileOp, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash};
use crate::TaskProg; use crate::{TaskOp, TaskProg};
pub struct File { pub struct File {
tx: async_channel::Sender<FileOp>, macro_: async_channel::Sender<TaskOp>,
rx: async_channel::Receiver<FileOp>, prog: mpsc::UnboundedSender<TaskProg>,
prog: mpsc::UnboundedSender<TaskProg>,
} }
impl File { impl File {
pub fn new(prog: mpsc::UnboundedSender<TaskProg>) -> Self { pub fn new(macro_: async_channel::Sender<TaskOp>, prog: mpsc::UnboundedSender<TaskProg>) -> Self {
let (tx, rx) = async_channel::unbounded(); Self { macro_, prog }
Self { tx, rx, prog }
}
#[inline]
pub async fn recv(&self) -> Result<(usize, FileOp)> {
Ok(match self.rx.recv().await? {
FileOp::Paste(t) => (t.id, FileOp::Paste(t)),
FileOp::Link(t) => (t.id, FileOp::Link(t)),
FileOp::Delete(t) => (t.id, FileOp::Delete(t)),
FileOp::Trash(t) => (t.id, FileOp::Trash(t)),
})
} }
pub async fn work(&self, op: &mut FileOp) -> Result<()> { pub async fn work(&self, op: &mut FileOp) -> Result<()> {
@ -63,7 +50,7 @@ impl File {
{ {
self.log(task.id, format!("Paste task retry: {:?}", task))?; self.log(task.id, format!("Paste task retry: {:?}", task))?;
task.retry += 1; task.retry += 1;
return Ok(self.tx.send(FileOp::Paste(task.clone())).await?); return Ok(self.macro_.send(FileOp::Paste(task.clone()).into()).await?);
} }
Err(e) => Err(e)?, Err(e) => Err(e)?,
} }
@ -157,9 +144,9 @@ impl File {
self.prog.send(TaskProg::New(id, meta.len()))?; self.prog.send(TaskProg::New(id, meta.len()))?;
if meta.is_file() { if meta.is_file() {
self.tx.send(FileOp::Paste(task)).await?; self.macro_.send(FileOp::Paste(task).into()).await?;
} else if meta.is_symlink() { } else if meta.is_symlink() {
self.tx.send(FileOp::Link(task.to_link(meta))).await?; self.macro_.send(FileOp::Link(task.to_link(meta)).into()).await?;
} }
return self.succ(id); return self.succ(id);
} }
@ -203,9 +190,9 @@ impl File {
self.prog.send(TaskProg::New(task.id, meta.len()))?; self.prog.send(TaskProg::New(task.id, meta.len()))?;
if meta.is_file() { if meta.is_file() {
self.tx.send(FileOp::Paste(task.clone())).await?; self.macro_.send(FileOp::Paste(task.clone()).into()).await?;
} else if meta.is_symlink() { } else if meta.is_symlink() {
self.tx.send(FileOp::Link(task.to_link(meta))).await?; self.macro_.send(FileOp::Link(task.to_link(meta)).into()).await?;
} }
} }
} }
@ -219,7 +206,7 @@ impl File {
} }
self.prog.send(TaskProg::New(id, task.meta.as_ref().unwrap().len()))?; self.prog.send(TaskProg::New(id, task.meta.as_ref().unwrap().len()))?;
self.tx.send(FileOp::Link(task)).await?; self.macro_.send(FileOp::Link(task).into()).await?;
self.succ(id) self.succ(id)
} }
@ -229,7 +216,7 @@ impl File {
let id = task.id; let id = task.id;
task.length = meta.len(); task.length = meta.len();
self.prog.send(TaskProg::New(id, meta.len()))?; self.prog.send(TaskProg::New(id, meta.len()))?;
self.tx.send(FileOp::Delete(task)).await?; self.macro_.send(FileOp::Delete(task).into()).await?;
return self.succ(id); return self.succ(id);
} }
@ -254,7 +241,7 @@ impl File {
task.target = Url::from(entry.path()); task.target = Url::from(entry.path());
task.length = meta.len(); task.length = meta.len();
self.prog.send(TaskProg::New(task.id, meta.len()))?; self.prog.send(TaskProg::New(task.id, meta.len()))?;
self.tx.send(FileOp::Delete(task.clone())).await?; self.macro_.send(FileOp::Delete(task.clone()).into()).await?;
} }
} }
self.succ(task.id) self.succ(task.id)
@ -265,7 +252,7 @@ impl File {
task.length = calculate_size(&task.target).await; task.length = calculate_size(&task.target).await;
self.prog.send(TaskProg::New(id, task.length))?; self.prog.send(TaskProg::New(id, task.length))?;
self.tx.send(FileOp::Trash(task)).await?; self.macro_.send(FileOp::Trash(task).into()).await?;
self.succ(id) self.succ(id)
} }

View file

@ -1,9 +1,10 @@
use crate::{file::FileOp, plugin::PluginOp, preload::PreloadOp};
#[derive(Debug)] #[derive(Debug)]
pub enum TaskOp { pub enum TaskOp {
File(Box<crate::file::FileOp>), File(Box<FileOp>),
Plugin(Box<crate::plugin::PluginOp>), Plugin(Box<PluginOp>),
Preload(Box<crate::preload::PreloadOp>), Preload(Box<PreloadOp>),
Process(Box<crate::process::ProcessOp>),
} }
impl TaskOp { impl TaskOp {
@ -12,7 +13,18 @@ impl TaskOp {
TaskOp::File(op) => op.id(), TaskOp::File(op) => op.id(),
TaskOp::Plugin(op) => op.id(), TaskOp::Plugin(op) => op.id(),
TaskOp::Preload(op) => op.id(), TaskOp::Preload(op) => op.id(),
TaskOp::Process(op) => op.id(),
} }
} }
} }
impl From<FileOp> for TaskOp {
fn from(op: FileOp) -> Self { Self::File(Box::new(op)) }
}
impl From<PluginOp> for TaskOp {
fn from(op: PluginOp) -> Self { Self::Plugin(Box::new(op)) }
}
impl From<PreloadOp> for TaskOp {
fn from(op: PreloadOp) -> Self { Self::Preload(Box::new(op)) }
}

View file

@ -3,19 +3,6 @@ use std::{ffi::OsString, mem};
use tokio::sync::oneshot; use tokio::sync::oneshot;
use yazi_plugin::external::ShellOpt; use yazi_plugin::external::ShellOpt;
#[derive(Debug)]
pub enum ProcessOp {
Open(ProcessOpOpen),
}
impl ProcessOp {
pub fn id(&self) -> usize {
match self {
Self::Open(op) => op.id,
}
}
}
#[derive(Debug)] #[derive(Debug)]
pub struct ProcessOpOpen { pub struct ProcessOpOpen {
pub id: usize, pub id: usize,
@ -27,12 +14,12 @@ pub struct ProcessOpOpen {
} }
impl From<&mut ProcessOpOpen> for ShellOpt { impl From<&mut ProcessOpOpen> for ShellOpt {
fn from(value: &mut ProcessOpOpen) -> Self { fn from(op: &mut ProcessOpOpen) -> Self {
Self { Self {
cmd: mem::take(&mut value.cmd), cmd: mem::take(&mut op.cmd),
args: mem::take(&mut value.args), args: mem::take(&mut op.args),
piped: false, piped: false,
orphan: value.orphan, orphan: op.orphan,
} }
} }
} }

View file

@ -28,7 +28,7 @@ impl Scheduler {
let (prog_tx, prog_rx) = mpsc::unbounded_channel(); let (prog_tx, prog_rx) = mpsc::unbounded_channel();
let scheduler = Self { let scheduler = Self {
file: Arc::new(File::new(prog_tx.clone())), file: Arc::new(File::new(macro_tx.clone(), prog_tx.clone())),
plugin: Arc::new(Plugin::new(prog_tx.clone())), plugin: Arc::new(Plugin::new(prog_tx.clone())),
preload: Arc::new(Preload::new(prog_tx.clone())), preload: Arc::new(Preload::new(prog_tx.clone())),
process: Arc::new(Process::new(prog_tx.clone())), process: Arc::new(Process::new(prog_tx.clone())),
@ -76,20 +76,19 @@ impl Scheduler {
Ok(fut) = micro.recv() => { Ok(fut) = micro.recv() => {
fut.await; fut.await;
} }
Ok(op) = macro_.recv() => {} Ok(op) = macro_.recv() => {
Ok((id, mut op)) = file.recv() => { let id = op.id();
if !running.read().exists(id) { if !running.read().exists(id) {
continue; continue;
} }
if let Err(e) = file.work(&mut op).await {
prog.send(TaskProg::Fail(id, format!("Failed to work on this task: {:?}", e))).ok(); let result = match op {
} TaskOp::File(mut op) => file.work(&mut op).await,
} TaskOp::Plugin(mut op) => todo!(),
Ok((id, mut op)) = plugin.recv() => { TaskOp::Preload(mut op) => todo!(),
if !running.read().exists(id) { };
continue;
} if let Err(e) = result {
if let Err(e) = plugin.work(&mut op).await {
prog.send(TaskProg::Fail(id, format!("Failed to work on this task: {:?}", e))).ok(); prog.send(TaskProg::Fail(id, format!("Failed to work on this task: {:?}", e))).ok();
} }
} }