From 153d163a239f9eb9722e1ff3806156ab7c0743bd Mon Sep 17 00:00:00 2001 From: lifegpc Date: Mon, 20 Jun 2022 01:04:58 +0000 Subject: [PATCH] Update --- src/downloader/downloader.rs | 43 ++++++++++++++++++++++++ src/downloader/pd_file/file.rs | 17 +++++++++- src/downloader/pd_file/part_status.rs | 18 ++++++++++ src/downloader/tasks.rs | 48 +++++++++++++++++++++++---- 4 files changed, 118 insertions(+), 8 deletions(-) diff --git a/src/downloader/downloader.rs b/src/downloader/downloader.rs index 1daa786..e1dcd9f 100644 --- a/src/downloader/downloader.rs +++ b/src/downloader/downloader.rs @@ -845,6 +845,49 @@ async fn test_downloader_dropped() { } } +#[proc_macros::async_timeout_test(120s)] +#[tokio::test(flavor = "multi_thread")] +async fn test_multi_downloader() { + let p = Path::new("./test"); + if !p.exists() { + let re = create_dir("./test"); + assert!(re.is_ok() || p.exists()); + } + let url = "https://i.pximg.net/img-original/img/2022/06/18/08/19/18/99124570_p0.jpg"; + let pb = p.join("99124570_p0.jpg"); + { + let mut file_name = pb.file_name().unwrap().to_owned(); + file_name.push(".pd"); + let mut pdf = pb.clone(); + pdf.set_file_name(file_name); + if pdf.exists() { + remove_file(&pdf).unwrap(); + } + LocalFile::create(&pdf).unwrap(); + assert!(pdf.exists()); + } + let downloader = Downloader::::new( + url, + json::object! {"referer": "https://www.pixiv.net/"}, + Some(&pb), + Some(true), + ) + .unwrap(); + match downloader { + DownloaderResult::Ok(v) => { + assert_eq!(v.is_created(), true); + v.disable_progress_bar(); + v.enable_multiple_download(); + v.download(); + v.join().await.unwrap(); + assert_eq!(v.is_downloaded(), true); + } + DownloaderResult::Canceled => { + panic!("This should not happened.") + } + } +} + #[proc_macros::async_timeout_test(120s)] #[tokio::test(flavor = "multi_thread")] async fn test_failed_multi_downloader() { diff --git a/src/downloader/pd_file/file.rs b/src/downloader/pd_file/file.rs index ab0a01d..9d2720a 100644 --- a/src/downloader/pd_file/file.rs +++ b/src/downloader/pd_file/file.rs @@ -209,12 +209,13 @@ impl PdFile { data: &mut Option>, ) -> Option { let datas = self.part_datas.get_ref(); - let index = 0; + let mut index = 0; for d in datas.iter() { if d.is_waited() { data.replace(Arc::clone(d)); return Some(index); } + index += 1; } None } @@ -278,10 +279,24 @@ impl PdFile { /// Returns true if all parts are downloaded. pub fn is_all_part_downloaded(&self) -> bool { + #[cfg(test)] + let mut index = 0u64; for part in self.part_datas.get_ref().iter() { if !part.is_downloaded() { + #[cfg(test)] + { + println!("Index {} is not downloaded: {}", index, part.status()); + } return false; } + #[cfg(test)] + { + index += 1; + } + } + #[cfg(test)] + { + println!("All parts downloaded."); } return true; } diff --git a/src/downloader/pd_file/part_status.rs b/src/downloader/pd_file/part_status.rs index e2d98e6..96da307 100644 --- a/src/downloader/pd_file/part_status.rs +++ b/src/downloader/pd_file/part_status.rs @@ -126,6 +126,24 @@ impl PdFilePartStatus { } } + #[inline] + /// Set the status to waited. + pub fn set_waited(&self) -> Result<(), PdFileError> { + self.set_status(PdFilePartStatus2::Waited) + } + + #[inline] + /// Set the status to downloading. + pub fn set_downloading(&self) -> Result<(), PdFileError> { + self.set_status(PdFilePartStatus2::Downloading) + } + + #[inline] + /// Set the status to downloaded. + pub fn set_downloaded(&self) -> Result<(), PdFileError> { + self.set_status(PdFilePartStatus2::Downloaded) + } + #[inline] /// Increase the size of the part /// * `size` - The size to increase diff --git a/src/downloader/tasks.rs b/src/downloader/tasks.rs index d08492f..ca1ed1a 100644 --- a/src/downloader/tasks.rs +++ b/src/downloader/tasks.rs @@ -2,6 +2,7 @@ use super::downloader::DownloaderInternal; use super::downloader::GetTargetFileName; use super::error::DownloaderError; use super::pd_file::PdFilePartStatus; +use crate::concat_error; use crate::ext::atomic::AtomicQuick; use crate::ext::io::ClearFile; use crate::ext::replace::ReplaceWith2; @@ -203,19 +204,38 @@ pub async fn create_download_tasks_multi< d: Arc>, pd: Arc, index: usize, +) -> Result<(), DownloaderError> { + let mut re = create_download_tasks_multi_internal(d, Arc::clone(&pd), index).await; + if re.is_err() { + concat_error!(re, pd.set_waited(), DownloaderError); + } + re +} + +/// Create a new download task in multiple thread mode. +pub async fn create_download_tasks_multi_internal< + T: Seek + Write + Send + Sync + ClearFile + GetTargetFileName, +>( + d: Arc>, + pd: Arc, + index: usize, ) -> Result<(), DownloaderError> { let part_size = d.get_part_size() as u64; let file_size = d.pd.get_file_size(); let start = part_size * (index as u64); let end = std::cmp::min(start + part_size - 1, file_size); let mut headers = d.headers.deref().clone(); - headers.insert(String::from("Range"), format!("{}-{}", start, end)); + headers.insert(String::from("Range"), format!("bytes={}-{}", start, end)); let result = d .client .get(d.url.deref().clone(), headers) .await .try_err(gettext("Failed to get url."))?; let status = result.status(); + #[cfg(test)] + { + println!("Index {}: HTTP Status {}", index, status); + } if status == 200 || status == 416 { d.fallback_to_simp(); d.tasks.replace_with2(Vec::new()); @@ -226,11 +246,7 @@ pub async fn create_download_tasks_multi< if status.as_u16() != 206 { return Err(DownloaderError::from(status)); } - let re = handle_download(d, result, Some(pd), Some(index)).await; - if re.is_err() { - // #TODO - } - re + handle_download(d, result, Some(pd), Some(index)).await } /// Handle download process @@ -266,6 +282,10 @@ pub async fn handle_download { @@ -302,6 +322,11 @@ pub async fn handle_download { + data.as_ref().unwrap().set_downloading().unwrap(); let task = tokio::spawn(create_download_tasks_multi( Arc::clone(d), data.unwrap(), @@ -445,7 +471,15 @@ pub async fn check_tasks< add_new_multi_tasks(&d).await?; } if d.pd.is_all_part_downloaded() { - need_break = true; + match d.pd.complete() { + Ok(_) => { + need_break = true; + d.set_downloaded(); + } + Err(e) => { + println!("{}", e); + } + } } } }