chore: 使用 kanel 框架代替 async-channel

Author: share121

Cargo.lock

@@ -40,13 +40,14 @@ tokio = { version = "1.47.0", features = [
     "io-util",
 ], default-features = false }
 urlencoding = "2.1.3"
+kanal = "0.1.1"
 
 [dev-dependencies]
 mockito = "1.7.0"
 tempfile = "3.20.0"
 
 [features]
-default = ["file"]
+default = ["file", "reqwest"]
 file = ["memmap2"]
 memmap2 = ["dep:memmap2"]
 reqwest = ["common.url", "dep:reqwest", "dep:http"]

Cargo.toml

@@ -1,11 +1,11 @@
 #[cfg(feature = "common.url")]
 mod url;
 
+use crate::{FetchResult, Fetcher, ProgressEntry, Puller, RandomPusher};
 use std::num::NonZeroUsize;
 use std::time::Duration;
 #[cfg(feature = "common.url")]
 pub use url::UrlInfo;
-use crate::{FetchResult, Fetcher, ProgressEntry, Puller, RandomPusher};
 
 #[derive(Debug, Eq, PartialEq)]
 pub struct DownloadOptions {
@@ -15,13 +15,15 @@ pub struct DownloadOptions {
 }
 
 pub trait AutoDownload {
-    async fn download<F, P>(
+    fn download<F, P>(
         &self,
         fetcher: F,
         pusher: P,
         maybe_chunks: Option<Vec<ProgressEntry>>,
         options: DownloadOptions,
-    ) -> FetchResult<F::Error, <F::Puller as Puller>::Error, P::Error>
+    ) -> impl std::future::Future<
+        Output = FetchResult<F::Error, <F::Puller as Puller>::Error, P::Error>,
+    > + Send
     where
         F: Fetcher + Send + 'static,
         P: RandomPusher + Send + 'static;

src/common/mod.rs

@@ -1,8 +1,7 @@
+use super::{AutoDownload, DownloadOptions};
+use crate::{FetchResult, Fetcher, ProgressEntry, Puller, RandomPusher, multi, single};
 use std::num::NonZeroUsize;
-use std::time::Duration;
 use url::Url;
-use crate::{multi, single, FetchResult, Fetcher, ProgressEntry, Puller, Pusher, RandomPusher};
-use super::{AutoDownload, DownloadOptions};
 
 #[derive(Debug, Clone)]
 pub struct UrlInfo {
@@ -28,18 +27,30 @@ impl AutoDownload for UrlInfo {
         P: RandomPusher + Send + 'static,
     {
         if self.fast_download {
-            multi::download_multi(fetcher, pusher, multi::DownloadOptions {
-                pull_chunks: maybe_chunks.unwrap_or_else(|| vec![0..self.size]),
-                concurrent: options.concurrent.unwrap_or(const { NonZeroUsize::new(1).unwrap() }),
-                retry_gap: options.retry_gap,
-                push_queue_cap: options.push_queue_cap,
-            }).await
+            multi::download_multi(
+                fetcher,
+                pusher,
+                multi::DownloadOptions {
+                    #[allow(clippy::single_range_in_vec_init)]
+                    pull_chunks: maybe_chunks.unwrap_or_else(|| vec![0..self.size]),
+                    concurrent: options
+                        .concurrent
+                        .unwrap_or(const { NonZeroUsize::new(1).unwrap() }),
+                    retry_gap: options.retry_gap,
+                    push_queue_cap: options.push_queue_cap,
+                },
+            )
+            .await
         } else {
-            single::download_single(fetcher, pusher, single::DownloadOptions {
-                retry_gap: options.retry_gap,
-                push_queue_cap: options.push_queue_cap,
-            }).await
+            single::download_single(
+                fetcher,
+                pusher,
+                single::DownloadOptions {
+                    retry_gap: options.retry_gap,
+                    push_queue_cap: options.push_queue_cap,
+                },
+            )
+            .await
         }
     }
 }
-

src/common/url.rs

@@ -1,7 +1,5 @@
 use crate::Event;
-use crate::base::pusher::Pusher;
-use crate::base::source::{Fetcher, Puller};
-use async_channel::Receiver;
+use kanal::AsyncReceiver;
 use std::{
     fmt::Debug,
     sync::{
@@ -22,15 +20,14 @@ pub mod single;
 
 #[derive(Debug, Clone)]
 pub struct FetchResult<FetchError, PullError, PushError> {
-    pub event_chain: Receiver<Event<FetchError, PullError, PushError>>,
+    pub event_chain: AsyncReceiver<Event<FetchError, PullError, PushError>>,
     handle: Arc<Mutex<Option<JoinHandle<()>>>>,
     is_running: Arc<AtomicBool>,
 }
 
-impl<FetchError, PullError, PushError> FetchResult<FetchError, PullError, PushError>
-{
+impl<FetchError, PullError, PushError> FetchResult<FetchError, PullError, PushError> {
     pub fn new(
-        event_chain: Receiver<Event<FetchError, PullError, PushError>>,
+        event_chain: AsyncReceiver<Event<FetchError, PullError, PushError>>,
         handle: JoinHandle<()>,
         is_running: Arc<AtomicBool>,
     ) -> Self {

src/core/mod.rs

@@ -30,9 +30,9 @@ where
     F: Fetcher + Send + 'static,
     P: RandomPusher + Send + 'static,
 {
-    let (tx, event_chain) = async_channel::unbounded();
+    let (tx, event_chain) = kanal::unbounded_async();
     let (tx_write, rx_write) =
-        async_channel::bounded::<(WorkerId, ProgressEntry, Bytes)>(options.push_queue_cap);
+        kanal::bounded_async::<(WorkerId, ProgressEntry, Bytes)>(options.push_queue_cap);
     let tx_clone = tx.clone();
     let handle = tokio::spawn(async move {
         while let Ok((id, spin, data)) = rx_write.recv().await {

src/core/multi.rs

@@ -25,9 +25,9 @@ where
     F: Fetcher + Send + 'static,
     P: Pusher + Send + 'static,
 {
-    let (tx, event_chain) = async_channel::unbounded();
+    let (tx, event_chain) = kanal::unbounded_async();
     let (tx_write, rx_write) =
-        async_channel::bounded::<(ProgressEntry, Bytes)>(options.push_queue_cap);
+        kanal::bounded_async::<(ProgressEntry, Bytes)>(options.push_queue_cap);
     let tx_clone = tx.clone();
     const ID: usize = 0;
     let handle = tokio::spawn(async move {