Browse Source

improved Errors for Client and ClientActorHandle

main
Georg Hopp 11 months ago
parent
commit
d008540a8b
Signed by: ghopp GPG Key ID: 4C5D226768784538
  1. 1
      Cargo.lock
  2. 1
      Cargo.toml
  3. 188
      src/client.rs
  4. 65
      src/client/error.rs
  5. 83
      src/client/error/download_error.rs
  6. 88
      src/client/error/request_error.rs
  7. 9
      src/client_actor.rs
  8. 30
      src/client_actor/error.rs
  9. 7
      src/client_actor/message.rs
  10. 8
      src/client_actor/util.rs
  11. 3
      src/m3u8_download.rs
  12. 4
      src/main.rs

1
Cargo.lock

@ -512,6 +512,7 @@ name = "hlsclient"
version = "0.2.3"
dependencies = [
"anyhow",
"bytes",
"clap",
"env_logger",
"futures-util",

1
Cargo.toml

@ -5,6 +5,7 @@ edition = "2021"
[dependencies]
anyhow = "1.0"
bytes = "1.9"
clap = { version = "4.5", features = [ "derive", "cargo" ] }
env_logger = "0.11"
futures-util = "0.3"

188
src/client.rs

@ -1,29 +1,25 @@
mod error;
pub(super) mod error;
mod util;
use std::{path::Path, time::Duration};
use anyhow::anyhow;
use bytes::Bytes;
use clap::{crate_name, crate_version};
use error::{DownloadError, RequestError};
use futures_util::StreamExt as _;
use http::{
header::{CONTENT_LENGTH, CONTENT_TYPE, ORIGIN, RANGE, USER_AGENT},
request::Builder as RequestBuilder,
header,
request,
HeaderMap,
HeaderValue,
Method,
Request,
Response,
Uri
Uri,
};
use http_body_util::BodyDataStream;
use log::debug;
use reqwest::{redirect::Policy, Body};
use tokio::{
fs::File,
io::AsyncWriteExt as _,
time::timeout
};
use tokio::{fs::File, io::AsyncWriteExt as _, time::timeout};
use tower::{util::BoxCloneService, ServiceBuilder, ServiceExt as _};
use tower_http::decompression::{DecompressionBody, DecompressionLayer};
use tower_http_client::{client::BodyReader, ServiceExt as _};
@ -32,8 +28,12 @@ use tower_reqwest::HttpClientLayer;
type ClientBody = DecompressionBody<Body>;
type ClientResponse = Response<ClientBody>;
type ClientResponseResult = Result<ClientResponse, RequestError>;
type HttpClient = BoxCloneService<Request<Body>, ClientResponse, anyhow::Error>;
pub(super) type DataResult = error::DownloadResult<Bytes>;
pub(super) type DownloadResult = error::DownloadResult<DownloadState>;
type HeadResult = error::ClientRequestResult<HeaderMap>;
type InvalidHeaderResult = Result<Client, header::InvalidHeaderValue>;
type RequestResult = error::ClientRequestResult<ClientResponse>;
#[derive(Clone, Debug)]
@ -57,7 +57,7 @@ impl Client {
pub(super) fn new( buffer: usize
, rate_limit: u64
, concurrency_limit: usize
, timeout: Duration ) -> anyhow::Result<Self> {
, timeout: Duration ) -> Result<Self, reqwest::Error> {
let client = ServiceBuilder::new()
// Add some layers.
. buffer(buffer)
@ -76,7 +76,7 @@ impl Client {
let body_timeout = None;
let mut default_headers = HeaderMap::new();
default_headers.insert(
USER_AGENT,
header::USER_AGENT,
HeaderValue::from_str(&( crate_name!().to_string() + "/"
+ crate_version!() )).unwrap() );
@ -88,60 +88,51 @@ impl Client {
self
}
pub(super) fn set_origin(mut self, origin: Option<String>) -> Self {
if let Some(origin) = origin {
self.default_headers.insert(
ORIGIN,
HeaderValue::from_str(&origin).unwrap() );
} else {
self.default_headers.remove(ORIGIN);
}
self
pub(super) fn set_origin( mut self, origin: Option<String>)
-> InvalidHeaderResult {
match origin.as_deref() {
Some(origin) =>
self.default_headers.insert( header::ORIGIN
, HeaderValue::from_str(origin)? ),
None => self.default_headers.remove(header::ORIGIN),
};
Ok(self)
}
pub(super) fn set_user_agent(mut self, user_agent: Option<String>) -> Self {
if let Some(user_agent) = user_agent {
self.default_headers.insert(
USER_AGENT,
HeaderValue::from_str(&user_agent).unwrap() );
} else {
self.default_headers.remove(USER_AGENT);
}
self
pub(super) fn set_user_agent(mut self, user_agent: Option<String>)
-> InvalidHeaderResult {
match user_agent.as_deref() {
Some(user_agent) =>
self.default_headers.insert(
header::USER_AGENT,
HeaderValue::from_str(user_agent)? ),
None => self.default_headers.remove(header::USER_AGENT),
};
Ok(self)
}
pub(super) async fn data( &mut self
, uri: &Uri
, headers: &HeaderMap )
-> anyhow::Result<Vec<u8>> {
let mut response = self.request("GET", uri, headers).await?;
// read body into Vec<u8>
let body = BodyReader::new(response.body_mut())
. bytes()
. await
. map_err(|e| anyhow!(e))?
. to_vec();
Ok(body)
, headers: &HeaderMap ) -> DataResult {
let mut response = self.request(Method::GET, uri, headers).await?;
Ok(BodyReader::new(response.body_mut()).bytes().await?)
}
pub(super) async fn download( &mut self
, filename: impl AsRef<Path>
, uri: &Uri
, headers: &HeaderMap )
-> anyhow::Result<DownloadState> {
, headers: &HeaderMap ) -> DownloadResult {
// - get all informations to eventually existing file
let mut from = util::file_size(&filename).await;
// - get infos to uri
let response_headers = &self.head(uri, headers).await?;
let content_length = util::get_header::<u64>( response_headers
, CONTENT_LENGTH );
, header::CONTENT_LENGTH );
let content_type = util::get_header::<String>( response_headers
, CONTENT_TYPE )
, header::CONTENT_TYPE )
. or(Some("unknown".into()));
if let Some(content_length) = content_length {
@ -154,51 +145,68 @@ impl Client {
// - do the neccessry request.
let headers = &mut headers.clone();
headers.insert(RANGE, format!("bytes={}-", from).parse().unwrap());
headers.insert( header::RANGE
, format!("bytes={}-", from).parse().unwrap() );
let mut response = self.request("GET", uri, headers).await?;
let mut response = self.request(Method::GET, uri, headers).await?;
// - open or create file
let file = util::open_or_create(&response.status(), &filename).await;
// - download Data
Ok( self.clone().store_body( file
, from as usize
, content_type
, response.body_mut() ).await? )
self.clone().store_body( file
, from as usize
, content_type
, response.body_mut() ).await
}
async fn head( &mut self
, uri: &Uri
, headers: &HeaderMap ) -> Result<HeaderMap, RequestError> {
Ok( self.request("HEAD", uri, headers)
, headers: &HeaderMap ) -> HeadResult {
Ok( self.request(Method::HEAD, uri, headers)
. await?
. headers()
. clone() )
}
async fn request( &mut self
, method: &str
, method: Method
, uri: &Uri
, headers: &HeaderMap ) -> ClientResponseResult {
let mut request = RequestBuilder::new()
. method(method)
. uri(uri)
. body(Body::default())
. map_err(|e| RequestError::new(None, Some(e.into())))?;
request.headers_mut().extend(headers.clone());
debug!("Request: {:?}", request);
, headers: &HeaderMap ) -> RequestResult {
let (mut request_parts, _) = request::Builder::new()
. method(method)
. uri(uri)
. body(())?
. into_parts();
request_parts.headers = headers.to_owned();
debug!("Request: {:?}", request_parts);
let request = Request::from_parts( request_parts.clone()
, Body::default() );
match self.client.execute(request).await {
Err(e) => Err(RequestError::new(None, Some(e))),
Err(e) => {
let request_parts = Some(request_parts.clone());
let e = Some(e);
Err(error::ClientRequestError::new( request_parts
, None
, None
, e ))?
},
Ok(response) => {
debug!("Response: {:?}", response.headers());
let (response_parts, response_body) = response.into_parts();
debug!("Response: {:?}", response_parts);
if response.status().is_success() {
Ok(response)
if response_parts.status.is_success() {
Ok(Response::from_parts(response_parts, response_body))
} else {
Err(RequestError::new(Some(response.map(|_| ())), None))
let request = Some(request_parts);
let response = Some(response_parts);
let response_body =
Some(BodyReader::new(response_body).bytes().await?);
Err(error::ClientRequestError::new( request
, response
, response_body
, None ))?
}
},
}
@ -208,30 +216,40 @@ impl Client {
, mut file: File
, mut size: usize
, content_type: Option<String>
, body: &mut ClientBody ) -> Result<DownloadState, DownloadError> {
, body: &mut ClientBody ) -> DownloadResult {
let mut body = BodyDataStream::new(body);
let mut state = DownloadState::Partial { content_type: content_type.clone(), size };
let mut state = DownloadState::Partial {
content_type: content_type.clone(),
size
};
loop {
let data_future = body.next();
let data = if let Some(io_timeout) = self.body_timeout {
// give timeout somehow... probably from client.
timeout(io_timeout, data_future).await
. map_err(|e| DownloadError::new(state.clone(), e.into()))?
timeout(io_timeout, data_future).await.map_err(|e| {
error::DownloadError::from(e).set_state(&state)
})?
} else {
data_future.await
};
match data {
None => break,
Some(Err(e)) => Err(DownloadError::new(state.clone(), anyhow!(e)))?,
Some(Err(e)) =>
Err(error::DownloadError::from(e).set_state(&state))?,
Some(Ok(data)) => {
size += data.len();
state = DownloadState::Partial { content_type: content_type.clone(), size };
file . write_all(&data).await
. map_err(|e| DownloadError::new(state.clone(), e.into()))?;
file . flush().await
. map_err(|e| DownloadError::new(state.clone(), e.into()))?;
state = DownloadState::Partial {
content_type: content_type.clone(),
size
};
file.write_all(&data).await.map_err(|e| {
error::DownloadError::from(e).set_state(&state)
})?;
file.flush().await.map_err(|e| {
error::DownloadError::from(e).set_state(&state)
})?;
},
}
};

65
src/client/error.rs

@ -1,63 +1,10 @@
use std::{error, fmt};
mod request_error;
mod download_error;
use http::Response;
use super::DownloadState;
pub(crate) use request_error::ClientRequestError;
pub(crate) use download_error::DownloadError;
#[derive(Debug)]
pub(super) struct DownloadError {
pub(super) state: DownloadState,
pub(super) source: anyhow::Error,
}
#[derive(Debug)]
pub(super) struct RequestError {
pub(super) response: Option<Response<()>>,
pub(super) source: Option<anyhow::Error>,
}
impl DownloadError {
pub(super) fn new( state: DownloadState
, source: anyhow::Error ) -> Self {
Self { state, source }
}
}
impl error::Error for DownloadError {
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
Some(self.source.as_ref())
}
}
impl fmt::Display for DownloadError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "download error ({:?}): {}", self.state, self.source)
}
}
impl RequestError {
pub(super) fn new( response: Option<Response<()>>
, source: Option<anyhow::Error> ) -> Self {
Self { response, source }
}
}
impl error::Error for RequestError {
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
match &self.source {
None => None,
Some(e) => Some(e.as_ref()),
}
}
}
impl fmt::Display for RequestError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match &self.source {
None => write!(f, "request error: {:?}", self.response),
Some(err) => write!(f, "request error ({:?}): {}", self.response, err),
}
}
}
pub(super) type ClientRequestResult<T> = Result<T, ClientRequestError>;
pub(super) type DownloadResult<T> = Result<T, DownloadError>;

83
src/client/error/download_error.rs

@ -0,0 +1,83 @@
use std::{error, fmt, io};
use tokio::time::error::Elapsed;
use crate::client::DownloadState;
use super::ClientRequestError;
#[derive(Debug)]
pub(crate) enum DownloadErrorSource {
#[allow(dead_code)]
Request(ClientRequestError),
Timeout(Elapsed),
#[allow(dead_code)]
IoError(io::Error),
#[allow(dead_code)]
Other(anyhow::Error),
}
#[derive(Debug)]
pub(crate) struct DownloadError {
state: DownloadState,
source: Option<DownloadErrorSource>,
}
impl DownloadError {
pub(crate) fn new( state: DownloadState
, source: Option<DownloadErrorSource> )
-> Self {
Self { state, source }
}
pub(crate) fn set_state(mut self, state: &DownloadState) -> Self {
self.state = state.to_owned();
self
}
}
impl error::Error for DownloadError {}
impl fmt::Display for DownloadError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!( f
, "download error: {{ state: {:?} / source: {:?} }}"
, self.state
, self.source )
}
}
impl From<ClientRequestError> for DownloadErrorSource {
fn from(value: ClientRequestError) -> Self {
Self::Request(value)
}
}
impl From<Elapsed> for DownloadErrorSource {
fn from(value: Elapsed) -> Self {
Self::Timeout(value)
}
}
impl From<io::Error> for DownloadErrorSource {
fn from(value: io::Error) -> Self {
Self::IoError(value)
}
}
impl From<Box<dyn error::Error + Send + Sync>> for DownloadErrorSource {
fn from(value: Box<dyn error::Error + Send + Sync>) -> Self {
Self::Other(anyhow::anyhow!(value))
}
}
impl<T: Into<DownloadErrorSource>> From<T> for DownloadError {
fn from(value: T) -> Self {
let state = DownloadState::None;
let source = Some(value.into());
Self::new(state, source)
}
}

88
src/client/error/request_error.rs

@ -0,0 +1,88 @@
use std::{error, fmt};
use bytes::Bytes;
use http::{request, response};
#[derive(Debug)]
pub(crate) struct ClientRequestError {
#[allow(dead_code)]
request: Option<request::Parts>,
response: Option<response::Parts>,
#[allow(dead_code)]
response_body: Option<Bytes>,
source: Option<anyhow::Error>,
}
impl ClientRequestError {
pub(crate) fn new( request: Option<request::Parts>
, response: Option<response::Parts>
, response_body: Option<Bytes>
, source: Option<anyhow::Error>) -> Self {
Self { request, response, response_body, source }
}
#[allow(dead_code)]
pub(crate) fn request(&self) -> Option<&request::Parts> {
self.request.as_ref()
}
#[allow(dead_code)]
pub(crate) fn response(&self) -> Option<&response::Parts> {
self.response.as_ref()
}
#[allow(dead_code)]
pub(crate) fn response_body(&self) -> Option<&Bytes> {
self.response_body.as_ref()
}
}
impl error::Error for ClientRequestError {
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
self.source.as_ref().map(|e| e.as_ref())
}
}
impl fmt::Display for ClientRequestError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!( f
, "request error: {{
request: {:?} / response: {:?} / source: {:?}
}}"
, self.request
, self.response
, self.source )
}
}
impl From<http::Error> for ClientRequestError {
fn from(value: http::Error) -> Self {
let request = None;
let response = None;
let response_body = None;
let source = Some(value.into());
Self { request, response, response_body, source }
}
}
impl From<anyhow::Error> for ClientRequestError {
fn from(value: anyhow::Error) -> Self {
let request = None;
let response = None;
let response_body = None;
let source = Some(value);
Self { request, response, response_body, source }
}
}
impl From<Box<dyn error::Error + Send + Sync>> for ClientRequestError {
fn from(value: Box<dyn error::Error + Send + Sync>) -> Self {
let request = None;
let response = None;
let response_body = None;
let source = Some(anyhow::anyhow!(value));
Self { request, response, response_body, source }
}
}

9
src/client_actor.rs

@ -4,6 +4,7 @@ mod util;
use std::{collections::HashMap, path::Path};
use bytes::Bytes;
use error::ClientActorError;
use http::{HeaderMap, Uri};
use message::{ClientActorMessage, ClientActorMessageHandle};
@ -80,7 +81,8 @@ impl ClientActor {
self.tasks.spawn(async move {
let result = client.download(handle.filename(), &handle.uri(), &HeaderMap::new()).await;
match result {
Err(source) => Err(ClientActorError::new(&handle, source)),
Err(source) =>
Err(ClientActorError::new(&handle, source.into())),
Ok(state) => {
handle.set_state(state);
Ok(Some(handle))
@ -102,7 +104,8 @@ impl ClientActor {
self.tasks.spawn(async move {
let result = client.data(&handle.uri(), &HeaderMap::new()).await;
match result {
Err(source) => Err(ClientActorError::new(&handle, source)),
Err(source) =>
Err(ClientActorError::new(&handle, source.into())),
Ok(data) => {
*handle.buffer_mut() = Some(data);
Ok(Some(handle))
@ -148,7 +151,7 @@ impl ClientActorHandle {
receive.await.expect("Actor cancelled unexpected")
}
pub(super) async fn body_bytes(&self, uri: &Uri) -> Option<Vec<u8>> {
pub(super) async fn body_bytes(&self, uri: &Uri) -> Option<Bytes> {
let uri = uri.to_owned();
let (send, receive) = oneshot::channel();
let msg = ClientActorMessage::GetData { uri, respond_to: send };

30
src/client_actor/error.rs

@ -1,30 +1,44 @@
use std::{error, fmt};
use crate::client::error as client_error;
use super::message::ClientActorMessageHandle;
#[derive(Debug)]
pub(crate) enum ClientActorErrorSource {
#[allow(dead_code)]
Download(client_error::DownloadError),
}
#[derive(Debug)]
pub(crate) struct ClientActorError {
pub(super) action: ClientActorMessageHandle,
pub(super) source: anyhow::Error,
pub(super) source: Option<ClientActorErrorSource>,
}
impl From<client_error::DownloadError> for Option<ClientActorErrorSource> {
fn from(value: client_error::DownloadError) -> Self {
Some(ClientActorErrorSource::Download(value))
}
}
impl ClientActorError {
pub(super) fn new( action: &ClientActorMessageHandle
, source: anyhow::Error ) -> Self {
, source: Option<ClientActorErrorSource> ) -> Self {
let action = action.to_owned();
Self { action, source }
}
}
impl error::Error for ClientActorError {
fn source(&self) -> Option<&(dyn error::Error + 'static)> {
Some(self.source.as_ref())
}
}
impl error::Error for ClientActorError {}
impl fmt::Display for ClientActorError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "download error ({:?}): {}", self.action, self.source)
write!( f
, "client actor error: {{ action: {:?} / source: {:?} }}"
, self.action
, self.source )
}
}

7
src/client_actor/message.rs

@ -1,5 +1,6 @@
use std::path::PathBuf;
use bytes::Bytes;
use http::Uri;
use tokio::sync::oneshot;
@ -18,7 +19,7 @@ pub(super) enum ClientActorMessage {
},
GetData {
uri: Uri,
respond_to: oneshot::Sender<Option<Vec<u8>>>,
respond_to: oneshot::Sender<Option<Bytes>>,
},
}
@ -32,7 +33,7 @@ pub(super) enum ClientActorMessageHandle {
},
GetData {
uri: Uri,
buffer: Option<Vec<u8>>,
buffer: Option<Bytes>,
message: ActionIndex,
},
}
@ -60,7 +61,7 @@ impl ClientActorMessageHandle {
}
}
pub(super) fn buffer_mut(&mut self) -> &mut Option<Vec<u8>> {
pub(super) fn buffer_mut(&mut self) -> &mut Option<Bytes> {
match self {
Self::GetData { ref mut buffer, .. } => buffer,
_ => panic!("Called with invalid variant"),

8
src/client_actor/util.rs

@ -26,7 +26,9 @@ async fn process_next_result(mut actor: ClientActor, result: ClientTaskResult) -
, &e.action.uri()
, &HeaderMap::new()).await;
match result {
Err(source) => Err(ClientActorError::new(&e.action, source)),
Err(source) =>
Err(ClientActorError::new( &e.action
, source.into() )),
Ok(state) => {
e.action.set_state(state);
Ok(Some(e.action))
@ -36,7 +38,9 @@ async fn process_next_result(mut actor: ClientActor, result: ClientTaskResult) -
GetData { .. } => {
let result = client.data(&e.action.uri(), &HeaderMap::new()).await;
match result {
Err(source) => Err(ClientActorError::new(&e.action, source)),
Err(source) =>
Err(ClientActorError::new( &e.action
, source.into() )),
Ok(data) => {
*e.action.buffer_mut() = Some(data);
Ok(Some(e.action))

3
src/m3u8_download.rs

@ -1,6 +1,7 @@
use std::path::{Path, PathBuf};
use anyhow::anyhow;
use bytes::Bytes;
use futures_util::future::join_all;
use http::{uri::{Authority, Scheme}, Uri};
use log::debug;
@ -60,7 +61,7 @@ impl TsPart {
}
impl M3u8Download {
pub(super) async fn new(m3u8_data: Vec<u8>, index_uri: Uri) -> anyhow::Result<Self> {
pub(super) async fn new(m3u8_data: Bytes, index_uri: Uri) -> anyhow::Result<Self> {
let scheme = index_uri.scheme()
. ok_or(anyhow!("Problem scheme in m3u8 uri"))?
. to_owned();

4
src/main.rs

@ -90,10 +90,10 @@ async fn main() -> anyhow::Result<()> {
let client = Client::new(buffer, rate_limit, concurrency_limit, timeout)?
. set_body_timeout(body_timeout)
. set_origin(args.origin);
. set_origin(args.origin)?;
let client = if let Some(user_agent) = args.agent {
client.set_user_agent(Some(user_agent))
client.set_user_agent(Some(user_agent))?
} else {
client
};

Loading…
Cancel
Save