refactor: 修复大量问题

Author: share121

Cargo.lock

@@ -10,7 +10,7 @@ license = "MIT"
 authors = ["share121 <[email protected]>", "Cyan Changes <[email protected]>"]
 
 [workspace.dependencies]
-fast-pull = { version = "3.2.3", path = "crates/fast-pull" }
+fast-pull = { version = "3.3.0", path = "crates/fast-pull" }
 fast-steal = { version = "6.0.10", path = "crates/fast-steal" }
 kanal = "0.1.1"
 bytes = { version = "1.10.1", default-features = false }

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "fast-down"
 description = "Download everything fast"
-version = "3.2.7"
+version = "3.3.0"
 readme = "README.md"
 documentation = "https://docs.rs/fast-down"
 keywords = ["download", "fast", "parallel", "concurrency"]

crates/fast-down/Cargo.toml

@@ -6,7 +6,7 @@ pub use puller::*;
 use crate::url_info::FileId;
 use bytes::Bytes;
 use fast_pull::ProgressEntry;
-use std::{fmt::Debug, future::Future};
+use std::{fmt::Debug, future::Future, time::Duration};
 use url::Url;
 
 pub trait HttpClient: Clone + Send + Sync + Unpin {
@@ -16,7 +16,9 @@ pub trait HttpClient: Clone + Send + Sync + Unpin {
 pub trait HttpRequestBuilder {
     type Response: HttpResponse;
     type RequestError: Send + Debug;
-    fn send(self) -> impl Future<Output = Result<Self::Response, Self::RequestError>> + Send;
+    fn send(
+        self,
+    ) -> impl Future<Output = Result<Self::Response, (Self::RequestError, Option<Duration>)>> + Send;
 }
 pub trait HttpResponse: Send + Unpin {
     type Headers: HttpHeaders;
@@ -30,10 +32,9 @@ pub trait HttpHeaders {
     fn get(&self, header: &str) -> Result<&str, Self::GetHeaderError>;
 }
 
-pub type GetResponse<Client> =
-    <<Client as HttpClient>::RequestBuilder as HttpRequestBuilder>::Response;
-pub type GetRequestError<Client> =
-    <<Client as HttpClient>::RequestBuilder as HttpRequestBuilder>::RequestError;
+pub type GetRequestBuilder<Client> = <Client as HttpClient>::RequestBuilder;
+pub type GetResponse<Client> = <GetRequestBuilder<Client> as HttpRequestBuilder>::Response;
+pub type GetRequestError<Client> = <GetRequestBuilder<Client> as HttpRequestBuilder>::RequestError;
 pub type GetChunkError<Client> = <GetResponse<Client> as HttpResponse>::ChunkError;
 pub type GetHeader<Client> = <GetResponse<Client> as HttpResponse>::Headers;
 pub type GetHeaderError<Client> = <GetHeader<Client> as HttpHeaders>::GetHeaderError;

crates/fast-down/src/http/mod.rs

@@ -4,20 +4,22 @@ use crate::{
     url_info::FileId,
 };
 use content_disposition;
-use std::{borrow::Borrow, future::Future};
+use std::{borrow::Borrow, future::Future, time::Duration};
 use url::Url;
 
+pub type PrefetchResult<Client, E> = Result<(UrlInfo, GetResponse<Client>), (E, Option<Duration>)>;
+
 pub trait Prefetch<Client: HttpClient> {
     type Error;
     fn prefetch(
         &self,
         url: Url,
-    ) -> impl Future<Output = Result<(UrlInfo, GetResponse<Client>), Self::Error>> + Send;
+    ) -> impl Future<Output = PrefetchResult<Client, Self::Error>> + Send;
 }
 
 impl<Client: HttpClient, BorrowClient: Borrow<Client> + Sync> Prefetch<Client> for BorrowClient {
     type Error = HttpError<Client>;
-    async fn prefetch(&self, url: Url) -> Result<(UrlInfo, GetResponse<Client>), Self::Error> {
+    async fn prefetch(&self, url: Url) -> PrefetchResult<Client, Self::Error> {
         prefetch(self.borrow(), url).await
     }
 }
@@ -36,18 +38,19 @@ fn get_filename(headers: &impl HttpHeaders, url: &Url) -> String {
                 .filter(|s| !s.trim().is_empty())
                 .map(|s| s.to_string())
         })
+        .or_else(|| url.domain().map(|s| s.to_string()))
         .unwrap_or_else(|| url.to_string())
 }
 
 async fn prefetch<Client: HttpClient>(
     client: &Client,
     url: Url,
-) -> Result<(UrlInfo, GetResponse<Client>), HttpError<Client>> {
+) -> PrefetchResult<Client, HttpError<Client>> {
     let resp = client
         .get(url, None)
         .send()
         .await
-        .map_err(HttpError::Request)?;
+        .map_err(|(e, d)| (HttpError::Request(e), d))?;
     let headers = resp.headers();
     let supports_range = headers
         .get("accept-ranges")

crates/fast-down/src/http/prefetch.rs

@@ -10,6 +10,7 @@ use std::{
     pin::{Pin, pin},
     sync::Arc,
     task::{Context, Poll},
+    time::Duration,
 };
 use url::Url;
 
@@ -36,8 +37,13 @@ impl<Client: HttpClient> HttpPuller<Client> {
     }
 }
 
-type ResponseFut<Client> =
-    Pin<Box<dyn Future<Output = Result<GetResponse<Client>, GetRequestError<Client>>> + Send>>;
+type ResponseFut<Client> = Pin<
+    Box<
+        dyn Future<
+                Output = Result<GetResponse<Client>, (GetRequestError<Client>, Option<Duration>)>,
+            > + Send,
+    >,
+>;
 enum ResponseState<Client: HttpClient> {
     Pending(ResponseFut<Client>),
     Ready(GetResponse<Client>),
@@ -49,7 +55,7 @@ impl<Client: HttpClient + 'static> RandPuller for HttpPuller<Client> {
     fn pull(
         &mut self,
         range: &ProgressEntry,
-    ) -> impl TryStream<Ok = Bytes, Error = Self::Error> + Send + Unpin {
+    ) -> impl TryStream<Ok = Bytes, Error = (Self::Error, Option<Duration>)> + Send + Unpin {
         RandRequestStream {
             client: self.client.clone(),
             url: self.url.clone(),
@@ -76,7 +82,7 @@ struct RandRequestStream<Client: HttpClient + 'static> {
     file_id: FileId,
 }
 impl<Client: HttpClient> Stream for RandRequestStream<Client> {
-    type Item = Result<Bytes, HttpError<Client>>;
+    type Item = Result<Bytes, (HttpError<Client>, Option<Duration>)>;
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let chunk_global;
         match &mut self.state {
@@ -89,15 +95,18 @@ impl<Client: HttpClient> Stream for RandRequestStream<Client> {
                             let new_file_id = FileId::new(etag, last_modified);
                             if new_file_id != self.file_id {
                                 self.state = ResponseState::None;
-                                Poll::Ready(Some(Err(HttpError::MismatchedBody(new_file_id))))
+                                Poll::Ready(Some(Err((
+                                    HttpError::MismatchedBody(new_file_id),
+                                    None,
+                                ))))
                             } else {
                                 self.state = ResponseState::Ready(resp);
                                 self.poll_next(cx)
                             }
                         }
-                        Err(e) => {
+                        Err((e, d)) => {
                             self.state = ResponseState::None;
-                            Poll::Ready(Some(Err(HttpError::Request(e))))
+                            Poll::Ready(Some(Err((HttpError::Request(e), d))))
                         }
                     },
                     Poll::Pending => Poll::Pending,
@@ -128,15 +137,17 @@ impl<Client: HttpClient> Stream for RandRequestStream<Client> {
             }
             Err(e) => {
                 self.state = ResponseState::None;
-                Poll::Ready(Some(Err(HttpError::Chunk(e))))
+                Poll::Ready(Some(Err((HttpError::Chunk(e), None))))
             }
         }
     }
 }
 
 impl<Client: HttpClient + 'static> SeqPuller for HttpPuller<Client> {
     type Error = HttpError<Client>;
-    fn pull(&mut self) -> impl TryStream<Ok = Bytes, Error = Self::Error> + Send + Unpin {
+    fn pull(
+        &mut self,
+    ) -> impl TryStream<Ok = Bytes, Error = (Self::Error, Option<Duration>)> + Send + Unpin {
         SeqRequestStream {
             state: if let Some(resp) = &self.resp
                 && let Some(resp) = resp.lock().take()
@@ -155,7 +166,7 @@ struct SeqRequestStream<Client: HttpClient + 'static> {
     file_id: FileId,
 }
 impl<Client: HttpClient> Stream for SeqRequestStream<Client> {
-    type Item = Result<Bytes, HttpError<Client>>;
+    type Item = Result<Bytes, (HttpError<Client>, Option<Duration>)>;
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let chunk_global;
         match &mut self.state {
@@ -168,21 +179,24 @@ impl<Client: HttpClient> Stream for SeqRequestStream<Client> {
                             let new_file_id = FileId::new(etag, last_modified);
                             if new_file_id != self.file_id {
                                 self.state = ResponseState::None;
-                                Poll::Ready(Some(Err(HttpError::MismatchedBody(new_file_id))))
+                                Poll::Ready(Some(Err((
+                                    HttpError::MismatchedBody(new_file_id),
+                                    None,
+                                ))))
                             } else {
                                 self.state = ResponseState::Ready(resp);
                                 self.poll_next(cx)
                             }
                         }
-                        Err(e) => {
+                        Err((e, d)) => {
                             self.state = ResponseState::None;
-                            Poll::Ready(Some(Err(HttpError::Request(e))))
+                            Poll::Ready(Some(Err((HttpError::Request(e), d))))
                         }
                     },
                     Poll::Pending => Poll::Pending,
                 };
             }
-            ResponseState::None => return Poll::Ready(Some(Err(HttpError::Irrecoverable))),
+            ResponseState::None => return Poll::Ready(Some(Err((HttpError::Irrecoverable, None)))),
             ResponseState::Ready(resp) => {
                 let mut chunk = pin!(resp.chunk());
                 match chunk.try_poll_unpin(cx) {
@@ -197,7 +211,7 @@ impl<Client: HttpClient> Stream for SeqRequestStream<Client> {
             Ok(chunk) => Poll::Ready(Some(Ok(chunk))),
             Err(e) => {
                 self.state = ResponseState::None;
-                Poll::Ready(Some(Err(HttpError::Chunk(e))))
+                Poll::Ready(Some(Err((HttpError::Chunk(e), None))))
             }
         }
     }

crates/fast-down/src/http/puller.rs

@@ -25,30 +25,27 @@ impl HttpClient for Client {
 impl HttpRequestBuilder for RequestBuilder {
     type Response = Response;
     type RequestError = ReqwestResponseError;
-    async fn send(self) -> Result<Self::Response, Self::RequestError> {
-        let res = self.send().await.map_err(ReqwestResponseError::Reqwest)?;
-        let retry_after = res.headers().get(header::RETRY_AFTER);
-        if let Some(retry_after) = retry_after
-            && let Ok(retry_after) = retry_after.to_str()
-        {
-            let retry_after = match retry_after.parse() {
-                Ok(retry_after) => Some(Duration::from_secs(retry_after)),
-                Err(_) => match parse_http_date(retry_after) {
-                    Ok(target_time) => target_time.duration_since(SystemTime::now()).ok(),
-                    Err(_) => None,
-                },
-            };
-            if let Some(retry_after) = retry_after
-                && retry_after > Duration::ZERO
-            {
-                tokio::time::sleep(retry_after).await;
-            }
-        }
+    async fn send(self) -> Result<Self::Response, (Self::RequestError, Option<Duration>)> {
+        let res = self
+            .send()
+            .await
+            .map_err(|e| (ReqwestResponseError::Reqwest(e), None))?;
         let status = res.status();
         if status.is_success() {
             Ok(res)
         } else {
-            Err(ReqwestResponseError::StatusCode(status))
+            let retry_after = res
+                .headers()
+                .get(header::RETRY_AFTER)
+                .and_then(|r| r.to_str().ok())
+                .and_then(|r| match r.parse() {
+                    Ok(r) => Some(Duration::from_secs(r)),
+                    Err(_) => match parse_http_date(r) {
+                        Ok(target_time) => target_time.duration_since(SystemTime::now()).ok(),
+                        Err(_) => None,
+                    },
+                });
+            Err((ReqwestResponseError::StatusCode(status), retry_after))
         }
     }
 }
@@ -196,7 +193,7 @@ mod tests {
         let url = Url::parse(&format!("{}/404", server.url())).unwrap();
         match client.prefetch(url).await {
             Ok(info) => unreachable!("404 status code should not success: {info:?}"),
-            Err(err) => match err {
+            Err((err, _)) => match err {
                 HttpError::Request(e) => match e {
                     ReqwestResponseError::Reqwest(error) => unreachable!("{error:?}"),
                     ReqwestResponseError::StatusCode(status_code) => {

crates/fast-down/src/reqwest/mod.rs

@@ -1,7 +1,7 @@
 [package]
 name = "fast-pull"
 description = "Pull everything fast"
-version = "3.2.3"
+version = "3.3.0"
 readme = "README.md"
 documentation = "https://docs.rs/fast-pull"
 keywords = ["performance", "multiplex", "fast", "parallel", "concurrency"]

crates/fast-pull/Cargo.toml

@@ -1,16 +1,19 @@
 use crate::ProgressEntry;
 use bytes::Bytes;
+use core::time::Duration;
 use futures::TryStream;
 
 pub trait RandPuller: Send + Clone {
     type Error: Send;
     fn pull(
         &mut self,
         range: &ProgressEntry,
-    ) -> impl TryStream<Ok = Bytes, Error = Self::Error> + Send + Unpin;
+    ) -> impl TryStream<Ok = Bytes, Error = (Self::Error, Option<Duration>)> + Send + Unpin;
 }
 
 pub trait SeqPuller: Send {
     type Error: Send;
-    fn pull(&mut self) -> impl TryStream<Ok = Bytes, Error = Self::Error> + Send + Unpin;
+    fn pull(
+        &mut self,
+    ) -> impl TryStream<Ok = Bytes, Error = (Self::Error, Option<Duration>)> + Send + Unpin;
 }

crates/fast-pull/src/base/puller.rs

@@ -1,4 +1,6 @@
 extern crate alloc;
+use core::time::Duration;
+
 use crate::{ProgressEntry, RandPuller, SeqPuller};
 use alloc::{sync::Arc, vec::Vec};
 use bytes::Bytes;
@@ -20,14 +22,16 @@ impl RandPuller for MockPuller {
     fn pull(
         &mut self,
         range: &ProgressEntry,
-    ) -> impl TryStream<Ok = Bytes, Error = Self::Error> + Send + Unpin {
+    ) -> impl TryStream<Ok = Bytes, Error = (Self::Error, Option<Duration>)> + Send + Unpin {
         let data = &self.0[range.start as usize..range.end as usize];
         stream::iter(data.iter().map(|e| Ok(Bytes::from_iter([*e]))))
     }
 }
 impl SeqPuller for MockPuller {
     type Error = ();
-    fn pull(&mut self) -> impl TryStream<Ok = Bytes, Error = Self::Error> + Send + Unpin {
+    fn pull(
+        &mut self,
+    ) -> impl TryStream<Ok = Bytes, Error = (Self::Error, Option<Duration>)> + Send + Unpin {
         stream::iter(self.0.iter().map(|e| Ok(Bytes::from_iter([*e]))))
     }
 }