downloader: support to drop.

Fix .pd file not created when overwriting existed file.

src/downloader/downloader.rs

@@ -30,6 +30,7 @@ use std::io::SeekFrom;
 use std::io::Write;
 use std::ops::Deref;
 use std::ops::DerefMut;
+use std::ops::Drop;
 use std::path::Path;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::AtomicI64;
@@ -89,6 +90,8 @@ pub struct DownloaderInternal<T: Write + Seek + Send + Sync + ClearFile + GetTar
     pub max_threads: AtomicU64,
     /// The size of the each part when downloading file.
     part_size: AtomicU32,
+    /// Is outter [Downloader] dropped
+    dropped: AtomicBool,
 }
 
 impl DownloaderInternal<LocalFile> {
@@ -120,7 +123,12 @@ impl DownloaderInternal<LocalFile> {
                                 return Ok(DownloaderResult::Canceled);
                             } else {
                                 remove_file(p)?;
-                                PdFile::new()
+                                match PdFile::open(p)? {
+                                    PdFileResult::Ok(f) => f,
+                                    _ => {
+                                        panic!("This should not happened.");
+                                    }
+                                }
                             }
                         }
                         None => {
@@ -171,6 +179,7 @@ impl DownloaderInternal<LocalFile> {
             max_part_retry_count: AtomicI64::new(3),
             max_threads: AtomicU64::new(8),
             part_size: AtomicU32::new(0x10000),
+            dropped: AtomicBool::new(false),
         }))
     }
 }
@@ -325,6 +334,12 @@ impl<T: Write + Seek + Send + Sync + ClearFile + GetTargetFileName> DownloaderIn
         *self.status.get_ref() == DownloaderStatus::Downloaded
     }
 
+    #[inline]
+    /// Return true if outter [Downloader] dropped
+    pub fn is_dropped(&self) -> bool {
+        self.dropped.qload()
+    }
+
     #[inline]
     /// Returns true if the downloader is panic.
     pub fn is_panic(&self) -> bool {
@@ -362,6 +377,12 @@ impl<T: Write + Seek + Send + Sync + ClearFile + GetTargetFileName> DownloaderIn
         self.status.replace_with2(DownloaderStatus::Downloaded)
     }
 
+    #[inline]
+    /// Set outter [Downloader] dropped
+    pub fn set_dropped(&self) {
+        self.dropped.qstore(true);
+    }
+
     #[inline]
     /// Set the downloader is panic and set the error.
     /// * `err` - Error
@@ -433,6 +454,9 @@ impl<T: Write + Seek + Send + Sync + ClearFile + GetTargetFileName> DownloaderIn
 }
 
 /// A file downloader
+///
+/// When dropping downloader, the downloader need some times to let all threads exited.
+/// This process is handled after [Drop].
 pub struct Downloader<T: Write + Seek + Send + Sync + ClearFile + GetTargetFileName> {
     /// internal type
     downloader: Arc<DownloaderInternal<T>>,
@@ -645,6 +669,16 @@ impl<T: Write + Seek + Send + Sync + ClearFile + GetTargetFileName + 'static> Do
     define_downloader_fn!(is_panic, bool, "Returns true if the downloader is panic.");
 }
 
+impl<T: Write + Seek + Send + Sync + ClearFile + GetTargetFileName> Drop for Downloader<T> {
+    fn drop(&mut self) {
+        self.downloader.set_dropped();
+        #[cfg(test)]
+        {
+            println!("The downloader is dropped.");
+        }
+    }
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn test_downloader() {
     let p = Path::new("./test");
@@ -721,3 +755,56 @@ async fn test_failed_downloader() {
         }
     }
 }
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn test_downloader_dropped() {
+    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/15/14/44/22/99066680_p0.png";
+    let pb = p.join("99066680_p0.png");
+    {
+        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.download();
+            }
+            DownloaderResult::Canceled => {
+                panic!("This should not happened.")
+            }
+        }
+    }
+    tokio::time::sleep(Duration::new(1, 0)).await;
+    {
+        println!("The new downloader is created.");
+        let downloader = Downloader::<LocalFile>::new(
+            url,
+            json::object! {"referer": "https://www.pixiv.net/"},
+            Some(&pb),
+            Some(false),
+        )
+        .unwrap();
+        match downloader {
+            DownloaderResult::Ok(v) => {
+                assert_eq!(v.is_created(), true);
+                v.disable_progress_bar();
+                v.download();
+                v.join().await.unwrap();
+                assert_eq!(v.is_downloaded(), true);
+            }
+            DownloaderResult::Canceled => {
+                println!("The file is already downloaded. (Too fast network. QaQ)");
+            }
+        }
+    }
+}

src/downloader/enums.rs

@@ -1,3 +1,5 @@
+use std::fmt::Display;
+
 #[derive(Debug)]
 /// The result when try create a new [super::Downloader] interface
 pub enum DownloaderResult<T> {
@@ -19,3 +21,14 @@ pub enum DownloaderStatus {
     /// The downloader is stoped
     Panic,
 }
+
+impl Display for DownloaderStatus {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(match self {
+            Self::Created => "Created",
+            Self::Downloading => "Downloading",
+            Self::Downloaded => "Downloaded",
+            Self::Panic => "Panic",
+        })
+    }
+}

src/downloader/pd_file/file.rs

@@ -505,14 +505,26 @@ impl PdFile {
 
 impl Drop for PdFile {
     fn drop(&mut self) {
+        #[cfg(test)]
+        {
+            println!("Is memory only: {}", self.is_mem_only());
+        }
         if self.is_mem_only() {
             return;
         }
+        #[cfg(test)]
+        {
+            println!("Is completed: {}", self.is_completed());
+        }
         if self.is_completed() {
             self.force_close();
             self.remove_pd_file_with_err_msg();
             return;
         }
+        #[cfg(test)]
+        {
+            println!("Is need saved: {}", self.is_need_saved());
+        }
         if self.is_need_saved() {
             match self.write() {
                 Ok(_) => {}

src/downloader/tasks.rs

@@ -16,6 +16,27 @@ use std::ops::Deref;
 use std::sync::Arc;
 use std::time::Duration;
 
+/// Return [Ok(())] if the [super::Downloader] is dropped.
+macro_rules! check_dropped {
+    ($d:expr) => {
+        if $d.is_dropped() {
+            #[cfg(test)]
+            {
+                println!("The downloader status: {}", $d.get_status());
+                if $d.is_downloading() {
+                    println!(
+                        "The downloader size: {} / {}",
+                        $d.pd.get_downloaded_file_size(),
+                        $d.pd.get_file_size()
+                    );
+                }
+                println!("The downloader is already dropped. Exit download.");
+            }
+            return Ok(());
+        }
+    };
+}
+
 /// Create a download tasks in simple thread mode.
 pub async fn create_download_tasks_simple<
     T: Seek + Write + Send + Sync + ClearFile + GetTargetFileName,
@@ -129,6 +150,7 @@ pub async fn handle_download<T: Seek + Write + Send + Sync + ClearFile + GetTarg
     let is_multi = d.is_multi_threads();
     loop {
         let mut n = stream.next();
+        check_dropped!(d);
         let re = tokio::time::timeout(std::time::Duration::from_secs(10), &mut n).await;
         match re {
             Ok(s) => match s {
@@ -203,6 +225,7 @@ pub async fn check_tasks<
         d.add_task(task);
     }
     loop {
+        check_dropped!(d);
         tokio::time::sleep(Duration::new(0, 10_000_000)).await;
         let mut need_break = false;
         let mut dur = None;