Update

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::<LocalFile>::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() {

src/downloader/pd_file/file.rs

@@ -209,12 +209,13 @@ impl PdFile {
         data: &mut Option<Arc<PdFilePartStatus>>,
     ) -> Option<usize> {
         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;
     }

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

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<DownloaderInternal<T>>,
     pd: Arc<PdFilePartStatus>,
     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<DownloaderInternal<T>>,
+    pd: Arc<PdFilePartStatus>,
+    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<T: Seek + Write + Send + Sync + ClearFile + GetTarg
                             pd.as_ref().unwrap().inc(len)?;
                             d.pd.inc(len as u64)?;
                             d.pd.update_part_data(index.unwrap())?;
+                            #[cfg(test)]
+                            {
+                                println!("Index {}: Writed data {} bytes", index.unwrap(), len);
+                            }
                         }
                     }
                     Err(e) => {
@@ -302,6 +322,11 @@ pub async fn handle_download<T: Seek + Write + Send + Sync + ClearFile + GetTarg
                         if !d.is_multi_threads() {
                             return Ok(());
                         }
+                        #[cfg(test)]
+                        {
+                            println!("Index {} set to downloaded.", index.unwrap());
+                        }
+                        pd.as_ref().unwrap().set_downloaded()?;
                     }
                     break;
                 }
@@ -341,6 +366,7 @@ pub async fn add_new_multi_tasks<
         let index = d.pd.get_next_waited_part_data(&mut data);
         match index {
             Some(index) => {
+                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);
+                        }
+                    }
                 }
             }
         }