This commit is contained in:
sxyazi 2024-03-12 20:52:09 +08:00
parent 9e4c70138c
commit 0233a0e952
No known key found for this signature in database
14 changed files with 139 additions and 103 deletions

1
Cargo.lock generated
View file

@ -2863,6 +2863,7 @@ dependencies = [
"regex", "regex",
"tokio", "tokio",
"tokio-stream", "tokio-stream",
"tokio-util",
"tracing", "tracing",
"trash", "trash",
"yazi-adaptor", "yazi-adaptor",

View file

@ -1,4 +1,4 @@
use std::{collections::HashMap, ffi::{OsStr, OsString}, io::{stdout, BufWriter, Write}, path::PathBuf}; use std::{collections::HashMap, ffi::OsStr, io::{stdout, BufWriter, Write}, path::PathBuf};
use anyhow::{anyhow, Result}; use anyhow::{anyhow, Result};
use tokio::{fs::{self, OpenOptions}, io::{stdin, AsyncReadExt, AsyncWriteExt}}; use tokio::{fs::{self, OpenOptions}, io::{stdin, AsyncReadExt, AsyncWriteExt}};

View file

@ -84,7 +84,7 @@ impl Manager {
if targets.is_empty() { if targets.is_empty() {
return; return;
} else if !opt.interactive { } else if !opt.interactive {
return tasks.process_from_files(&opt.hovered, &targets); return tasks.process_from_files(opt.hovered, targets);
} }
let openers: Vec<_> = OPEN.common_openers(&targets).into_iter().cloned().collect(); let openers: Vec<_> = OPEN.common_openers(&targets).into_iter().cloned().collect();

View file

@ -5,7 +5,10 @@ use crate::tasks::Tasks;
impl Tasks { impl Tasks {
pub fn open_with(&mut self, opt: impl TryInto<OpenWithOpt>) { pub fn open_with(&mut self, opt: impl TryInto<OpenWithOpt>) {
if let Ok(opt) = opt.try_into() { if let Ok(opt) = opt.try_into() {
self.process_from_opener(&opt.opener, &opt.targets); self.process_from_opener(
opt.opener,
opt.targets.into_iter().map(|u| u.into_os_string()).collect(),
);
} }
} }
} }

View file

@ -5,8 +5,7 @@ use crate::tasks::Tasks;
impl Tasks { impl Tasks {
pub fn process_exec(&mut self, opt: impl TryInto<ProcessExecOpt>) { pub fn process_exec(&mut self, opt: impl TryInto<ProcessExecOpt>) {
if let Ok(opt) = opt.try_into() { if let Ok(opt) = opt.try_into() {
// FIXME self.scheduler.process_open(opt.opener, opt.args, Some(opt.done));
// self.process_from_opener(&opt.opener, &opt.targets);
} }
} }
} }

View file

@ -1,4 +1,4 @@
use std::{collections::HashMap, ffi::OsStr}; use std::{collections::HashMap, ffi::OsString, mem};
use yazi_config::{open::Opener, OPEN}; use yazi_config::{open::Opener, OPEN};
use yazi_shared::fs::Url; use yazi_shared::fs::Url;
@ -6,25 +6,36 @@ use yazi_shared::fs::Url;
use super::Tasks; use super::Tasks;
impl Tasks { impl Tasks {
pub fn process_from_files(&self, hovered: &Url, targets: &[(Url, String)]) { pub fn process_from_files(&self, hovered: Url, targets: Vec<(Url, String)>) {
let mut openers = HashMap::new(); let mut openers = HashMap::new();
for (url, mime) in targets { for (url, mime) in targets {
if let Some(opener) = OPEN.openers(url, mime).and_then(|o| o.first().copied()) { if let Some(opener) = OPEN.openers(&url, mime).and_then(|o| o.first().copied()) {
openers.entry(opener).or_insert_with(|| vec![hovered]).push(url); openers.entry(opener).or_insert_with(|| vec![hovered.clone()]).push(url);
} }
} }
for (opener, args) in openers { for (opener, args) in openers {
self.process_from_opener(opener, &args); self.process_from_opener(
opener.clone(),
args.into_iter().map(|u| u.into_os_string()).collect(),
);
} }
} }
pub fn process_from_opener(&self, opener: &Opener, args: &[impl AsRef<OsStr>]) { pub fn process_from_opener(&self, opener: Opener, mut args: Vec<OsString>) {
if opener.spread { if opener.spread {
self.scheduler.process_open(opener, args); self.scheduler.process_open(opener, args, None);
return; return;
} }
for target in args.iter().skip(1) { if args.is_empty() {
self.scheduler.process_open(opener, &[&args[0], target]); return;
}
if args.len() == 2 {
self.scheduler.process_open(opener, args, None);
return;
}
let hovered = mem::take(&mut args[0]);
for target in args.into_iter().skip(1) {
self.scheduler.process_open(opener.clone(), vec![hovered.clone(), target], None);
} }
} }
} }

View file

@ -1,8 +1,8 @@
use std::{collections::{HashMap, HashSet}, ffi::OsStr, mem, sync::Arc, time::Duration}; use std::{collections::{HashMap, HashSet}, mem, sync::Arc, time::Duration};
use tokio::time::sleep; use tokio::time::sleep;
use tracing::debug; use tracing::debug;
use yazi_config::{manager::SortBy, open::Opener, plugin::{PluginRule, MAX_PRELOADERS}, OPEN, PLUGIN}; use yazi_config::{manager::SortBy, plugin::{PluginRule, MAX_PRELOADERS}, PLUGIN};
use yazi_plugin::ValueSendable; use yazi_plugin::ValueSendable;
use yazi_scheduler::{Scheduler, TaskSummary}; use yazi_scheduler::{Scheduler, TaskSummary};
use yazi_shared::{emit, event::Cmd, fs::{File, Url}, term::Term, Layer, MIME_DIR}; use yazi_shared::{emit, event::Cmd, fs::{File, Url}, term::Term, Layer, MIME_DIR};

View file

@ -156,6 +156,7 @@ impl<'a> Executor<'a> {
on!(inspect); on!(inspect);
on!(cancel); on!(cancel);
on!(open_with); on!(open_with);
on!(process_exec);
#[allow(clippy::single_match)] #[allow(clippy::single_match)]
match cmd.name.as_str() { match cmd.name.as_str() {

View file

@ -1,14 +1,14 @@
use std::ffi::OsString; use std::ffi::OsString;
use tokio::sync::oneshot;
use yazi_config::open::Opener;
use yazi_shared::event::Cmd; use yazi_shared::event::Cmd;
// --- Exec // --- Exec
#[derive(Default)]
pub struct ProcessExecOpt { pub struct ProcessExecOpt {
pub cmd: OsString, pub opener: Opener,
pub args: Vec<OsString>, pub args: Vec<OsString>,
pub block: bool, pub done: oneshot::Sender<()>,
pub orphan: bool,
} }
impl TryFrom<Cmd> for ProcessExecOpt { impl TryFrom<Cmd> for ProcessExecOpt {

View file

@ -25,6 +25,7 @@ parking_lot = "^0"
regex = "^1" regex = "^1"
tokio = { version = "^1", features = [ "parking_lot", "rt-multi-thread" ] } tokio = { version = "^1", features = [ "parking_lot", "rt-multi-thread" ] }
tokio-stream = "^0" tokio-stream = "^0"
tokio-util = "^0"
# Logging # Logging
tracing = { version = "^0", features = [ "max_level_debug", "release_max_level_warn" ] } tracing = { version = "^0", features = [ "max_level_debug", "release_max_level_warn" ] }

View file

@ -1,21 +1,45 @@
use std::ffi::OsString; use std::ffi::OsString;
use tokio::sync::oneshot; use tokio_util::sync::CancellationToken;
use super::ShellOpt; use super::ShellOpt;
#[derive(Debug)] #[derive(Debug)]
pub struct ProcessOpOpen { pub struct ProcessOpOrphan {
pub id: usize, pub id: usize,
pub cmd: OsString, pub cmd: OsString,
pub args: Vec<OsString>, pub args: Vec<OsString>,
pub block: bool,
pub orphan: bool,
pub cancel: oneshot::Sender<()>,
} }
impl From<ProcessOpOpen> for ShellOpt { impl From<ProcessOpOrphan> for ShellOpt {
fn from(op: ProcessOpOpen) -> Self { fn from(op: ProcessOpOrphan) -> Self {
Self { cmd: op.cmd, args: op.args, piped: false, orphan: op.orphan } Self { cmd: op.cmd, args: op.args, piped: false, orphan: true }
}
}
#[derive(Debug)]
pub struct ProcessOpBlock {
pub id: usize,
pub cmd: OsString,
pub args: Vec<OsString>,
}
impl From<ProcessOpBlock> for ShellOpt {
fn from(op: ProcessOpBlock) -> Self {
Self { cmd: op.cmd, args: op.args, piped: false, orphan: false }
}
}
#[derive(Debug)]
pub struct ProcessOpBg {
pub id: usize,
pub cmd: OsString,
pub args: Vec<OsString>,
pub ct: CancellationToken,
}
impl From<ProcessOpBg> for ShellOpt {
fn from(op: ProcessOpBg) -> Self {
Self { cmd: op.cmd, args: op.args, piped: true, orphan: false }
} }
} }

View file

@ -3,7 +3,7 @@ use tokio::{io::{AsyncBufReadExt, BufReader}, select, sync::mpsc};
use yazi_proxy::{AppProxy, HIDER}; use yazi_proxy::{AppProxy, HIDER};
use yazi_shared::Defer; use yazi_shared::Defer;
use super::{ProcessOpOpen, ShellOpt}; use super::{ProcessOpBg, ProcessOpBlock, ProcessOpOrphan, ShellOpt};
use crate::TaskProg; use crate::TaskProg;
pub struct Process { pub struct Process {
@ -13,15 +13,44 @@ pub struct Process {
impl Process { impl Process {
pub fn new(prog: mpsc::UnboundedSender<TaskProg>) -> Self { Self { prog } } pub fn new(prog: mpsc::UnboundedSender<TaskProg>) -> Self { Self { prog } }
pub async fn open(&self, mut task: ProcessOpOpen) -> Result<()> { pub async fn orphan(&self, task: ProcessOpOrphan) -> Result<()> {
if task.block { let id = task.id;
return self.open_block(task).await; match super::shell(task.into()) {
Ok(_) => self.succ(id)?,
Err(e) => {
self.prog.send(TaskProg::New(id, 0))?;
self.fail(id, format!("Failed to spawn process: {e}"))?;
}
} }
if task.orphan { Ok(())
return self.open_orphan(task).await;
} }
pub async fn block(&self, task: ProcessOpBlock) -> Result<()> {
let _permit = HIDER.acquire().await.unwrap();
let _defer = Defer::new(AppProxy::resume);
AppProxy::stop().await;
let (id, cmd) = (task.id, task.cmd.clone());
let result = super::shell(task.into());
if let Err(e) = result {
AppProxy::notify_warn(&cmd.to_string_lossy(), &format!("Failed to spawn process: {e}"));
return self.succ(id);
}
let status = result.unwrap().wait().await?;
if !status.success() {
let content = match status.code() {
Some(code) => format!("Process exited with status code: {code}"),
None => "Process terminated by signal".to_string(),
};
AppProxy::notify_warn(&cmd.to_string_lossy(), &content);
}
self.succ(id)
}
pub async fn bg(&self, task: ProcessOpBg) -> Result<()> {
self.prog.send(TaskProg::New(task.id, 0))?; self.prog.send(TaskProg::New(task.id, 0))?;
let mut child = let mut child =
super::shell(ShellOpt { cmd: task.cmd, args: task.args, piped: true, ..Default::default() })?; super::shell(ShellOpt { cmd: task.cmd, args: task.args, piped: true, ..Default::default() })?;
@ -30,7 +59,7 @@ impl Process {
let mut stderr = BufReader::new(child.stderr.take().unwrap()).lines(); let mut stderr = BufReader::new(child.stderr.take().unwrap()).lines();
loop { loop {
select! { select! {
_ = task.cancel.closed() => { _ = task.ct.cancelled() => {
child.start_kill().ok(); child.start_kill().ok();
break; break;
} }
@ -56,43 +85,6 @@ impl Process {
self.prog.send(TaskProg::Adv(task.id, 1, 0))?; self.prog.send(TaskProg::Adv(task.id, 1, 0))?;
self.succ(task.id) self.succ(task.id)
} }
async fn open_block(&self, task: ProcessOpOpen) -> Result<()> {
let _permit = HIDER.acquire().await.unwrap();
let _defer = Defer::new(AppProxy::resume);
AppProxy::stop().await;
let (id, cmd) = (task.id, task.cmd.clone());
let result = super::shell(task.into());
if let Err(e) = result {
AppProxy::notify_warn(&cmd.to_string_lossy(), &format!("Failed to spawn process: {e}"));
return self.succ(id);
}
let status = result.unwrap().wait().await?;
if !status.success() {
let content = match status.code() {
Some(code) => format!("Process exited with status code: {code}"),
None => "Process terminated by signal".to_string(),
};
AppProxy::notify_warn(&cmd.to_string_lossy(), &content);
}
self.succ(id)
}
async fn open_orphan(&self, task: ProcessOpOpen) -> Result<()> {
let id = task.id;
match super::shell(task.into()) {
Ok(_) => self.succ(id)?,
Err(e) => {
self.prog.send(TaskProg::New(id, 0))?;
self.fail(id, format!("Failed to spawn process: {e}"))?;
}
}
Ok(())
}
} }
impl Process { impl Process {

View file

@ -12,11 +12,6 @@ pub struct ShellOpt {
} }
impl ShellOpt { impl ShellOpt {
pub fn with_piped(mut self) -> Self {
self.piped = true;
self
}
#[inline] #[inline]
fn stdio(&self) -> Stdio { fn stdio(&self) -> Stdio {
if self.orphan { if self.orphan {

View file

@ -1,14 +1,15 @@
use std::{ffi::OsStr, sync::Arc, time::Duration}; use std::{ffi::OsString, sync::Arc, time::Duration};
use futures::{future::BoxFuture, FutureExt}; use futures::{future::BoxFuture, FutureExt};
use parking_lot::Mutex; use parking_lot::Mutex;
use tokio::{fs, select, sync::{mpsc::{self, UnboundedReceiver}, oneshot}}; use tokio::{fs, select, sync::{mpsc::{self, UnboundedReceiver}, oneshot}};
use tokio_util::sync::CancellationToken;
use yazi_config::{open::Opener, plugin::PluginRule, TASKS}; use yazi_config::{open::Opener, plugin::PluginRule, TASKS};
use yazi_plugin::ValueSendable; use yazi_plugin::ValueSendable;
use yazi_shared::{fs::{unique_path, Url}, Throttle}; use yazi_shared::{fs::{unique_path, Url}, Throttle};
use super::{Ongoing, TaskProg, TaskStage}; use super::{Ongoing, TaskProg, TaskStage};
use crate::{file::{File, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash}, plugin::{Plugin, PluginOpEntry}, preload::{Preload, PreloadOpRule, PreloadOpSize}, process::{Process, ProcessOpOpen}, TaskKind, TaskOp, HIGH, LOW, NORMAL}; use crate::{file::{File, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash}, plugin::{Plugin, PluginOpEntry}, preload::{Preload, PreloadOpRule, PreloadOpSize}, process::{Process, ProcessOpBg, ProcessOpBlock, ProcessOpOrphan}, TaskKind, TaskOp, HIGH, LOW, NORMAL};
pub struct Scheduler { pub struct Scheduler {
pub file: Arc<File>, pub file: Arc<File>,
@ -332,46 +333,54 @@ impl Scheduler {
} }
} }
pub fn process_open(&self, opener: &Opener, args: &[impl AsRef<OsStr>]) { pub fn process_open(
&self,
opener: Opener,
args: Vec<OsString>,
done: Option<oneshot::Sender<()>>,
) {
let name = { let name = {
let s = format!("Run `{}`", opener.run); let args = args.iter().map(|a| a.to_string_lossy()).collect::<Vec<_>>().join(" ");
let args = args.iter().map(|a| a.as_ref().to_string_lossy()).collect::<Vec<_>>().join(" "); if args.is_empty() {
if args.is_empty() { s } else { format!("{s} with `{args}`") } format!("Run {:?}", opener.run)
} else {
format!("Run {:?} with `{args}`", opener.run)
}
}; };
let ct = CancellationToken::new();
let mut ongoing = self.ongoing.lock(); let mut ongoing = self.ongoing.lock();
let id = ongoing.add(TaskKind::User, name);
let (cancel_tx, mut cancel_rx) = oneshot::channel(); let id = ongoing.add(TaskKind::User, name);
ongoing.hooks.insert(id, { ongoing.hooks.insert(id, {
let ct = ct.clone();
let ongoing = self.ongoing.clone(); let ongoing = self.ongoing.clone();
Box::new(move |canceled: bool| { Box::new(move |canceled: bool| {
async move { async move {
if canceled {
cancel_rx.close();
}
ongoing.lock().try_remove(id, TaskStage::Hooked); ongoing.lock().try_remove(id, TaskStage::Hooked);
if canceled {
ct.cancel();
}
if let Some(tx) = done {
tx.send(()).ok();
}
} }
.boxed() .boxed()
}) })
}); });
let args = args.iter().map(|a| a.as_ref().to_os_string()).collect::<Vec<_>>(); // FIXME: use micro instead
tokio::spawn({ tokio::spawn({
let process = self.process.clone(); let process = self.process.clone();
let opener = opener.clone(); let opener = opener.clone();
async move { async move {
process if opener.orphan {
.open(ProcessOpOpen { process.orphan(ProcessOpOrphan { id, cmd: opener.run.into(), args }).await.ok();
id, } else if opener.block {
cmd: opener.run.into(), process.block(ProcessOpBlock { id, cmd: opener.run.into(), args }).await.ok();
args, } else {
block: opener.block, process.bg(ProcessOpBg { id, cmd: opener.run.into(), args, ct }).await.ok();
orphan: opener.orphan, }
cancel: cancel_tx,
})
.await
.ok();
} }
}); });
} }