This commit is contained in:
949036910@qq.com
2026-05-28 22:29:21 +08:00
parent fa467b65b6
commit b50a463296
2 changed files with 49 additions and 14 deletions

View File

@@ -26,6 +26,7 @@ use serde_json::Value;
use std::collections::HashMap; use std::collections::HashMap;
use tauri::{AppHandle, Emitter, Window}; use tauri::{AppHandle, Emitter, Window};
use tokio::sync::{mpsc, watch, Mutex}; use tokio::sync::{mpsc, watch, Mutex};
use tokio::time::{timeout, Duration};
use crate::app_config::AppConfig; use crate::app_config::AppConfig;
use crate::js_runtime::PipelineEvent; use crate::js_runtime::PipelineEvent;
@@ -102,6 +103,14 @@ pub(crate) fn spawn_event_pump(
}) })
} }
async fn finish_event_pump(mut pump: tokio::task::JoinHandle<()>) {
// 正常情况下 sender drop 后 pump 会自行退出;超时则强制结束,避免前端 invoke 卡住。
if timeout(Duration::from_millis(300), &mut pump).await.is_err() {
pump.abort();
let _ = pump.await;
}
}
/// 按脚本名跑:去后端拉 `scriptName`,包装 runnerspawn 子进程,回送结果。 /// 按脚本名跑:去后端拉 `scriptName`,包装 runnerspawn 子进程,回送结果。
#[tauri::command] #[tauri::command]
pub async fn run_nodejs_script( pub async fn run_nodejs_script(
@@ -136,7 +145,7 @@ pub async fn run_nodejs_script(
let result = result?; let result = result?;
pump.await.ok(); finish_event_pump(pump).await;
Ok(result) Ok(result)
} }
@@ -175,7 +184,7 @@ pub async fn run_nodejs_script_source(
let result = result?; let result = result?;
pump.await.ok(); finish_event_pump(pump).await;
Ok(result) Ok(result)
} }

View File

@@ -29,6 +29,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::process::Command; use tokio::process::Command;
use tokio::sync::mpsc::UnboundedSender; use tokio::sync::mpsc::UnboundedSender;
use tokio::sync::watch; use tokio::sync::watch;
use tokio::time::{timeout, Duration};
use uuid::Uuid; use uuid::Uuid;
use crate::js_runtime::PipelineEvent; use crate::js_runtime::PipelineEvent;
@@ -431,7 +432,7 @@ async fn run_script_bundle(
// stderr → 解析 __EVT__<json> → PipelineEvent // stderr → 解析 __EVT__<json> → PipelineEvent
let stderr = child.stderr.take().expect("stderr piped"); let stderr = child.stderr.take().expect("stderr piped");
let events_tx = events.clone(); let events_tx = events.clone();
let stderr_task = tokio::spawn(async move { let mut stderr_task = tokio::spawn(async move {
let mut reader = BufReader::new(stderr).lines(); let mut reader = BufReader::new(stderr).lines();
let mut tail = String::new(); let mut tail = String::new();
while let Ok(Some(line)) = reader.next_line().await { while let Ok(Some(line)) = reader.next_line().await {
@@ -488,12 +489,11 @@ async fn run_script_bundle(
// stdout → 找 __RESULT__ 那一行 // stdout → 找 __RESULT__ 那一行
let stdout = child.stdout.take().expect("stdout piped"); let stdout = child.stdout.take().expect("stdout piped");
let events_tx2 = events.clone(); let events_tx2 = events.clone();
let stdout_task = tokio::spawn(async move { let mut stdout_task = tokio::spawn(async move {
let mut reader = BufReader::new(stdout).lines(); let mut reader = BufReader::new(stdout).lines();
let mut result: Option<String> = None;
while let Ok(Some(line)) = reader.next_line().await { while let Ok(Some(line)) = reader.next_line().await {
if let Some(rest) = line.strip_prefix("__RESULT__") { if let Some(rest) = line.strip_prefix("__RESULT__") {
result = Some(rest.to_string()); return Some(rest.to_string());
} else if !line.is_empty() { } else if !line.is_empty() {
// 用户 console.log 的内容也透传给前端 // 用户 console.log 的内容也透传给前端
let _ = events_tx2.send(PipelineEvent::Log { let _ = events_tx2.send(PipelineEvent::Log {
@@ -503,22 +503,48 @@ async fn run_script_bundle(
}); });
} }
} }
result None
}); });
// 与取消信号竞争等待;取消时立即杀掉子进程 // 与取消信号、__RESULT__ 竞争等待:
let (status, cancelled) = tokio::select! { // - 收到 __RESULT__ 后不再等待脚本自然退出,主动结束子进程,避免 invoke 卡住。
status = child.wait() => (status?, false), let (status, cancelled, result_line, stdout_done) = tokio::select! {
result = &mut stdout_task => {
let result_line = result.unwrap_or(None);
let _ = child.start_kill();
let status = child.wait().await?;
(status, false, result_line, true)
}
status = child.wait() => (status?, false, None, false),
_ = wait_for_cancel(cancel_rx) => { _ = wait_for_cancel(cancel_rx) => {
let _ = child.start_kill(); let _ = child.start_kill();
let status = child.wait().await?; let status = child.wait().await?;
(status, true) (status, true, None, false)
} }
}; };
let (stderr_tail, result_line) = tokio::join!(stderr_task, stdout_task); let stderr_tail = match timeout(Duration::from_millis(500), &mut stderr_task).await {
let stderr_tail = stderr_tail.unwrap_or_default(); Ok(joined) => joined.unwrap_or_default(),
let result_line = result_line.unwrap_or(None); Err(_) => {
stderr_task.abort();
let _ = stderr_task.await;
String::new()
}
};
let result_line_fallback = if stdout_done {
None
} else {
match timeout(Duration::from_millis(500), &mut stdout_task).await {
Ok(joined) => joined.unwrap_or(None),
Err(_) => {
stdout_task.abort();
let _ = stdout_task.await;
None
}
}
};
let result_line = result_line.or(result_line_fallback);
let _ = tokio::fs::remove_dir_all(&bundle_dir).await; let _ = tokio::fs::remove_dir_all(&bundle_dir).await;