diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index e490c43..da73f07 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -301,6 +301,7 @@ dependencies = [ "parking_lot 0.12.4", "prost", "rand 0.9.1", + "rayon", "reqwest", "reqwest-middleware", "reqwest-retry", @@ -678,6 +679,25 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-deque" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -3360,6 +3380,26 @@ version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20675572f6f24e9e76ef639bc5552774ed45f1c30e2951e1e99c59888861c539" +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "redox_syscall" version = "0.2.16" diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 7d6e18c..501d0c0 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -60,6 +60,7 @@ memchr = { version = "2.7.5" } md-5 = { version = "0.10.6" } rand = { version = "0.9.1" } base64 = { version = "0.22.1" } +rayon = { version = "1.12.0" } [profile.release] strip = true diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index 8430d4b..21c7a5c 100644 --- a/src-tauri/src/commands.rs +++ b/src-tauri/src/commands.rs @@ -11,6 +11,7 @@ use tracing::instrument; use crate::{ config::Config, + downloader::download_task::RestoredDownloadTask, errors::{CommandError, CommandResult}, extensions::AppHandleExt, logger, @@ -310,13 +311,14 @@ pub fn restart_download_task(app: AppHandle, params: RestartDownloadTaskParams) #[tauri::command(async)] #[specta::specta] #[instrument(level = "error", skip_all)] -pub fn restore_download_tasks(app: AppHandle) -> CommandResult<()> { +pub async fn restore_download_tasks(app: AppHandle) -> CommandResult> { let download_manager = app.get_download_manager(); - download_manager + let restored_tasks = download_manager .restore_download_tasks() + .await .map_err(|err| CommandError::from("恢复下载任务失败", err))?; tracing::debug!("恢复下载任务成功"); - Ok(()) + Ok(restored_tasks) } #[tauri::command(async)] diff --git a/src-tauri/src/downloader/download_manager.rs b/src-tauri/src/downloader/download_manager.rs index 91ca0c8..30ccd71 100644 --- a/src-tauri/src/downloader/download_manager.rs +++ b/src-tauri/src/downloader/download_manager.rs @@ -10,12 +10,17 @@ use std::{ use eyre::{WrapErr, eyre}; use parking_lot::RwLock; +use rayon::prelude::*; use tauri::{AppHandle, Manager}; use tauri_specta::Event; -use tokio::sync::Semaphore; -use tracing::instrument; +use tokio::{ + sync::{Semaphore, oneshot}, + task::JoinSet, +}; +use tracing::{Instrument, instrument}; use crate::{ + downloader::download_progress::DownloadProgress, events::DownloadEvent, extensions::{AppHandleExt, EyreReportToMessage}, types::{ @@ -24,7 +29,7 @@ use crate::{ }, }; -use super::{download_progress::DownloadProgress, download_task::DownloadTask}; +use super::download_task::{DownloadTask, RestoredDownloadTask}; pub struct DownloadManager { pub app: AppHandle, @@ -58,41 +63,92 @@ impl DownloadManager { } #[instrument(level = "error", skip_all)] - pub fn restore_download_tasks(&self) -> eyre::Result<()> { + pub async fn restore_download_tasks(&self) -> eyre::Result> { + struct TaskFile { + path: PathBuf, + content: String, + } + let task_dir = self.get_task_dir()?; std::fs::create_dir_all(&task_dir) .wrap_err(format!("创建下载任务目录`{}`失败", task_dir.display()))?; - let mut tasks = self.download_tasks.write(); + let mut join_set = JoinSet::new(); for entry in std::fs::read_dir(&task_dir)?.filter_map(Result::ok) { - let path = entry.path(); - let extension = path.extension().and_then(|s| s.to_str()); - if extension != Some("json") { - // 如果这个文件不是json则删除 - let _ = std::fs::remove_file(&path); - continue; - } + let read_task_file = async move { + let path = entry.path(); - let progress_json = std::fs::read_to_string(&path)?; + let extension = path.extension().and_then(|s| s.to_str()); + if extension != Some("json") { + // 如果这个文件不是json则删除 + let _ = tokio::fs::remove_file(path).await; + return None; + } - let progress: DownloadProgress = - if let Ok(progress) = serde_json::from_str(&progress_json) { - progress - } else { - // 如果这个json解析失败则删除 - let _ = std::fs::remove_file(&path); - continue; + let content = match tokio::fs::read_to_string(&path) + .await + .map_err(eyre::Report::from) + { + Ok(content) => content, + Err(err) => { + let err_title = format!("读取下载任务文件`{}`失败", path.display()); + let message = err.to_message(); + tracing::error!(err_title, message); + return None; + } }; - let new_task = DownloadTask::from_progress(self.app.clone(), progress); + Some(TaskFile { path, content }) + }; + + join_set.spawn(read_task_file.in_current_span()); + } + + let mut task_files = Vec::new(); + while let Some(join_result) = join_set.join_next().await { + let Ok(Some(task_file)) = join_result else { + continue; + }; + + task_files.push(task_file); + } + + let (progresses_sender, progresses_receiver) = oneshot::channel(); + rayon::spawn(move || { + let progresses = task_files + .into_par_iter() + .filter_map(|task_file| { + if let Ok(progress) = serde_json::from_str(&task_file.content) { + Some(progress) + } else { + // 如果这个json解析失败则删除 + let _ = std::fs::remove_file(&task_file.path); + None + } + }) + .collect(); + + let _ = progresses_sender.send(progresses); + }); + let progresses: Vec = progresses_receiver.await?; + + let mut tasks = self.download_tasks.write(); + let mut restored_tasks = Vec::new(); + + for progress in progresses { + let new_task = DownloadTask::from_progress(self.app.clone(), progress.clone()); + let state = *new_task.state_sender.borrow(); + let old_task = tasks.insert(new_task.task_id.clone(), new_task); if let Some(old_task) = old_task { // 如果同一个ID的下载任务已经存在,则取消旧的任务 old_task.cancel(); } + + restored_tasks.push(RestoredDownloadTask { state, progress }); } - Ok(()) + Ok(restored_tasks) } pub fn create_download_tasks(&self, params: &CreateDownloadTaskParams) { diff --git a/src-tauri/src/downloader/download_task.rs b/src-tauri/src/downloader/download_task.rs index f6227c2..0f610f1 100644 --- a/src-tauri/src/downloader/download_task.rs +++ b/src-tauri/src/downloader/download_task.rs @@ -1,7 +1,9 @@ -use std::{sync::Arc, time::Duration}; +use std::{sync::Arc, time::Duration}; use eyre::WrapErr; use parking_lot::RwLock; +use serde::Serialize; +use specta::Type; use tauri::{AppHandle, Manager}; use tauri_specta::Event; use tokio::{ @@ -33,6 +35,12 @@ pub struct DownloadTask { pub progress: RwLock, } +#[derive(Debug, Clone, Serialize, Type)] +pub struct RestoredDownloadTask { + pub state: DownloadTaskState, + pub progress: DownloadProgress, +} + impl DownloadTask { #[allow(clippy::too_many_lines)] #[instrument(level = "error", skip_all)] @@ -168,7 +176,7 @@ impl DownloadTask { progress: RwLock::new(progress), }); - tauri::async_runtime::spawn(task.clone().process()); + tauri::async_runtime::spawn(task.clone().process(true)); tasks.push(task); } @@ -198,7 +206,7 @@ impl DownloadTask { progress: RwLock::new(progress), }); - tauri::async_runtime::spawn(task.clone().process()); + tauri::async_runtime::spawn(task.clone().process(false)); task } @@ -287,14 +295,14 @@ impl DownloadTask { up_uid = self.trace_fields.up_uid, ) )] - async fn process(self: Arc) { - let state = *self.state_sender.borrow(); - let progress = self.progress.read().clone(); - let _ = DownloadEvent::TaskCreate { state, progress }.emit(&self.app); + async fn process(self: Arc, emit_create_events: bool) { + if emit_create_events { + let state = *self.state_sender.borrow(); + let progress = self.progress.read().clone(); + let _ = DownloadEvent::TaskCreate { state, progress }.emit(&self.app); + } let mut state_receiver = self.state_sender.subscribe(); - state_receiver.mark_changed(); - let mut restart_receiver = self.restart_sender.subscribe(); let mut cancel_receiver = self.cancel_sender.subscribe(); let mut delete_receiver = self.delete_sender.subscribe(); diff --git a/src/bindings.ts b/src/bindings.ts index a57b762..582a145 100644 --- a/src/bindings.ts +++ b/src/bindings.ts @@ -125,7 +125,7 @@ async restartDownloadTasks(taskIds: string[]) : Promise { async restartDownloadTask(params: RestartDownloadTaskParams) : Promise { await TAURI_INVOKE("restart_download_task", { params }); }, -async restoreDownloadTasks() : Promise> { +async restoreDownloadTasks() : Promise> { try { return { status: "ok", data: await TAURI_INVOKE("restore_download_tasks") }; } catch (e) { @@ -429,6 +429,7 @@ export type RatingInBangumi = { count: number; score: number } export type RatingInBangumiFollow = { score: number; count: number } export type RecommendSeason = { cover: string; ep_count: string; id: number; season_url: string; subtitle: string; title: string; view: number } export type RestartDownloadTaskParams = { task_id: string; video_task_selected: boolean; audio_task_selected: boolean; merge_selected: boolean; embed_chapter_selected: boolean; embed_skip_selected: boolean; subtitle_task_selected: boolean; xml_danmaku_selected: boolean; ass_danmaku_selected: boolean; json_danmaku_selected: boolean; cover_task_selected: boolean; nfo_task_selected: boolean; json_task_selected: boolean; video_quality: VideoQuality; codec_type: CodecType; audio_quality: AudioQuality } +export type RestoredDownloadTask = { state: DownloadTaskState; progress: DownloadProgress } export type Rights = { bp: number; elec: number; download: number; movie: number; pay: number; hd5: number; no_reprint: number; autoplay: number; ugc_pay: number; is_cooperation: number; ugc_pay_preview: number; no_background: number; clean_mode: number; is_stein_gate: number; is_360: number; no_share: number; arc_pay: number; free_watch: number } export type RightsInBangumi = { allow_bp: number; allow_bp_rank: number; allow_download: number; allow_review: number; area_limit: number; ban_area_show: number; can_watch: number; copyright: string; forbid_pre: number; freya_white: number; is_cover_show: number; is_preview: number; only_vip_download: number; resource: string; watch_platform: number } export type RightsInBangumiEp = { allow_dm: number; allow_download: number; area_limit: number } diff --git a/src/panes/DownloadPane/DownloadPane.vue b/src/panes/DownloadPane/DownloadPane.vue index 17ebfd2..6fd32d2 100644 --- a/src/panes/DownloadPane/DownloadPane.vue +++ b/src/panes/DownloadPane/DownloadPane.vue @@ -33,7 +33,7 @@ onMounted(async () => { ...progress, state, percentage: 0, - stateIndicator: '', + stateIndicator: getStateIndicator(state), taskIndicator: '', } store.updateProgresses((progresses) => { @@ -48,21 +48,8 @@ onMounted(async () => { return } - let stateIndicator = '' - if (state === 'Pending') { - stateIndicator = '排队中' - } else if (state === 'Downloading') { - stateIndicator = '下载中' - } else if (state === 'Paused') { - stateIndicator = '已暂停' - } else if (state === 'Completed') { - stateIndicator = '下载完成' - } else if (state === 'Failed') { - stateIndicator = '下载失败' - } - progressData.state = state - progressData.stateIndicator = stateIndicator + progressData.stateIndicator = getStateIndicator(state) }) } else if (event === 'TaskSleeping') { const { task_id, remaining_sec } = data @@ -151,8 +138,41 @@ onMounted(async () => { const result = await commands.restoreDownloadTasks() if (result.status === 'error') { console.error(result.error) + return } + + store.updateProgresses((progresses) => { + for (const { state, progress } of result.data) { + const progressData: ProgressData = { + ...progress, + state, + percentage: 0, + stateIndicator: getStateIndicator(state), + taskIndicator: '', + } + + progresses.set(progress.task_id, progressData) + } + }) }) + +function getStateIndicator(state: DownloadTaskState) { + let stateIndicator = '' + + if (state === 'Pending') { + stateIndicator = '排队中' + } else if (state === 'Downloading') { + stateIndicator = '下载中' + } else if (state === 'Paused') { + stateIndicator = '已暂停' + } else if (state === 'Completed') { + stateIndicator = '下载完成' + } else if (state === 'Failed') { + stateIndicator = '下载失败' + } + + return stateIndicator +}