|
|
|
@ -1,5 +1,6 @@ |
|
|
|
use std::{collections::HashMap, io::ErrorKind, path::{Path, PathBuf}, time::Duration};
|
|
|
|
|
|
|
|
use anyhow::anyhow;
|
|
|
|
use futures_util::StreamExt as _;
|
|
|
|
use http::{
|
|
|
|
header::{CONTENT_LENGTH, RANGE},
|
|
|
|
@ -22,6 +23,7 @@ use tokio::{ |
|
|
|
time::timeout
|
|
|
|
};
|
|
|
|
use tower::{util::BoxCloneService, ServiceBuilder, ServiceExt as _};
|
|
|
|
use tower_http::decompression::{DecompressionBody, DecompressionLayer};
|
|
|
|
use tower_http_client::ServiceExt as _;
|
|
|
|
use tower_reqwest::HttpClientLayer;
|
|
|
|
|
|
|
|
@ -53,7 +55,7 @@ pub(super) enum ClientActorMessageHandle { |
|
|
|
|
|
|
|
pub(super) type ActionIndex = u64;
|
|
|
|
type JoinSetResult = Result<Option<ClientActorMessageHandle>, DownloadError>;
|
|
|
|
type HttpClient = BoxCloneService<Request<Body>, Response<Body>, anyhow::Error>;
|
|
|
|
type HttpClient = BoxCloneService<Request<Body>, Response<DecompressionBody<Body>>, anyhow::Error>;
|
|
|
|
|
|
|
|
|
|
|
|
#[derive(Debug)]
|
|
|
|
@ -143,7 +145,7 @@ async fn file_size(filename: &Path) -> u64 { |
|
|
|
async fn request( client: &mut HttpClient
|
|
|
|
, method: &str |
|
|
|
, uri: &Uri
|
|
|
|
, headers: HeaderMap ) -> anyhow::Result<Response<Body>> {
|
|
|
|
, headers: HeaderMap ) -> anyhow::Result<Response<DecompressionBody<Body>>> {
|
|
|
|
let mut request = RequestBuilder::new()
|
|
|
|
. method(method)
|
|
|
|
. uri(uri)
|
|
|
|
@ -151,11 +153,11 @@ async fn request( client: &mut HttpClient |
|
|
|
|
|
|
|
request.headers_mut().extend(headers);
|
|
|
|
|
|
|
|
debug!("New Request: {:?}", request);
|
|
|
|
debug!("Request: {:?}", request);
|
|
|
|
|
|
|
|
let response = client.execute(request).await?;
|
|
|
|
|
|
|
|
debug!("New Response: {:?}", response);
|
|
|
|
debug!("Response: {:?}", response.headers());
|
|
|
|
|
|
|
|
anyhow::ensure!( response.status().is_success()
|
|
|
|
, "resonse status failed: {}"
|
|
|
|
@ -195,7 +197,7 @@ async fn open_outfile(status: &StatusCode, filename: &Path) -> File { |
|
|
|
}
|
|
|
|
|
|
|
|
async fn store_body( file: &mut File
|
|
|
|
, body: &mut Body
|
|
|
|
, body: &mut DecompressionBody<Body>
|
|
|
|
, io_timeout: Duration ) -> anyhow::Result<()> {
|
|
|
|
let mut body = BodyDataStream::new(body);
|
|
|
|
|
|
|
|
@ -204,7 +206,10 @@ async fn store_body( file: &mut File |
|
|
|
let data = timeout(io_timeout, body.next()).await?;
|
|
|
|
match data {
|
|
|
|
None => break,
|
|
|
|
Some(Err(e)) => return Err(e.into()),
|
|
|
|
Some(Err(e)) => {
|
|
|
|
return Err(anyhow!(e.to_string()));
|
|
|
|
}
|
|
|
|
|
|
|
|
Some(Ok(data)) => {
|
|
|
|
file . write_all(data.as_ref()).await?;
|
|
|
|
file . flush().await?;
|
|
|
|
@ -266,6 +271,7 @@ impl ClientActor { |
|
|
|
// Add some layers.
|
|
|
|
. concurrency_limit(concurrency_limit)
|
|
|
|
. timeout(timeout)
|
|
|
|
. layer(DecompressionLayer::new())
|
|
|
|
// Make client compatible with the `tower-http` layers.
|
|
|
|
. layer(HttpClientLayer)
|
|
|
|
. service( reqwest::Client::builder()
|
|
|
|
@ -334,6 +340,7 @@ impl ClientActorHandle { |
|
|
|
|
|
|
|
pub(super) fn stop(self) {
|
|
|
|
let _ = self.abort.send(Ok(None));
|
|
|
|
drop(self.sender);
|
|
|
|
}
|
|
|
|
|
|
|
|
pub(super) async fn download(&self, filename: impl AsRef<Path>, uri: &Uri) {
|
|
|
|
|
