diff --git a/src/api/download_api.rs b/src/api/download_api.rs index 68d5a620a..e483671bc 100644 --- a/src/api/download_api.rs +++ b/src/api/download_api.rs @@ -8,10 +8,11 @@ use futures::stream::TryStreamExt; use log::info; use serde_json::{json, Value}; use std::fs::File; -use std::io::{ErrorKind, Write}; +use std::io::{Write}; use std::ops::Deref; use std::sync::Arc; -use std::{fs, io}; +use std::{fs}; +use crate::m3u_filter_error::to_io_error; async fn download_file(active: Arc>>, client: &reqwest::Client) -> Result<(), String> { let file_download = { active.read().await.as_ref().unwrap().clone() }; @@ -24,7 +25,7 @@ async fn download_file(active: Arc>>, client: &reqwe match File::create(&file_download.file_path) { Ok(mut file) => { let mut downloaded: u64 = 0; - let mut stream = response.bytes_stream().map_err(|err| io::Error::new(ErrorKind::Other, err)); + let mut stream = response.bytes_stream().map_err(to_io_error); loop { match stream.try_next().await { Ok(item) => { diff --git a/src/api/xmltv_api.rs b/src/api/xmltv_api.rs index 26d0c95b1..3ec87941e 100644 --- a/src/api/xmltv_api.rs +++ b/src/api/xmltv_api.rs @@ -7,7 +7,6 @@ use quick_xml::{Reader, Writer}; use flate2::write::GzEncoder; use flate2::Compression; use quick_xml::events::{BytesStart, Event}; -use std::io::{BufReader}; use chrono::{Duration, NaiveDateTime, TimeDelta}; use crate::api::api_utils::{get_user_target, serve_file}; @@ -20,6 +19,7 @@ use crate::repository::m3u_repository::m3u_get_epg_file_path; use crate::repository::storage::get_target_storage_path; use crate::repository::xtream_repository::{xtream_get_epg_file_path, xtream_get_storage_path}; use crate::utils::{file_utils}; +use crate::utils::file_utils::file_reader; fn time_correct(date_time: &str, correction: &TimeDelta) -> String { // Split the dateTime string into date and time parts @@ -120,7 +120,7 @@ async fn serve_epg(epg_path: &Path, req: &HttpRequest, user: &ProxyUserCredentia } fn serve_epg_with_timeshift(epg_file: File, offset_minutes: i32) -> HttpResponse { - let reader = BufReader::new(epg_file); + let reader = file_reader(epg_file); let encoder = GzEncoder::new(Vec::with_capacity(4096), Compression::default()); let mut xml_reader = Reader::from_reader(reader); let mut xml_writer = Writer::new(encoder); diff --git a/src/api/xtream_api.rs b/src/api/xtream_api.rs index 8e6d361d4..51d3d157d 100644 --- a/src/api/xtream_api.rs +++ b/src/api/xtream_api.rs @@ -3,7 +3,6 @@ use crate::Arc; use std::collections::HashMap; use std::fmt::{Display, Formatter}; -use std::io::{Error, ErrorKind}; use std::path::Path; use std::rc::Rc; use std::str::FromStr; @@ -19,7 +18,7 @@ use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, res use crate::api::model::app_state::AppState; use crate::api::model::request::UserApiRequest; use crate::api::model::xtream::XtreamAuthorizationResponse; -use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; +use crate::m3u_filter_error::{str_to_io_error, M3uFilterError, M3uFilterErrorKind}; use crate::model::api_proxy::{ProxyType, ProxyUserCredentials}; use crate::model::config::TargetType; use crate::model::config::{Config, ConfigInput, ConfigTarget}; @@ -514,7 +513,7 @@ async fn xtream_player_api_handle_content_action(config: &Config, target_name: & ACTION_GET_LIVE_CATEGORIES => xtream_repository::xtream_get_collection_path(config, target_name, xtream_repository::COL_CAT_LIVE), ACTION_GET_VOD_CATEGORIES => xtream_repository::xtream_get_collection_path(config, target_name, xtream_repository::COL_CAT_VOD), ACTION_GET_SERIES_CATEGORIES => xtream_repository::xtream_get_collection_path(config, target_name, xtream_repository::COL_CAT_SERIES), - _ => Err(Error::new(ErrorKind::Other, "")) + _ => Err(str_to_io_error("")) } { if let Some(file_path) = path { let category_id = category_id.trim(); diff --git a/src/auth/authenticator.rs b/src/auth/authenticator.rs index fededceae..188823fbe 100644 --- a/src/auth/authenticator.rs +++ b/src/auth/authenticator.rs @@ -4,6 +4,7 @@ use chrono::{Local, Duration}; use jsonwebtoken::{Algorithm, DecodingKey, encode, decode, EncodingKey, Header, Validation}; use crate::model::config::WebAuthConfig; use crate::api::model::app_state::AppState; +use crate::m3u_filter_error::to_io_error; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct Claims { @@ -25,7 +26,7 @@ pub fn create_jwt(web_auth_config: &WebAuthConfig) -> Result Ok(jwt), - Err(err) => Err(std::io::Error::new(std::io::ErrorKind::Other, err.to_string())) + Err(err) => Err(to_io_error(err)) } } diff --git a/src/auth/password.rs b/src/auth/password.rs index 91c7db1c1..2e0a65edf 100644 --- a/src/auth/password.rs +++ b/src/auth/password.rs @@ -1,5 +1,5 @@ -use std::io::ErrorKind; use rand::{Rng, distributions::Alphanumeric, rngs::OsRng}; +use crate::m3u_filter_error::str_to_io_error; fn generate_salt(length: usize) -> String { let rng = OsRng; @@ -33,14 +33,14 @@ pub fn generate_password() -> std::io::Result { match rpassword::prompt_password("password> ") { Ok(pwd1) => { if pwd1.len() < 8 { - return Err(std::io::Error::new(ErrorKind::Other, "Password too short min length 8")) + return Err(str_to_io_error("Password too short min length 8")) } match rpassword::prompt_password("retype password> ") { Ok(pwd2) => { if pwd1.eq(&pwd2) { - hash(pwd1.as_bytes()).map_or_else(|| Err(std::io::Error::new(ErrorKind::Other, "Failed to generate hash")), Ok) + hash(pwd1.as_bytes()).map_or_else(|| Err(str_to_io_error("Failed to generate hash")), Ok) } else { - Err(std::io::Error::new(ErrorKind::Other, "Passwords don't match")) + Err(str_to_io_error("Passwords don't match")) } } Err(err) => Err(err) diff --git a/src/filter.rs b/src/filter.rs index ac7938d87..a27b86836 100644 --- a/src/filter.rs +++ b/src/filter.rs @@ -102,7 +102,7 @@ struct FilterParser; #[derive(Debug, Clone)] pub enum UnaryOperator { - Not + Not, } #[derive(Debug, Clone)] @@ -155,33 +155,29 @@ impl Filter { Self::TypeComparison(field, item_type) => { let value = provider.call(field); get_filter_item_type(value.as_str()).is_some_and(|pli_type| { - let is_match = pli_type.eq(item_type); - if log_enabled!(Level::Trace) { - if is_match { - debug!("Match found: {:?} {}", &field, value); - } else { - debug!("Match failed: {self}: {:?} {}", &field, &value); - } + let is_match = pli_type.eq(item_type); + if log_enabled!(Level::Trace) { + if is_match { + debug!("Match found: {:?} {}", &field, value); + } else { + debug!("Match failed: {self}: {:?} {}", &field, &value); } - is_match - }) + } + is_match + }) } - Self::Group(expr) => { - expr.filter(provider, processor) - } - Self::UnaryExpression(op, expr) => { - match op { - UnaryOperator::Not => !expr.filter(provider, processor), + Self::Group(expr) => expr.filter(provider, processor), + Self::UnaryExpression(op, expr) => match op { + UnaryOperator::Not => !expr.filter(provider, processor), + }, + Self::BinaryExpression(left, op, right) => match op { + BinaryOperator::And => { + left.filter(provider, processor) && right.filter(provider, processor) } - } - Self::BinaryExpression(left, op, right) => { - match op { - BinaryOperator::And => left.filter(provider, processor) - && right.filter(provider, processor), - BinaryOperator::Or => left.filter(provider, processor) - || right.filter(provider, processor), + BinaryOperator::Or => { + left.filter(provider, processor) || right.filter(provider, processor) } - } + }, } } } @@ -235,7 +231,10 @@ fn get_parser_item_field(expr: &Pair) -> Result create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "unknown field: {}", expr.as_str()) } -fn get_parser_regexp(expr: &Pair, templates: &Vec) -> Result { +fn get_parser_regexp( + expr: &Pair, + templates: &Vec, +) -> Result { if expr.as_rule() == Rule::regexp { let mut parsed_text = String::from(expr.as_str()); parsed_text.pop(); @@ -246,8 +245,12 @@ fn get_parser_regexp(expr: &Pair, templates: &Vec) -> Res return create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "cant parse regex: {}", regstr); } let regexp = re.unwrap(); - let captures = regexp.capture_names() - .flatten().map(String::from).filter(|x| !x.is_empty()).collect::>(); + let captures = regexp + .capture_names() + .flatten() + .map(String::from) + .filter(|x| !x.is_empty()) + .collect::>(); if log_enabled!(Level::Trace) { trace!("Created regex: {} with captures: [{}]", regstr, captures.join(", ")); } @@ -260,23 +263,27 @@ fn get_parser_regexp(expr: &Pair, templates: &Vec) -> Res create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "unknown field: {}", expr.as_str()) } -fn get_parser_field_comparison(expr: Pair, templates: &Vec) -> Result { +fn get_parser_field_comparison( + expr: Pair, + templates: &Vec, +) -> Result { let mut expr_inner = expr.into_inner(); match get_parser_item_field(&expr_inner.next().unwrap()) { - Ok(field) => { - match get_parser_regexp(&expr_inner.next().unwrap(), templates) { - Ok(regexp) => Ok(Filter::FieldComparison(field, regexp)), - Err(err) => Err(err), - } - } - Err(err) => Err(err) + Ok(field) => match get_parser_regexp(&expr_inner.next().unwrap(), templates) { + Ok(regexp) => Ok(Filter::FieldComparison(field, regexp)), + Err(err) => Err(err), + }, + Err(err) => Err(err), } } fn get_filter_item_type(text_item_type: &str) -> Option { if text_item_type.eq_ignore_ascii_case("live") { Some(PlaylistItemType::Live) - } else if text_item_type.eq_ignore_ascii_case("movie") || text_item_type.eq_ignore_ascii_case("video") || text_item_type.eq_ignore_ascii_case("vod") { + } else if text_item_type.eq_ignore_ascii_case("movie") + || text_item_type.eq_ignore_ascii_case("video") + || text_item_type.eq_ignore_ascii_case("vod") + { Some(PlaylistItemType::Video) } else if text_item_type.eq_ignore_ascii_case("series") { Some(PlaylistItemType::Series) @@ -297,28 +304,30 @@ fn get_parser_type_comparison(expr: Pair) -> Result { - { - let result = match $bop { - Some(binop) => { - let lhs = $stmts.pop().unwrap(); - $bop = None; - Filter::BinaryExpression(Box::new(lhs), binop.clone(), Box::new($exp)) + ($bop: expr, $uop: expr, $stmts: expr, $exp: expr) => {{ + let result = match $bop { + Some(binop) => { + let lhs = $stmts.pop().unwrap(); + $bop = None; + Filter::BinaryExpression(Box::new(lhs), binop.clone(), Box::new($exp)) + } + _ => match $uop { + Some(unop) => { + $uop = None; + Filter::UnaryExpression(unop.clone(), Box::new($exp)) } - _ => match $uop { - Some(unop) => { - $uop = None; - Filter::UnaryExpression(unop.clone(), Box::new($exp)) - } - _ => $exp - } - }; - $stmts.push(result); - } - } + _ => $exp, + }, + }; + $stmts.push(result); + }}; } -fn get_parser_expression(expr: Pair, templates: &Vec, errors: &mut Vec) -> Filter { +fn get_parser_expression( + expr: Pair, + templates: &Vec, + errors: &mut Vec, +) -> Filter { let mut stmts = Vec::with_capacity(128); let pairs = expr.into_inner(); let mut bop: Option = None; @@ -349,16 +358,14 @@ fn get_parser_expression(expr: Pair, templates: &Vec, err Rule::not => { uop = Some(UnaryOperator::Not); } - Rule::bool_op => { - match get_parser_binary_op(&pair.into_inner().next().unwrap()) { - Ok(binop) => { - bop = Some(binop); - } - Err(err) => { - errors.push(format!("{err}")); - } + Rule::bool_op => match get_parser_binary_op(&pair.into_inner().next().unwrap()) { + Ok(binop) => { + bop = Some(binop); } - } + Err(err) => { + errors.push(format!("{err}")); + } + }, _ => { errors.push(format!("did not expect rule: {pair:?}")); } @@ -378,11 +385,18 @@ fn get_parser_binary_op(expr: &Pair) -> Result Ok(BinaryOperator::And), Rule::or => Ok(BinaryOperator::Or), - _ => create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "Unknown binary operator {}", expr.as_str()) + _ => create_m3u_filter_error_result!( + M3uFilterErrorKind::Info, + "Unknown binary operator {}", + expr.as_str() + ), } } -pub fn get_filter(filter_text: &str, templates: Option<&Vec>) -> Result { +pub fn get_filter( + filter_text: &str, + templates: Option<&Vec>, +) -> Result { let empty_list = Vec::with_capacity(0); let template_list: &Vec = templates.unwrap_or(&empty_list); let source = apply_templates_to_pattern(filter_text, template_list); @@ -398,13 +412,18 @@ pub fn get_filter(filter_text: &str, templates: Option<&Vec>) - for expr in pair.into_inner() { match expr.as_rule() { Rule::expr => { - let expr = get_parser_expression(expr, template_list, &mut errors); + let expr = + get_parser_expression(expr, template_list, &mut errors); match &op { Some(binop) => { - result = Some(Filter::BinaryExpression(Box::new(result.unwrap()), binop.clone(), Box::new(expr))); + result = Some(Filter::BinaryExpression( + Box::new(result.unwrap()), + binop.clone(), + Box::new(expr), + )); op = None; } - _ => result = Some(expr) + _ => result = Some(expr), } } Rule::bool_op => { @@ -435,18 +454,30 @@ pub fn get_filter(filter_text: &str, templates: Option<&Vec>) - return Err(info_err!(errors.join("\n"))); } - result.map_or_else(|| create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "Unable to parse filter: {}", &filter_text), Ok) + result.map_or_else( + || { + create_m3u_filter_error_result!( + M3uFilterErrorKind::Info, + "Unable to parse filter: {}", + &filter_text + ) + }, + Ok, + ) } - Err(err) => create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "{}", err) + Err(err) => create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "{}", err), } } -fn build_dependency_graph(templates: &Vec) -> Result, M3uFilterError> { +fn build_dependency_graph( + templates: &Vec, +) -> Result, M3uFilterError> { let mut graph = DirectedGraph::::new(); - let re_template= regex::Regex::new("!(.*?)!").unwrap(); + let re_template = regex::Regex::new("!(.*?)!").unwrap(); for template in templates { graph.add_node(&template.name); - re_template.captures_iter(&template.value) + re_template + .captures_iter(&template.value) .filter(|caps| caps.len() > 1) .filter_map(|caps| caps.get(1)) .map(|caps| String::from(caps.as_str())) @@ -457,18 +488,27 @@ fn build_dependency_graph(templates: &Vec) -> Result ")); + error!( + "Cyclic template dependencies detected [{}]", + cyclic.join(" <-> ") + ); } if !cycles.is_empty() { - return create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "Cyclic dependencies in templates detected!"); + return create_m3u_filter_error_result!( + M3uFilterErrorKind::Info, + "Cyclic dependencies in templates detected!" + ); } Ok(graph) } -pub fn prepare_templates(templates: &Vec) -> Result, M3uFilterError> { +pub fn prepare_templates( + templates: &Vec, +) -> Result, M3uFilterError> { let graph = build_dependency_graph(templates)?; let mut template_values = HashMap::::new(); - let mut template_map: HashMap = templates.iter() + let mut template_map: HashMap = templates + .iter() .map(|item| { template_values.insert(item.name.clone(), item.value.clone()); (item.name.clone(), item.clone()) @@ -482,7 +522,8 @@ pub fn prepare_templates(templates: &Vec) -> Result) -> Result) -> String { let mut new_pattern = pattern.to_string(); for template in templates { - new_pattern = new_pattern.replace(format!("!{}!", &template.name).as_str(), &template.value); + new_pattern = + new_pattern.replace(format!("!{}!", &template.name).as_str(), &template.value); } new_pattern } - #[cfg(test)] mod tests { use std::cell::RefCell; @@ -522,7 +563,7 @@ mod tests { name: Rc::new(name.to_string()), group: Rc::new(group.to_string()), ..Default::default() - }) + }), } } @@ -577,26 +618,40 @@ mod tests { create_mock_pli("Entertainment", "US Channels"), ]; let mut processor = MockValueProcessor {}; - let filtered: Vec<&PlaylistItem> = channels.iter().filter(|&chan| { - let provider = ValueProvider { pli: RefCell::new(chan) }; - filter.filter(&provider, &mut processor) - }).collect(); + let filtered: Vec<&PlaylistItem> = channels + .iter() + .filter(|&chan| { + let provider = ValueProvider { + pli: RefCell::new(chan), + }; + filter.filter(&provider, &mut processor) + }) + .collect(); assert_eq!(filtered.len(), 2); - assert_eq!(filtered.iter().any(|&chan| { - let group = chan.header.borrow().group.to_string(); - let name = chan.header.borrow().name.to_string(); - name.eq("24/7: Cars") && group.eq("FR Channels") - }), true); - assert_eq!(filtered.iter().any(|&chan| { - let group = chan.header.borrow().group.to_string(); - let name = chan.header.borrow().name.to_string(); - name.eq("Entertainment") && group.eq("US Channels") - }), true); - assert_eq!(filtered.iter().any(|&chan| { - let group = chan.header.borrow().group.to_string(); - let name = chan.header.borrow().name.to_string(); - name.eq("24/7: Cars") && group.eq("US Channels") - }), false); + assert_eq!( + filtered.iter().any(|&chan| { + let group = chan.header.borrow().group.to_string(); + let name = chan.header.borrow().name.to_string(); + name.eq("24/7: Cars") && group.eq("FR Channels") + }), + true + ); + assert_eq!( + filtered.iter().any(|&chan| { + let group = chan.header.borrow().group.to_string(); + let name = chan.header.borrow().name.to_string(); + name.eq("Entertainment") && group.eq("US Channels") + }), + true + ); + assert_eq!( + filtered.iter().any(|&chan| { + let group = chan.header.borrow().group.to_string(); + let name = chan.header.borrow().name.to_string(); + name.eq("24/7: Cars") && group.eq("US Channels") + }), + false + ); } Err(e) => { panic!("{}", e) @@ -619,10 +674,15 @@ mod tests { create_mock_pli("NA", "GC"), ]; let mut processor = MockValueProcessor {}; - let filtered: Vec<&PlaylistItem> = channels.iter().filter(|&chan| { - let provider = ValueProvider { pli: RefCell::new(chan) }; - filter.filter(&provider, &mut processor) - }).collect(); + let filtered: Vec<&PlaylistItem> = channels + .iter() + .filter(|&chan| { + let provider = ValueProvider { + pli: RefCell::new(chan), + }; + filter.filter(&provider, &mut processor) + }) + .collect(); assert_eq!(filtered.len(), 1); } Err(e) => { @@ -663,4 +723,4 @@ mod tests { } } } -} \ No newline at end of file +} diff --git a/src/m3u_filter_error.rs b/src/m3u_filter_error.rs index 671e8c51c..1bd27e3f3 100644 --- a/src/m3u_filter_error.rs +++ b/src/m3u_filter_error.rs @@ -7,18 +7,19 @@ macro_rules! get_errors_notify_message { if $errors.is_empty() { None } else { - let text = $errors.iter() - .filter(|&err| err.kind == M3uFilterErrorKind::Notify) - .map(|err| err.message.as_str()) - .collect::>() - .join("\n"); - if $size > 0 && text.len() > std::cmp::max($size-3, 3) { + let text = $errors + .iter() + .filter(|&err| err.kind == M3uFilterErrorKind::Notify) + .map(|err| err.message.as_str()) + .collect::>() + .join("\n"); + if $size > 0 && text.len() > std::cmp::max($size - 3, 3) { Some(format!("{}...", text.get(0..$size).unwrap())) } else { Some(text) } } - } + }; } #[macro_export] @@ -46,15 +47,11 @@ pub enum M3uFilterErrorKind { pub struct M3uFilterError { pub kind: M3uFilterErrorKind, pub message: String, - } impl M3uFilterError { pub const fn new(kind: M3uFilterErrorKind, message: String) -> Self { - Self { - kind, - message, - } + Self { kind, message } } } @@ -66,3 +63,11 @@ impl Display for M3uFilterError { impl Error for M3uFilterError {} +pub fn to_io_error(err: E) -> std::io::Error +where E: std::error::Error { + std::io::Error::new(std::io::ErrorKind::Other, err.to_string()) +} + +pub fn str_to_io_error(err: &str) -> std::io::Error { + std::io::Error::new(std::io::ErrorKind::Other, err.to_string()) +} diff --git a/src/main.rs b/src/main.rs index af230c3f9..a96e83e10 100644 --- a/src/main.rs +++ b/src/main.rs @@ -18,20 +18,26 @@ use crate::utils::{config_reader, file_utils}; use clap::Parser; use env_logger::Builder; use log::{error, info, LevelFilter}; -mod m3u_filter_error; -mod model; -mod filter; -mod repository; -mod messaging; mod api; -mod processing; -mod utils; mod auth; +mod filter; +mod m3u_filter_error; +mod messaging; +mod model; +mod processing; +mod repository; +mod utils; -const LOG_ERROR_LEVEL_MOD: &[&str] = &["actix_web::middleware::logger", - "reqwest::async_impl::client", "reqwest::connect", "hyper_util::client", "actix_server::worker", - "actix_server::server", "actix_server::builder", "actix_server::accept"]; - +const LOG_ERROR_LEVEL_MOD: &[&str] = &[ + "actix_web::middleware::logger", + "reqwest::async_impl::client", + "reqwest::connect", + "hyper_util::client", + "actix_server::worker", + "actix_server::server", + "actix_server::builder", + "actix_server::accept", +]; #[derive(Parser)] #[command(name = "m3u-filter")] @@ -77,7 +83,6 @@ struct Args { #[arg(short = None, long = "healthcheck", default_value_t = false, default_missing_value = "true" )] healthcheck: bool, - } const VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -108,7 +113,7 @@ fn main() { if args.genpwd { match generate_password() { Ok(pwd) => println!("{pwd}"), - Err(err) => error!("{err}") + Err(err) => error!("{err}"), } return; } @@ -137,7 +142,6 @@ fn main() { Err(err) => exit!("{err}"), } - let mut temp_path = PathBuf::from(&cfg.working_dir); temp_path.push("tmp"); let _ = tempfile::env::override_temp_dir(&temp_path); @@ -240,4 +244,4 @@ fn healthcheck(config_file: &str) { } std::process::exit(1); -} \ No newline at end of file +} diff --git a/src/messaging.rs b/src/messaging.rs index 1abd5d911..93fbecdc4 100644 --- a/src/messaging.rs +++ b/src/messaging.rs @@ -1,6 +1,6 @@ +use crate::model::config::MessagingConfig; use log::{debug, error}; use reqwest::header; -use crate::model::config::{MessagingConfig}; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)] pub enum MsgKind { @@ -24,12 +24,11 @@ pub fn send_message(kind: &MsgKind, cfg: Option<&MessagingConfig>, msg: &str) { if let Some(telegram) = &messaging.telegram { for chat_id in &telegram.chat_ids { let bot = rustelebot::create_instance(&telegram.bot_token, chat_id); - match rustelebot::send_message(&bot, msg, None) - { + match rustelebot::send_message(&bot, msg, None) { Ok(()) => debug!("Text message sent successfully to {}", chat_id), Err(e) => error!("Text message wasn't sent to {} because of: {}", chat_id, e) } - }; + } } if let Some(rest) = &messaging.rest { @@ -37,17 +36,18 @@ pub fn send_message(kind: &MsgKind, cfg: Option<&MessagingConfig>, msg: &str) { let data = msg.to_owned(); actix_rt::spawn(async move { let client = reqwest::Client::new(); - match client.post(&url) + match client + .post(&url) .header(header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) .body(data) .send() - .await { + .await + { Ok(_) => debug!("Text message sent successfully to rest api"), - Err(e) => error!("Text message wasn't sent to rest api because of: {}", e) + Err(e) => error!("Text message wasn't sent to rest api because of: {}", e), } }); } } } } - diff --git a/src/model/config.rs b/src/model/config.rs index e6674d0c9..4ac14083d 100644 --- a/src/model/config.rs +++ b/src/model/config.rs @@ -24,6 +24,7 @@ use crate::utils::default_utils::{default_as_default, default_as_true, default_a use crate::utils::file_lock_manager::FileLockManager; use crate::utils::{config_reader, file_utils}; use crate::{exit, info_err}; +use crate::utils::file_utils::file_reader; use crate::utils::size_utils::parse_size_base_2; pub const MAPPER_ATTRIBUTE_FIELDS: &[&str] = &[ @@ -818,7 +819,7 @@ impl WebAuthConfig { if let Ok(file) = File::open(&userfile_path) { let mut users = vec![]; - let reader = std::io::BufReader::new(file); + let reader = file_reader(file); for credentials in reader.lines().map_while(Result::ok) { let mut parts = credentials.split(':'); if let (Some(username), Some(password)) = (parts.next(), parts.next()) { diff --git a/src/processing/playlist_processor.rs b/src/processing/playlist_processor.rs index 5d11062db..043729f14 100644 --- a/src/processing/playlist_processor.rs +++ b/src/processing/playlist_processor.rs @@ -300,9 +300,9 @@ async fn process_source(client: Arc, cfg: Arc, source_i let enabled_inputs = source.inputs.iter().filter(|item| item.enabled).count(); // Downlod the sources for input in &source.inputs { - let start_time = Instant::now(); let input_id = input.id; if is_input_enabled(enabled_inputs, input.enabled, input_id, &user_targets) { + let start_time = Instant::now(); let (mut playlistgroups, mut error_list) = match input.input_type { InputType::M3u => download::get_m3u_playlist(Arc::clone(&client), &cfg, input, &cfg.working_dir).await, InputType::Xtream => download::get_xtream_playlist(Arc::clone(&client), input, &cfg.working_dir).await, diff --git a/src/processing/xtream_processor.rs b/src/processing/xtream_processor.rs index 73852e5b5..81a958c3f 100644 --- a/src/processing/xtream_processor.rs +++ b/src/processing/xtream_processor.rs @@ -1,4 +1,4 @@ -use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; +use crate::m3u_filter_error::{str_to_io_error, to_io_error, M3uFilterError, M3uFilterErrorKind}; use crate::model::config::{Config, ConfigInput}; use crate::model::playlist::{FetchedPlaylist, PlaylistEntry, PlaylistItem, PlaylistItemType, XtreamCluster}; use crate::repository::storage::get_input_storage_path; @@ -7,7 +7,7 @@ use crate::{info_err, notify_err}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::fs::File; -use std::io::{BufWriter, Error, ErrorKind, Write}; +use std::io::{BufWriter, Write}; use std::path::PathBuf; use std::sync::Arc; @@ -73,7 +73,7 @@ pub(in crate::processing) async fn playlist_resolve_download_playlist_item(clien } pub(in crate::processing) fn write_info_content_to_wal_file(writer: &mut BufWriter<&File>, provider_id: u32, content: &str) -> std::io::Result<()> { - let length = u32::try_from(content.len()).map_err(|err| Error::new(ErrorKind::Other, err))?; + let length = u32::try_from(content.len()).map_err(to_io_error)?; if length > 0 { writer.write_all(&provider_id.to_le_bytes())?; writer.write_all(&length.to_le_bytes())?; @@ -131,7 +131,7 @@ where let mut processed_info_ids = HashMap::new(); let file_path = match get_input_storage_path(fpl.input, &cfg.working_dir) - .map(|storage_path| xtream_get_record_file_path(&storage_path, item_type)).and_then(|opt| opt.ok_or_else(|| Error::new(ErrorKind::Other, "Not supported".to_string()))) + .map(|storage_path| xtream_get_record_file_path(&storage_path, item_type)).and_then(|opt| opt.ok_or_else(|| str_to_io_error("Not supported"))) { Ok(file_path) => file_path, Err(err) => { diff --git a/src/processing/xtream_processor_series.rs b/src/processing/xtream_processor_series.rs index df80abf3b..25ee35312 100644 --- a/src/processing/xtream_processor_series.rs +++ b/src/processing/xtream_processor_series.rs @@ -15,6 +15,7 @@ use std::sync::Arc; use std::time::Instant; use log::{info, log_enabled, Level}; use crate::model::xtream::{XtreamSeriesEpisode, XtreamSeriesInfoEpisode}; +use crate::utils::file_utils::file_writer; const TAG_SERIES_INFO_LAST_MODIFIED: &str = "last_modified"; @@ -62,8 +63,8 @@ async fn playlist_resolve_series_info(client: Arc, cfg: &Config let Some((wal_content_file, wal_record_file, wal_content_path, wal_record_path)) = create_resolve_info_wal_files(cfg, fpl.input, XtreamCluster::Series) else { return !processed_info_ids.is_empty(); }; - let mut content_writer = BufWriter::new(&wal_content_file); - let mut record_writer = BufWriter::new(&wal_record_file); + let mut content_writer = file_writer(&wal_content_file); + let mut record_writer = file_writer(&wal_record_file); let mut content_updated = false; let series_info_iter = fpl.playlistgroups.iter() @@ -149,7 +150,7 @@ async fn process_series_info( errors.push(notify_err!("Could not create wal file for series episodes record".to_string())); return result; }; - let mut wal_writer = BufWriter::new(&wal_file); + let mut wal_writer = file_writer(&wal_file); for plg in fpl .playlistgroups diff --git a/src/processing/xtream_processor_vod.rs b/src/processing/xtream_processor_vod.rs index eb54d5174..c2a548f18 100644 --- a/src/processing/xtream_processor_vod.rs +++ b/src/processing/xtream_processor_vod.rs @@ -12,6 +12,7 @@ use std::io::{BufWriter, Write}; use std::sync::Arc; use std::time::Instant; use log::{info, log_enabled, Level}; +use crate::utils::file_utils::file_writer; const TAG_VOD_INFO_INFO: &str = "info"; const TAG_VOD_INFO_MOVIE_DATA: &str = "movie_data"; @@ -75,8 +76,8 @@ pub async fn playlist_resolve_vod(client: Arc, cfg: &Config, ta else { return; }; let mut processed_info_ids = read_processed_vod_info_ids(cfg, errors, fpl).await; - let mut content_writer = BufWriter::new(&wal_content_file); - let mut record_writer = BufWriter::new(&wal_record_file); + let mut content_writer = file_writer(&wal_content_file); + let mut record_writer = file_writer(&wal_record_file); let mut content_updated = false; let vod_info_iter = fpl.playlistgroups.iter() diff --git a/src/repository/bplustree.rs b/src/repository/bplustree.rs index e5228abe5..ca627e4cd 100644 --- a/src/repository/bplustree.rs +++ b/src/repository/bplustree.rs @@ -1,6 +1,5 @@ -use std::array::TryFromSliceError; use std::fs::{File}; -use std::io::{self, BufReader, BufWriter, Error, ErrorKind, Read, Seek, SeekFrom, Write}; +use std::io::{self, BufReader, Read, Seek, SeekFrom, Write}; use std::marker::PhantomData; use std::mem::size_of; use std::path::Path; @@ -10,7 +9,8 @@ use ruzstd::decoding::StreamingDecoder; use ruzstd::encoding::{compress_to_vec, CompressionLevel}; use serde::{Deserialize, Serialize}; use tempfile::NamedTempFile; -use crate::utils::file_utils::{open_read_write_file, rename_or_copy}; +use crate::m3u_filter_error::{str_to_io_error, to_io_error}; +use crate::utils::file_utils::{file_reader, file_writer, open_read_write_file, rename_or_copy}; const BLOCK_SIZE: usize = 4096; const BINCODE_OVERHEAD: usize = 8; @@ -36,7 +36,7 @@ fn is_file_valid(file: File) -> io::Result { #[inline] fn u32_from_bytes(bytes: &[u8]) -> io::Result { - Ok(u32::from_le_bytes(bytes.try_into().map_err(|e: TryFromSliceError| io::Error::new(io::ErrorKind::Other, e.to_string()))?)) + Ok(u32::from_le_bytes(bytes.try_into().map_err(to_io_error)?)) } #[inline] @@ -44,7 +44,7 @@ fn bincode_serialize(value: &T) -> io::Result> where T: ?Sized + serde::Serialize, { - bincode::serialize(value).map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string())) + bincode::serialize(value).map_err(to_io_error) } #[inline] @@ -52,7 +52,7 @@ fn bincode_deserialize(value: &[u8]) -> io::Result where T: for<'a> serde::Deserialize<'a>, { - bincode::deserialize(value).map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string())) + bincode::deserialize(value).map_err(to_io_error) } @@ -221,7 +221,7 @@ where // Serialize and write keys let keys_encoded = bincode_serialize(&self.keys)?; let keys_bytes_len = keys_encoded.len(); - buffer_slice[write_pos..write_pos + LEN_SIZE].copy_from_slice(&(u32::try_from(keys_bytes_len).map_err(|err| Error::new(ErrorKind::Other, err))?).to_le_bytes()); + buffer_slice[write_pos..write_pos + LEN_SIZE].copy_from_slice(&(u32::try_from(keys_bytes_len).map_err(to_io_error)?).to_le_bytes()); write_pos += LEN_SIZE; buffer_slice[write_pos..write_pos + keys_bytes_len].copy_from_slice(&keys_encoded); write_pos += keys_bytes_len; @@ -243,7 +243,7 @@ where values_encoded }; let values_bytes_len = content_bytes.len(); - buffer_slice[write_pos..write_pos + LEN_SIZE].copy_from_slice(&(u32::try_from(values_bytes_len).map_err(|err| Error::new(ErrorKind::Other, err))?).to_le_bytes()); + buffer_slice[write_pos..write_pos + LEN_SIZE].copy_from_slice(&(u32::try_from(values_bytes_len).map_err(to_io_error)?).to_le_bytes()); write_pos += LEN_SIZE; let mut value_bytes_to_write = values_bytes_len; @@ -284,7 +284,7 @@ where } let pointer_encoded = bincode_serialize(&pointer)?; - let pointer_bytes_len = u32::try_from(pointer_encoded.len()).map_err(|err| Error::new(ErrorKind::Other, err))?; + let pointer_bytes_len = u32::try_from(pointer_encoded.len()).map_err(to_io_error)?; file.seek(SeekFrom::Start(pointer_offset))?; file.write_all(&pointer_bytes_len.to_le_bytes())?; @@ -466,14 +466,14 @@ where pub fn store(&mut self, filepath: &Path) -> io::Result { if self.dirty { let tempfile = NamedTempFile::new()?; - let mut file = BufWriter::new(&tempfile); //create_new_file_for_write(&tempfile)?); + let mut file = file_writer(&tempfile); //create_new_file_for_write(&tempfile)?); let mut buffer = vec![0u8; BLOCK_SIZE]; match self.root.serialize_to_block(&mut file, &mut buffer, 0u64) { Ok(result) => { file.flush()?; drop(file); if let Err(err) = rename_or_copy(tempfile.path(), filepath, false) { - return Err(Error::new(io::ErrorKind::Other, format!("Temp file rename/copy did not work {} {err}", tempfile.path().to_string_lossy()))); + return Err(str_to_io_error(&format!("Temp file rename/copy did not work {} {err}", tempfile.path().to_string_lossy()))); } self.dirty = false; Ok(result) @@ -489,7 +489,7 @@ where pub fn load(filepath: &Path) -> io::Result { let file = is_file_valid(File::open(filepath)?)?; - let mut reader = BufReader::new(file); + let mut reader = file_reader(file); let mut buffer = vec![0u8; BLOCK_SIZE]; let (root, _) = BPlusTreeNode::deserialize_from_block(&mut reader, &mut buffer, 0, true)?; Ok(Self::new_with_root(root)) @@ -574,7 +574,7 @@ where pub fn try_from_file(file: File) -> io::Result { let file = is_file_valid(file)?; Ok(Self { - file: BufReader::new(file), + file: file_reader(file), _marker_k: PhantomData, _marker_v: PhantomData, }) @@ -621,7 +621,7 @@ where } pub fn query(&mut self, key: &K) -> Option { - let mut reader = BufReader::new(&mut self.file); + let mut reader = file_reader(&mut self.file); query_tree(&mut reader, key) } @@ -635,7 +635,7 @@ where pub fn update(&mut self, key: &K, value: V) -> io::Result { let mut offset = 0; let mut buffer = vec![0u8; BLOCK_SIZE]; - let mut reader = BufReader::new(&mut self.file); + let mut reader = file_reader(&mut self.file); loop { match BPlusTreeNode::::deserialize_from_block(&mut reader, &mut buffer, offset, false) { Ok((mut node, pointers)) => { diff --git a/src/repository/indexed_document.rs b/src/repository/indexed_document.rs index b9a1b3800..77ca997ef 100644 --- a/src/repository/indexed_document.rs +++ b/src/repository/indexed_document.rs @@ -1,6 +1,6 @@ use std::fmt::Debug; use std::fs::{File}; -use std::io::{BufReader, BufWriter, Error, ErrorKind, Read, Seek, SeekFrom, Write}; +use std::io::{BufReader, Error, ErrorKind, Read, Seek, SeekFrom, Write}; use std::marker::PhantomData; use std::path::{Path, PathBuf}; @@ -9,7 +9,8 @@ use crate::utils::file_utils; use log::error; use serde::{Deserialize, Serialize}; use tempfile::NamedTempFile; -use crate::utils::file_utils::{create_new_file_for_read_write, open_read_write_file, open_readonly_file, rename_or_copy}; +use crate::m3u_filter_error::{str_to_io_error, to_io_error}; +use crate::utils::file_utils::{create_new_file_for_read_write, file_reader, file_writer, open_read_write_file, open_readonly_file, rename_or_copy}; const BLOCK_SIZE: usize = 4096; const LEN_SIZE: usize = 4; @@ -100,7 +101,7 @@ where // Retrieve file size and convert to `u32` for `main_offset`, if possible let mut main_offset = main_file .metadata() - .and_then(|meta| SizeType::try_from(meta.len()).map_err(|err| Error::new(ErrorKind::Other, err))) + .and_then(|meta| SizeType::try_from(meta.len()).map_err(to_io_error)) .unwrap_or(0); let mut fragmented = false; @@ -196,18 +197,18 @@ where self.dirty = true; - let encoded_bytes_len = SizeType::try_from(encoded_bytes.len()).map_err(|err| Error::new(ErrorKind::Other, err))?; + let encoded_bytes_len = SizeType::try_from(encoded_bytes.len()).map_err(to_io_error)?; self.main_file.write_all(&encoded_bytes_len.to_le_bytes())?; match file_utils::check_write(&self.main_file.write_all(&encoded_bytes)) { Ok(()) => { if new_record_appended { self.index_tree.insert(doc_id, self.main_offset); - let written_bytes = SizeType::try_from(encoded_bytes.len() + LEN_SIZE).map_err(|err| Error::new(ErrorKind::Other, err))?; + let written_bytes = SizeType::try_from(encoded_bytes.len() + LEN_SIZE).map_err(to_io_error)?; self.main_offset += written_bytes; } } Err(err) => { - return Err(Error::new(ErrorKind::Other, format!("failed to write document: {} - {}", self.main_path.to_str().unwrap(), err))); + return Err(str_to_io_error(&format!("failed to write document: {} - {}", self.main_path.to_str().unwrap(), err))); } } Ok(()) @@ -251,7 +252,7 @@ where let index_tree = IndexedDocumentIndex::::load(index_path)?; Ok(Self { - main_file: BufReader::new(main_file), + main_file: file_reader(main_file), index_tree, t_type: PhantomData, }) @@ -307,7 +308,7 @@ where Ok(file) => { Ok(Self { main_path: main_path.to_path_buf(), - main_file: BufReader::new(file), + main_file: file_reader(file), offsets, index: 0, failed: false, @@ -354,7 +355,7 @@ where Ok(value) => Ok(Some(value)), Err(err) => { self.failed = true; - Err(Error::new(ErrorKind::Other, format!("Failed to deserialize document {err}"))) + Err(str_to_io_error(&format!("Failed to deserialize document {err}"))) } } } @@ -402,7 +403,7 @@ impl IndexedDocumentDirectAccess { return Ok(item); } } - Err(Error::new(ErrorKind::Other, format!("Failed to read item for id {:?} - {}", doc_id, main_path.to_str().unwrap()))) + Err(str_to_io_error(&format!("Failed to read item for id {:?} - {}", doc_id, main_path.to_str().unwrap()))) } } @@ -432,7 +433,7 @@ where // Retrieve file size and convert to `u32` for `main_file`, if possible let size = main_file .metadata() - .and_then(|meta| SizeType::try_from(meta.len()).map_err(|err| Error::new(ErrorKind::Other, err))) + .and_then(|meta| SizeType::try_from(meta.len()).map_err(to_io_error)) .unwrap_or(0); if size < 1 { return Err(Error::new(ErrorKind::UnexpectedEof, format!("File empty main:{main_path:?}"))); @@ -465,7 +466,7 @@ where self.index_tree.traverse(|keys, values| { keys.iter().zip(values.iter()).for_each(|(key, &offset)| key_offset.push((key.clone(), offset))); }); - let mut gc_writer = BufWriter::new(&gc_file); + let mut gc_writer = file_writer(&gc_file); let fragmented_byte = 0u8.to_le_bytes(); gc_writer.write_all(&fragmented_byte)?; @@ -488,7 +489,7 @@ where gc_writer.write_all(&size_bytes)?; gc_writer.write_all(&buffer[0..buf_size])?; - let pointer = OffsetPointer::try_from(gc_offset).map_err(|err| Error::new(ErrorKind::Other, err))?; + let pointer = OffsetPointer::try_from(gc_offset).map_err(to_io_error)?; self.index_tree.insert(key, pointer); gc_offset += size_bytes.len() + buf_size; // gc_file.stream_position(); } diff --git a/src/repository/m3u_repository.rs b/src/repository/m3u_repository.rs index 01c927a77..5beb186f8 100644 --- a/src/repository/m3u_repository.rs +++ b/src/repository/m3u_repository.rs @@ -1,10 +1,10 @@ use std::fs::File; -use std::io::{BufWriter, Error, ErrorKind, Write}; +use std::io::{Error, Write}; use std::path::{Path, PathBuf}; use log::error; use crate::{create_m3u_filter_error, info_err}; -use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; +use crate::m3u_filter_error::{str_to_io_error, M3uFilterError, M3uFilterErrorKind}; use crate::model::api_proxy::{ProxyUserCredentials}; use crate::model::config::{Config, ConfigTarget}; use crate::model::playlist::{M3uPlaylistItem, PlaylistGroup, PlaylistItem, PlaylistItemType}; @@ -12,6 +12,7 @@ use crate::repository::indexed_document::{IndexedDocumentDirectAccess, IndexedDo use crate::repository::m3u_playlist_iterator::M3uPlaylistIterator; use crate::repository::storage::{FILE_SUFFIX_DB, FILE_SUFFIX_INDEX}; use crate::utils::file_utils; +use crate::utils::file_utils::file_writer; const FILE_M3U: &str = "m3u"; macro_rules! cant_write_result { @@ -36,7 +37,7 @@ fn persist_m3u_playlist_as_text(target: &ConfigTarget, cfg: &Config, m3u_playlis if let Some(m3u_filename) = file_utils::get_file_path(&cfg.working_dir, Some(PathBuf::from(filename))) { match File::create(&m3u_filename) { Ok(file) => { - let mut buf_writer = BufWriter::new(file); + let mut buf_writer = file_writer(&file); let _ = buf_writer.write(b"#EXTM3U\n"); for m3u in m3u_playlist { let _ = buf_writer.write(m3u.to_m3u(target.options.as_ref(), None).as_bytes()); @@ -90,7 +91,7 @@ pub async fn m3u_load_rewrite_playlist( pub async fn m3u_get_item_for_stream_id(cfg: &Config, stream_id: u32, m3u_path: &Path, idx_path: &Path) -> Result { if stream_id < 1 { - return Err(Error::new(ErrorKind::Other, "id should start with 1")); + return Err(str_to_io_error("id should start with 1")); } { let _file_lock = cfg.file_locks.read_lock(m3u_path).await?; diff --git a/src/repository/xtream_repository.rs b/src/repository/xtream_repository.rs index 395dfc200..ce63900f2 100644 --- a/src/repository/xtream_repository.rs +++ b/src/repository/xtream_repository.rs @@ -1,3 +1,5 @@ +use crate::m3u_filter_error::str_to_io_error; +use crate::file_utils::file_reader; use std::collections::HashMap; use std::fs; use std::fs::File; @@ -161,7 +163,7 @@ fn load_old_category_ids(path: &Path) -> (u32, HashMap) { let col_path = get_collection_path(path, cat); if col_path.exists() { if let Ok(file) = File::open(col_path) { - let reader = BufReader::new(file); + let reader = file_reader(file); for entry in json_iter_array::>(reader).flatten() { if let Some(category_id) = entry.get(TAG_CATEGORY_ID).and_then(get_u32_from_serde_value) { if let Value::Object(item) = entry { @@ -335,10 +337,7 @@ pub fn xtream_get_collection_path( return Ok((Some(col_path), None)); } } - Err(Error::new( - ErrorKind::Other, - format!("Cant find collection: {target_name}/{collection_name}"), - )) + Err(str_to_io_error(&format!("Cant find collection: {target_name}/{collection_name}"))) } async fn xtream_read_item_for_stream_id( @@ -370,7 +369,7 @@ macro_rules! try_cluster { ($xtream_cluster:expr, $item_type:expr, $virtual_id:expr) => { $xtream_cluster .or_else(|| XtreamCluster::try_from($item_type).ok()) - .ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not determine cluster for xtream item with stream-id {}",$virtual_id))) + .ok_or_else(|| str_to_io_error(&format!("Could not determine cluster for xtream item with stream-id {}",$virtual_id))) }; } @@ -380,14 +379,14 @@ pub async fn xtream_get_item_for_stream_id( target: &ConfigTarget, xtream_cluster: Option, ) -> Result { - let target_path = get_target_storage_path(config, target.name.as_str()).ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not find path for target {}", &target.name)))?; - let storage_path = xtream_get_storage_path(config, target.name.as_str()).ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not find path for target {} xtream output", &target.name)))?; + let target_path = get_target_storage_path(config, target.name.as_str()).ok_or_else(|| str_to_io_error(&format!("Could not find path for target {}", &target.name)))?; + let storage_path = xtream_get_storage_path(config, target.name.as_str()).ok_or_else(|| str_to_io_error(&format!("Could not find path for target {} xtream output", &target.name)))?; { let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = config.file_locks.read_lock(&target_id_mapping_file).await.map_err(|err| Error::new(ErrorKind::Other, format!("Could not get lock for id mapping for target {} err:{err}", target.name)))?; + let _file_lock = config.file_locks.read_lock(&target_id_mapping_file).await.map_err(|err|str_to_io_error(&format!("Could not get lock for id mapping for target {} err:{err}", target.name)))?; - let mut target_id_mapping = BPlusTreeQuery::::try_new(&target_id_mapping_file).map_err(|err| Error::new(ErrorKind::Other, format!("Could not load id mapping for target {} err:{err}", target.name)))?; - let mapping = target_id_mapping.query(&virtual_id).ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not find mapping for target {} and id {}", target.name, virtual_id)))?; + let mut target_id_mapping = BPlusTreeQuery::::try_new(&target_id_mapping_file).map_err(|err| str_to_io_error(&format!("Could not load id mapping for target {} err:{err}", target.name)))?; + let mapping = target_id_mapping.query(&virtual_id).ok_or_else(|| str_to_io_error(&format!("Could not find mapping for target {} and id {}", target.name, virtual_id)))?; match mapping.item_type { PlaylistItemType::SeriesInfo => { xtream_read_series_item_for_stream_id(config, virtual_id, &storage_path).await @@ -440,7 +439,7 @@ pub async fn xtream_write_series_info( { let _file_lock = config.file_locks.write_lock(&info_path).await?; let mut writer = IndexedDocumentWriter::new_append(info_path, idx_path)?; - writer.write_doc(series_info_id, content).map_err(|_| Error::new(ErrorKind::Other, format!("failed to write xtream series info for target {target_name}")))?; + writer.write_doc(series_info_id, content).map_err(|_| str_to_io_error(&format!("failed to write xtream series info for target {target_name}")))?; writer.store()?; } { @@ -468,7 +467,7 @@ pub async fn xtream_write_vod_info( { let _file_lock = config.file_locks.write_lock(&info_path).await?; let mut writer = IndexedDocumentWriter::new_append(info_path, idx_path)?; - writer.write_doc(virtual_id, content).map_err(|_| Error::new(ErrorKind::Other, format!("failed to write xtream vod info for target {target_name}")))?; + writer.write_doc(virtual_id, content).map_err(|_| str_to_io_error(&format!("failed to write xtream vod info for target {target_name}")))?; writer.store()?; } Ok(()) @@ -488,11 +487,11 @@ async fn xtream_get_info_mapping(config: &Config, target_name: &str, info_id: u3 let target_id_mapping_file = get_target_id_mapping_file(&target_path); let _file_lock = config.file_locks.read_lock(&target_id_mapping_file).await.map_err(|err| { error!("Could not lock id mapping for target {target_name}: {}", err); - Error::new(ErrorKind::Other, format!("ID mapping load error for target {target_name}")) + str_to_io_error(&format!("ID mapping load error for target {target_name}")) }).ok()?; BPlusTreeQuery::::try_new(&target_id_mapping_file).map_err(|err| { error!("Could not load id mapping for target {target_name}: {}", err); - Error::new(ErrorKind::Other, format!("ID mapping load error for target {target_name}")) + str_to_io_error(&format!("ID mapping load error for target {target_name}")) }).ok().map(|mut tree| tree.query(&info_id))? } @@ -512,7 +511,7 @@ pub async fn xtream_load_series_info( { let _file_lock = config.file_locks.read_lock(&info_path).await.map_err(|err| { error!("Could not lock document {:?}: {}", info_path, err); - Error::new(ErrorKind::Other, format!("Document Reader error for target {target_name}")) + str_to_io_error(&format!("Document Reader error for target {target_name}")) }).ok()?; return match IndexedDocumentDirectAccess::read_indexed_item::(&info_path, &idx_path, &series_id) { Ok(content) => Some(content), @@ -553,7 +552,7 @@ pub async fn xtream_load_vod_info( { let _file_lock = config.file_locks.read_lock(&info_path).await.map_err(|err| { error!("Could not lock document {:?}: {}", info_path, err); - Error::new(ErrorKind::Other, format!("Document Reader error for target {target_name}")) + str_to_io_error(&format!("Document Reader error for target {target_name}")) }).ok()?; return match IndexedDocumentDirectAccess::read_indexed_item::(&info_path, &idx_path, &vod_id) { Ok(content) => Some(content), @@ -608,7 +607,7 @@ fn rewrite_xtream_vod_info

( ); } } - let result = serde_json::to_string(&doc).map_err(|_| Error::new(ErrorKind::Other, "Failed to serialize vod info"))?; + let result = serde_json::to_string(&doc).map_err(|_| str_to_io_error("Failed to serialize vod info"))?; Ok(result) } @@ -622,7 +621,7 @@ pub fn rewrite_xtream_vod_info_content

( ) -> Result where P: PlaylistEntry, { - let mut doc = serde_json::from_str::>(content).map_err(|_| Error::new(ErrorKind::Other, "Failed to parse JSON content"))?; + let mut doc = serde_json::from_str::>(content).map_err(|_| str_to_io_error("Failed to parse JSON content"))?; rewrite_xtream_vod_info(config, target, pli, user, &mut doc) } @@ -635,7 +634,7 @@ pub async fn write_and_get_xtream_vod_info

( ) -> Result where P: PlaylistEntry, { - let mut doc = serde_json::from_str::>(content).map_err(|_| Error::new(ErrorKind::Other, "Failed to parse JSON content"))?; + let mut doc = serde_json::from_str::>(content).map_err(|_| str_to_io_error("Failed to parse JSON content"))?; xtream_write_vod_info(config, target.name.as_str(), pli.get_virtual_id(), content).await.ok(); rewrite_xtream_vod_info(config, target, pli, user, &mut doc) } @@ -649,7 +648,7 @@ async fn rewrite_xtream_series_info

( ) -> Result where P: PlaylistEntry, { - let target_path = get_target_storage_path(config, target.name.as_str()).ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not find path for target {}", target.name)))?; + let target_path = get_target_storage_path(config, target.name.as_str()).ok_or_else(|| str_to_io_error(&format!("Could not find path for target {}", target.name)))?; let resource_url = match user.proxy { ProxyType::Reverse => { @@ -677,12 +676,12 @@ async fn rewrite_xtream_series_info

( } } - let episodes = doc.get_mut(TAG_EPISODES).and_then(Value::as_object_mut).ok_or_else(|| Error::new(ErrorKind::Other, "No episodes found in content"))?; + let episodes = doc.get_mut(TAG_EPISODES).and_then(Value::as_object_mut).ok_or_else(|| str_to_io_error("No episodes found in content"))?; let virtual_id = pli.get_virtual_id(); { let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = config.file_locks.write_lock(&target_id_mapping_file).await.map_err(|err| Error::new(ErrorKind::Other, format!("Could not load id mapping for target {} err:{err}", target.name)))?; + let _file_lock = config.file_locks.write_lock(&target_id_mapping_file).await.map_err(|err| str_to_io_error(&format!("Could not load id mapping for target {} err:{err}", target.name)))?; let mut target_id_mapping = TargetIdMapping::new(&target_id_mapping_file); let options = XtreamMappingOptions::from_target_options(target.options.as_ref()); @@ -715,7 +714,7 @@ async fn rewrite_xtream_series_info

( drop(target_id_mapping); } - let result = serde_json::to_string(&doc).map_err(|_| Error::new(ErrorKind::Other, "Failed to serialize updated series info"))?; + let result = serde_json::to_string(&doc).map_err(|_| str_to_io_error("Failed to serialize updated series info"))?; Ok(result) } @@ -729,7 +728,7 @@ pub async fn rewrite_xtream_series_info_content

( ) -> Result where P: PlaylistEntry, { - let mut doc = serde_json::from_str::>(content).map_err(|_| Error::new(ErrorKind::Other, "Failed to parse JSON content"))?; + let mut doc = serde_json::from_str::>(content).map_err(|_| str_to_io_error("Failed to parse JSON content"))?; rewrite_xtream_series_info(config, target, pli_series_info, user, &mut doc).await } @@ -742,7 +741,7 @@ pub async fn write_and_get_xtream_series_info

( ) -> Result where P: PlaylistEntry, { - let mut doc = serde_json::from_str::>(content).map_err(|_| Error::new(ErrorKind::Other, "Failed to parse JSON content"))?; + let mut doc = serde_json::from_str::>(content).map_err(|_| str_to_io_error("Failed to parse JSON content"))?; let virtual_id = pli_series_info.get_virtual_id(); xtream_write_series_info(config, target.name.as_str(), virtual_id, content).await.ok(); rewrite_xtream_series_info(config, target, pli_series_info, user, &mut doc).await @@ -775,7 +774,7 @@ pub async fn xtream_update_input_info_file( Ok(Some((info_path, idx_path))) => { match cfg.file_locks.write_lock(&info_path).await { Ok(_file_lock) => { - let mut reader = BufReader::new(open_readonly_file(wal_path).map_err(|err| notify_err!(format!("Could not read {cluster} info {err}")))?); + let mut reader = file_reader(open_readonly_file(wal_path).map_err(|err| notify_err!(format!("Could not read {cluster} info {err}")))?); match IndexedDocumentWriter::::new_append(info_path, idx_path) { Ok(mut writer) => { let mut provider_id_bytes = [0u8; 4]; @@ -822,7 +821,7 @@ pub async fn xtream_update_input_vod_record_from_wal_file( match cfg.file_locks.write_lock(&record_path).await { Ok(_file_lock) => { - let mut reader = BufReader::new(open_readonly_file(wal_path).map_err(|err| notify_err!(format!("Could not read vod wal info {err}")))?); + let mut reader = file_reader(open_readonly_file(wal_path).map_err(|err| notify_err!(format!("Could not read vod wal info {err}")))?); let mut provider_id_bytes = [0u8; 4]; let mut tmdb_id_bytes = [0u8; 4]; let mut ts_bytes = [0u8; 8]; @@ -863,7 +862,7 @@ pub async fn xtream_update_input_series_record_from_wal_file( .and_then(|opt| opt.ok_or_else(|| notify_err!(format!("Error accessing storage path for input: {}", input.name.clone().unwrap_or_else(|| input.id.to_string())))))?; match cfg.file_locks.write_lock(&record_path).await { Ok(_file_lock) => { - let mut reader = BufReader::new(open_readonly_file(wal_path).map_err(|err| notify_err!(format!("Could not read series wal info {err}")))?); + let mut reader = file_reader(open_readonly_file(wal_path).map_err(|err| notify_err!(format!("Could not read series wal info {err}")))?); let mut provider_id_bytes = [0u8; 4]; let mut ts_bytes = [0u8; 8]; let mut tree_record_index: BPlusTree = BPlusTree::load(&record_path).unwrap_or_else(|_| BPlusTree::new()); @@ -900,7 +899,7 @@ pub async fn xtream_update_input_series_episodes_record_from_wal_file( .and_then(|opt| opt.ok_or_else(|| notify_err!(format!("Error accessing storage path for input: {}", input.name.clone().unwrap_or_else(|| input.id.to_string())))))?; match cfg.file_locks.write_lock(&record_path).await { Ok(_file_lock) => { - let mut reader = BufReader::new(open_readonly_file(wal_path).map_err(|err| notify_err!(format!("Could not read series episode wal info {err}")))?); + let mut reader = file_reader(open_readonly_file(wal_path).map_err(|err| notify_err!(format!("Could not read series episode wal info {err}")))?); let mut provider_id_bytes = [0u8; 4]; let mut len_bytes = [0u8; 4]; let mut tree_record_index: BPlusTree = BPlusTree::load(&record_path).unwrap_or_else(|_| BPlusTree::new()); diff --git a/src/utils/compressed_file_reader.rs b/src/utils/compressed_file_reader.rs index 662620d96..2b71a5957 100644 --- a/src/utils/compressed_file_reader.rs +++ b/src/utils/compressed_file_reader.rs @@ -2,7 +2,7 @@ use std::io::{BufRead, BufReader, Read, Seek, SeekFrom}; use std::path::Path; use flate2::bufread::{GzDecoder, ZlibDecoder}; use crate::utils::compression_utils::{is_deflate, is_gzip}; -use crate::utils::file_utils::open_readonly_file; +use crate::utils::file_utils::{file_reader, open_readonly_file}; pub struct CompressedFileReader { reader: BufReader>, @@ -12,20 +12,22 @@ impl CompressedFileReader { pub fn new(path: &Path) -> std::io::Result { let file = open_readonly_file(path)?; - let mut buffered_file = BufReader::new(file); + let mut buffered_file = file_reader(file); let mut header = [0u8; 2]; buffered_file.read_exact(&mut header)?; buffered_file.seek(SeekFrom::Start(0))?; - let reader: BufReader> = if is_gzip(&header) { - BufReader::new(Box::new(GzDecoder::new(buffered_file)) as Box) + let reader: Box = if is_gzip(&header) { + Box::new(GzDecoder::new(buffered_file)) } else if is_deflate(&header) { - BufReader::new(Box::new(ZlibDecoder::new(buffered_file)) as Box) + Box::new(ZlibDecoder::new(buffered_file)) } else { - BufReader::new(Box::new(buffered_file) as Box) + Box::new(buffered_file) }; - Ok(Self { reader }) + Ok(Self { + reader: BufReader::new(reader), + }) } } diff --git a/src/utils/config_reader.rs b/src/utils/config_reader.rs index 0de249085..e2c1ad2dd 100644 --- a/src/utils/config_reader.rs +++ b/src/utils/config_reader.rs @@ -6,9 +6,8 @@ use chrono::Local; use log::{debug, error, info, warn}; use regex::Regex; use serde::Serialize; - -use crate::{create_m3u_filter_error_result, handle_m3u_filter_error_result, info_err}; -use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; +use crate::{create_m3u_filter_error, create_m3u_filter_error_result, handle_m3u_filter_error_result, info_err}; +use crate::m3u_filter_error::{to_io_error, M3uFilterError, M3uFilterErrorKind}; use crate::model::api_proxy::ApiProxyConfig; use crate::model::config::{Config, ConfigDto}; use crate::model::mapping::Mappings; @@ -127,13 +126,12 @@ fn write_config_file(file_path: &str, backup_dir: &str, config: &T, default_n Err(err) => { error!("Could not backup file {}:{}", &backup_path.to_str().unwrap_or("?"), err) } } info!("Saving file to {}", &path.to_str().unwrap_or("?")); - match File::create(&path) { - Ok(f) => { - serde_yaml::to_writer(f, &config).unwrap(); - Ok(()) - } - Err(err) => create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "Could not write file {}: {}", &path.to_str().unwrap_or("?"), err) - } + + File::create(&path) + .and_then(|f| serde_yaml::to_writer(f, &config).map_err(to_io_error)) + .map_err(|err| create_m3u_filter_error!(M3uFilterErrorKind::Info, "Could not write file {}: {}", &path.to_str().unwrap_or("?"), err)) + + } pub fn save_api_proxy(file_path: &str, backup_dir: &str, config: &ApiProxyConfig) -> Result<(), M3uFilterError> { diff --git a/src/utils/download.rs b/src/utils/download.rs index 3d56469a6..de215c532 100644 --- a/src/utils/download.rs +++ b/src/utils/download.rs @@ -1,6 +1,6 @@ use crate::Arc; use std::borrow::Cow; -use crate::m3u_filter_error::M3uFilterError; +use crate::m3u_filter_error::{str_to_io_error, M3uFilterError}; use crate::model::config::{Config, ConfigInput, ConfigTarget}; use crate::model::playlist::{PlaylistEntry, PlaylistGroup, XtreamCluster}; use crate::model::xmltv::TVGuide; @@ -10,7 +10,7 @@ use crate::repository::xtream_repository; use crate::utils::{file_utils, request_utils}; use log::{debug, info}; use std::cmp::Ordering; -use std::io::{Error, ErrorKind}; +use std::io::{Error}; use std::path::PathBuf; use crate::debug_if_enabled; use crate::model::api_proxy::{ProxyUserCredentials}; @@ -120,7 +120,7 @@ where }; } - Err(Error::new(ErrorKind::Other, format!("Cant find stream with id: {}/{}/{}", + Err(str_to_io_error(&format!("Cant find stream with id: {}/{}/{}", target.name.replace(' ', "_").as_str(), &cluster, pli.get_virtual_id()))) } diff --git a/src/utils/file_lock_manager.rs b/src/utils/file_lock_manager.rs index b9a609933..7968836c4 100644 --- a/src/utils/file_lock_manager.rs +++ b/src/utils/file_lock_manager.rs @@ -3,6 +3,7 @@ use std::sync::{Arc}; use std::{fmt, io}; use std::path::{Path, PathBuf}; use async_std::sync::{Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard}; +use crate::m3u_filter_error::str_to_io_error; #[derive(Clone)] pub struct FileLockManager { @@ -18,9 +19,7 @@ impl FileLockManager { // Acquires a read lock for the specified file and returns a FileReadGuard. pub async fn read_lock(&self, path: &Path) -> io::Result { - let file_lock = self.get_or_create_lock(path).await.map_err(|_| { - io::Error::new(io::ErrorKind::Other, "Failed to acquire write lock") - })?; + let file_lock = self.get_or_create_lock(path).await.map_err(|_| str_to_io_error("Failed to acquire write lock"))?; let guard = file_lock.read().await; // Clone the Arc to avoid moving `file_lock` out, as it is still borrowed by `guard` Ok(FileReadGuard::new(Arc::clone(&file_lock), guard)) @@ -28,9 +27,7 @@ impl FileLockManager { // Acquires a write lock for the specified file and returns a FileWriteGuard. pub async fn write_lock(&self, path: &Path) -> io::Result { - let file_lock = self.get_or_create_lock(path).await.map_err(|_| { - io::Error::new(io::ErrorKind::Other, "Failed to acquire write lock") - })?; + let file_lock = self.get_or_create_lock(path).await.map_err(|_| str_to_io_error("Failed to acquire write lock"))?; let guard = file_lock.write().await; // Clone the Arc to avoid moving `file_lock` out, as it is still borrowed by `guard` Ok(FileWriteGuard::new(Arc::clone(&file_lock), guard)) @@ -38,14 +35,12 @@ impl FileLockManager { // Tries to acquire a write lock for the specified file and returns a FileWriteGuard. pub async fn try_write_lock(&self, path: &Path) -> io::Result { - let file_lock = self.get_or_create_lock(path).await.map_err(|_| { - io::Error::new(io::ErrorKind::Other, "Failed to acquire write lock") - })?; + let file_lock = self.get_or_create_lock(path).await.map_err(|_|str_to_io_error("Failed to acquire write lock"))?; let guard = file_lock.try_write(); match guard { // Clone the Arc to avoid moving `file_lock` out, as it is still borrowed by `guard` Some(lock_guard) => Ok(FileWriteGuard::new(Arc::clone(&file_lock), lock_guard)), - None => Err(io::Error::new(io::ErrorKind::Other, "Failed to acquire write lock")) + None => Err(str_to_io_error("Failed to acquire write lock")) } } diff --git a/src/utils/file_utils.rs b/src/utils/file_utils.rs index 60dfbdd3a..99b74486e 100644 --- a/src/utils/file_utils.rs +++ b/src/utils/file_utils.rs @@ -1,10 +1,11 @@ use std::fs; use std::fs::{File, OpenOptions}; -use std::io::Write; +use std::io::{BufReader, BufWriter, Read, Write}; use std::path::{Path, PathBuf}; use log::{debug, error}; use path_clean::PathClean; +use crate::m3u_filter_error::str_to_io_error; const USER_FILE: &str = "user.txt"; const CONFIG_PATH: &str = "config"; @@ -21,6 +22,18 @@ macro_rules! exit { }}; } +pub fn file_writer(w: W) -> BufWriter +where W: Write +{ + BufWriter::with_capacity(131_072, w) +} + +pub fn file_reader(r: R) -> BufReader +where R: Read +{ + BufReader::with_capacity(131_072, r) +} + pub fn get_exe_path() -> PathBuf { let default_path = std::path::PathBuf::from("./"); let current_exe = std::env::current_exe(); @@ -159,7 +172,7 @@ pub fn path_exists(file_path: &Path) -> bool { pub fn check_write(res: &std::io::Result<()>) -> Result<(), std::io::Error> { match res { Ok(()) => Ok(()), - Err(_) => Err(std::io::Error::new(std::io::ErrorKind::Other, "Unable to write file")), + Err(_) => Err(str_to_io_error("Unable to write file")), } } diff --git a/src/utils/json_utils.rs b/src/utils/json_utils.rs index cfdbcaa38..c8e26e2f3 100644 --- a/src/utils/json_utils.rs +++ b/src/utils/json_utils.rs @@ -1,11 +1,12 @@ use std::collections::HashMap; use std::fs::File; -use std::io::{self, BufReader, BufWriter, Error, Read, Write}; +use std::io::{self, BufReader, Error, Read, Write}; use std::path::Path; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use serde_json::{self, Deserializer, Value}; +use crate::utils::file_utils::{file_reader, file_writer}; fn read_skipping_ws(mut reader: impl Read) -> io::Result { loop { @@ -73,7 +74,7 @@ pub fn json_filter_file(file_path: &Path, filter: &HashMap<&str, &str>) -> Vec>(reader).flatten() { if let Some(item) = entry.as_object() { if filter.iter().all(|(&key, &value)| { @@ -95,17 +96,10 @@ pub fn json_write_documents_to_file(file: &Path, value: &T) -> Result<(), Err where T: ?Sized + Serialize, { - match File::create(file) { - Ok(file) => { - let mut writer = BufWriter::new(file); - serde_json::to_writer(&mut writer, value)?; - match writer.flush() { - Ok(()) => Ok(()), - Err(e) => Err(e), - } - } - Err(e) => Err(e), - } + let file = File::create(file)?; + let mut writer = file_writer(&file); + serde_json::to_writer(&mut writer, value)?; + writer.flush() } pub fn string_or_number_u32<'de, D>(deserializer: D) -> Result diff --git a/src/utils/multi_file_reader.rs b/src/utils/multi_file_reader.rs index a35538d3f..d595fe740 100644 --- a/src/utils/multi_file_reader.rs +++ b/src/utils/multi_file_reader.rs @@ -1,6 +1,7 @@ use std::fs::File; use std::io::{self, BufReader, ErrorKind, Read}; use std::path::{PathBuf}; +use crate::utils::file_utils::file_reader; pub struct MultiFileReader { files: Vec, @@ -33,7 +34,7 @@ impl Read for MultiFileReader { if self.files.is_empty() { return Ok(0); } - self.current_reader = Some(BufReader::new(self.files.pop().unwrap())); + self.current_reader = Some(file_reader(self.files.pop().unwrap())); // we put a newline if the config does not have one if !buf.is_empty() && buf[0] != b'\n' { buf[0] = b'\n'; diff --git a/src/utils/request_utils.rs b/src/utils/request_utils.rs index 3bef6a22d..5cd582786 100644 --- a/src/utils/request_utils.rs +++ b/src/utils/request_utils.rs @@ -15,7 +15,7 @@ use reqwest::header::CONTENT_ENCODING; use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; use url::Url; -use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; +use crate::m3u_filter_error::{str_to_io_error, M3uFilterError, M3uFilterErrorKind}; use crate::model::config::ConfigInput; use crate::model::stats::format_elapsed_time; use crate::repository::storage::get_input_storage_path; @@ -169,7 +169,7 @@ fn get_local_file_content(file_path: &PathBuf) -> Result { let mut decode_buffer = String::new(); match decoder.read_to_string(&mut decode_buffer) { Ok(_) => return Ok(decode_buffer), - Err(err) => return Err(Error::new(ErrorKind::Other, format!("failed to decode gzip content {err}"))) + Err(err) => return Err(str_to_io_error(&format!("failed to decode gzip content {err}"))) }; } return Ok(String::from_utf8_lossy(&content).parse().unwrap()); @@ -196,7 +196,7 @@ async fn get_remote_content_as_file(client: Arc, input: &Config file.write_all(&bytes)?; } Err(err) => { - return Err(std::io::Error::new(std::io::ErrorKind::Other, format!("Failed to read chunk: {err}"))); + return Err(str_to_io_error(&format!("Failed to read chunk: {err}"))); } } } @@ -206,10 +206,10 @@ async fn get_remote_content_as_file(client: Arc, input: &Config debug!("File downloaded successfully to {file_path:?}, took:{}", format_elapsed_time(elapsed)); Ok(file_path.to_path_buf()) } else { - Err(std::io::Error::new(std::io::ErrorKind::Other, format!("Request failed with status {} {}", response.status(), mask_sensitive_info(url.as_str())))) + Err(str_to_io_error(&format!("Request failed with status {} {}", response.status(), mask_sensitive_info(url.as_str())))) } } - Err(err) => Err(std::io::Error::new(std::io::ErrorKind::Other, format!("Request failed: {} {err}", mask_sensitive_info(url.as_str())))), + Err(err) => Err(str_to_io_error(&format!("Request failed: {} {err}", mask_sensitive_info(url.as_str())))), } } @@ -239,14 +239,14 @@ async fn get_remote_content(client: Arc, input: &ConfigInput, u let mut decoder = GzDecoder::new(&bytes[..]); match decoder.read_to_string(&mut decode_buffer) { Ok(_) => {} - Err(err) => return Err(std::io::Error::new(ErrorKind::Other, format!("failed to decode gzip content {err}"))) + Err(err) => return Err(str_to_io_error(&format!("failed to decode gzip content {err}"))) }; } ENCODING_DEFLATE => { let mut decoder = ZlibDecoder::new(&bytes[..]); match decoder.read_to_string(&mut decode_buffer) { Ok(_) => {} - Err(err) => return Err(std::io::Error::new(ErrorKind::Other, format!("failed to decode zlib content {err}"))) + Err(err) => return Err(str_to_io_error(&format!("failed to decode zlib content {err}"))) } } _ => {} @@ -259,20 +259,20 @@ async fn get_remote_content(client: Arc, input: &ConfigInput, u debug_if_enabled!("Request took:{} {}", format_elapsed_time(start_time.elapsed().as_secs()), mask_sensitive_info(url.as_str())); Ok(decoded_content) } - Err(err) => Err(std::io::Error::new(ErrorKind::Other, format!("failed to plain text content {err}"))) + Err(err) => Err(str_to_io_error(&format!("failed to plain text content {err}"))) } } else { debug_if_enabled!("Request took:{}, {}", format_elapsed_time(start_time.elapsed().as_secs()), mask_sensitive_info(url.as_str())); Ok(decode_buffer) } } - Err(err) => Err(std::io::Error::new(ErrorKind::Other, format!("failed to read response {} {err}", mask_sensitive_info(url.as_str())))) + Err(err) => Err(str_to_io_error(&format!("failed to read response {} {err}", mask_sensitive_info(url.as_str())))) } } else { - Err(std::io::Error::new(ErrorKind::Other, format!("Request failed with status {} {}", response.status(), mask_sensitive_info(url.as_str())))) + Err(str_to_io_error(&format!("Request failed with status {} {}", response.status(), mask_sensitive_info(url.as_str())))) } } - Err(err) => Err(std::io::Error::new(ErrorKind::Other, format!("Request failed {} {err}", mask_sensitive_info(url.as_str())))) + Err(err) => Err(str_to_io_error(&format!("Request failed {} {err}", mask_sensitive_info(url.as_str())))) } } @@ -305,7 +305,7 @@ pub async fn download_text_content_as_file(client: Arc, input: pub async fn download_text_content(client: Arc, input: &ConfigInput, url_str: &str, persist_filepath: Option) -> Result { if let Ok(url) = url_str.parse::() { let result = if url.scheme() == "file" { - url.to_file_path().map_or_else(|()| Err(Error::new(ErrorKind::Other, format!("Unknown file {}", mask_sensitive_info(url_str)))), |file_path| get_local_file_content(&file_path)) + url.to_file_path().map_or_else(|()| Err(str_to_io_error(&format!("Unknown file {}", mask_sensitive_info(url_str)))), |file_path| get_local_file_content(&file_path)) } else { get_remote_content(client, input, &url).await }; @@ -319,7 +319,7 @@ pub async fn download_text_content(client: Arc, input: &ConfigI Err(err) => Err(err) } } else { - Err(std::io::Error::new(ErrorKind::Other, format!("Malformed URL {}", mask_sensitive_info(url_str)))) + Err(str_to_io_error(&format!("Malformed URL {}", mask_sensitive_info(url_str)))) } } @@ -329,7 +329,7 @@ async fn download_json_content(client: Arc, input: &ConfigInput Ok(content) => { match serde_json::from_str::(&content) { Ok(value) => Ok(value), - Err(err) => Err(Error::new(ErrorKind::Other, format!("Failed to parse json {err}"))) + Err(err) => Err(str_to_io_error(&format!("Failed to parse json {err}"))) } } Err(err) => Err(err)