From 2fad9bcd1600366da342cd432cc75cdeb77ad1f5 Mon Sep 17 00:00:00 2001 From: sxyazi Date: Fri, 29 Dec 2023 02:02:54 +0800 Subject: [PATCH] .. --- yazi-scheduler/src/file/file.rs | 41 +++++++++++--------------------- yazi-scheduler/src/op.rs | 22 +++++++++++++---- yazi-scheduler/src/process/op.rs | 21 ++++------------ yazi-scheduler/src/scheduler.rs | 23 +++++++++--------- 4 files changed, 46 insertions(+), 61 deletions(-) diff --git a/yazi-scheduler/src/file/file.rs b/yazi-scheduler/src/file/file.rs index ff373512..84387de8 100644 --- a/yazi-scheduler/src/file/file.rs +++ b/yazi-scheduler/src/file/file.rs @@ -8,29 +8,16 @@ use yazi_config::TASKS; use yazi_shared::fs::{calculate_size, copy_with_progress, path_relative_to, Url}; use super::{FileOp, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash}; -use crate::TaskProg; +use crate::{TaskOp, TaskProg}; pub struct File { - tx: async_channel::Sender, - rx: async_channel::Receiver, - - prog: mpsc::UnboundedSender, + macro_: async_channel::Sender, + prog: mpsc::UnboundedSender, } impl File { - pub fn new(prog: mpsc::UnboundedSender) -> Self { - let (tx, rx) = async_channel::unbounded(); - 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 fn new(macro_: async_channel::Sender, prog: mpsc::UnboundedSender) -> Self { + Self { macro_, prog } } pub async fn work(&self, op: &mut FileOp) -> Result<()> { @@ -63,7 +50,7 @@ impl File { { self.log(task.id, format!("Paste task retry: {:?}", task))?; 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)?, } @@ -157,9 +144,9 @@ impl File { self.prog.send(TaskProg::New(id, meta.len()))?; if meta.is_file() { - self.tx.send(FileOp::Paste(task)).await?; + self.macro_.send(FileOp::Paste(task).into()).await?; } 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); } @@ -203,9 +190,9 @@ impl File { self.prog.send(TaskProg::New(task.id, meta.len()))?; 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() { - 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.tx.send(FileOp::Link(task)).await?; + self.macro_.send(FileOp::Link(task).into()).await?; self.succ(id) } @@ -229,7 +216,7 @@ impl File { let id = task.id; task.length = 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); } @@ -254,7 +241,7 @@ impl File { task.target = Url::from(entry.path()); task.length = 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) @@ -265,7 +252,7 @@ impl File { task.length = calculate_size(&task.target).await; 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) } diff --git a/yazi-scheduler/src/op.rs b/yazi-scheduler/src/op.rs index 147bb2eb..6afcbebe 100644 --- a/yazi-scheduler/src/op.rs +++ b/yazi-scheduler/src/op.rs @@ -1,9 +1,10 @@ +use crate::{file::FileOp, plugin::PluginOp, preload::PreloadOp}; + #[derive(Debug)] pub enum TaskOp { - File(Box), - Plugin(Box), - Preload(Box), - Process(Box), + File(Box), + Plugin(Box), + Preload(Box), } impl TaskOp { @@ -12,7 +13,18 @@ impl TaskOp { TaskOp::File(op) => op.id(), TaskOp::Plugin(op) => op.id(), TaskOp::Preload(op) => op.id(), - TaskOp::Process(op) => op.id(), } } } + +impl From for TaskOp { + fn from(op: FileOp) -> Self { Self::File(Box::new(op)) } +} + +impl From for TaskOp { + fn from(op: PluginOp) -> Self { Self::Plugin(Box::new(op)) } +} + +impl From for TaskOp { + fn from(op: PreloadOp) -> Self { Self::Preload(Box::new(op)) } +} diff --git a/yazi-scheduler/src/process/op.rs b/yazi-scheduler/src/process/op.rs index a9b31c03..8436f37a 100644 --- a/yazi-scheduler/src/process/op.rs +++ b/yazi-scheduler/src/process/op.rs @@ -3,19 +3,6 @@ use std::{ffi::OsString, mem}; use tokio::sync::oneshot; 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)] pub struct ProcessOpOpen { pub id: usize, @@ -27,12 +14,12 @@ pub struct ProcessOpOpen { } impl From<&mut ProcessOpOpen> for ShellOpt { - fn from(value: &mut ProcessOpOpen) -> Self { + fn from(op: &mut ProcessOpOpen) -> Self { Self { - cmd: mem::take(&mut value.cmd), - args: mem::take(&mut value.args), + cmd: mem::take(&mut op.cmd), + args: mem::take(&mut op.args), piped: false, - orphan: value.orphan, + orphan: op.orphan, } } } diff --git a/yazi-scheduler/src/scheduler.rs b/yazi-scheduler/src/scheduler.rs index 8f21c688..54473dc0 100644 --- a/yazi-scheduler/src/scheduler.rs +++ b/yazi-scheduler/src/scheduler.rs @@ -28,7 +28,7 @@ impl Scheduler { let (prog_tx, prog_rx) = mpsc::unbounded_channel(); 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())), preload: Arc::new(Preload::new(prog_tx.clone())), process: Arc::new(Process::new(prog_tx.clone())), @@ -76,20 +76,19 @@ impl Scheduler { Ok(fut) = micro.recv() => { fut.await; } - Ok(op) = macro_.recv() => {} - Ok((id, mut op)) = file.recv() => { + Ok(op) = macro_.recv() => { + let id = op.id(); if !running.read().exists(id) { continue; } - if let Err(e) = file.work(&mut op).await { - prog.send(TaskProg::Fail(id, format!("Failed to work on this task: {:?}", e))).ok(); - } - } - Ok((id, mut op)) = plugin.recv() => { - if !running.read().exists(id) { - continue; - } - if let Err(e) = plugin.work(&mut op).await { + + let result = match op { + TaskOp::File(mut op) => file.work(&mut op).await, + TaskOp::Plugin(mut op) => todo!(), + TaskOp::Preload(mut op) => todo!(), + }; + + if let Err(e) = result { prog.send(TaskProg::Fail(id, format!("Failed to work on this task: {:?}", e))).ok(); } }