diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs index 5f603493b..cc24bdcec 100644 --- a/src/api/api_utils.rs +++ b/src/api/api_utils.rs @@ -126,27 +126,26 @@ pub async fn get_user_target<'a>(api_req: &'a UserApiRequest, app_state: &'a App pub struct StreamOptions { pub stream_retry: bool, pub stream_force_retry_secs: u32, - pub stream_connect_timeout_secs: u32, pub buffer_enabled: bool, pub buffer_size: usize, pub pipe_provider_stream: bool, } fn get_stream_options(app_state: &AppState) -> StreamOptions { - let (stream_retry, stream_force_retry_secs, stream_connect_timeout_secs, buffer_enabled, buffer_size) = app_state + let (stream_retry, stream_force_retry_secs, buffer_enabled, buffer_size) = app_state .config .reverse_proxy .as_ref() .and_then(|reverse_proxy| reverse_proxy.stream.as_ref()) - .map_or((false, 0, 0, false, 0), |stream| { + .map_or((false, 0, false, 0), |stream| { let (buffer_enabled, buffer_size) = stream .buffer .as_ref() .map_or((false, 0), |buffer| (buffer.enabled, buffer.size)); - (stream.retry, stream.forced_retry_interval_secs, stream.connect_timeout_secs, buffer_enabled, buffer_size) + (stream.retry, stream.forced_retry_interval_secs, buffer_enabled, buffer_size) }); let pipe_provider_stream = !stream_retry && !buffer_enabled; - StreamOptions { stream_retry, stream_force_retry_secs, stream_connect_timeout_secs, buffer_enabled, buffer_size, pipe_provider_stream } + StreamOptions { stream_retry, stream_force_retry_secs, buffer_enabled, buffer_size, pipe_provider_stream } } // fn get_stream_content_length(provider_response: Option<&(Vec<(String, String)>, StatusCode)>) -> u64 { @@ -256,7 +255,7 @@ async fn create_stream_response_details(app_state: &AppState, stream_options: &S let parsed_url = Url::parse(&request_url); let ((stream, stream_info), reconnect_flag) = if let Ok(url) = parsed_url { if stream_options.pipe_provider_stream { - (provider_stream::get_provider_pipe_stream(app_state, &url, req_headers, input_headers, item_type, &stream_options).await, None) + (provider_stream::get_provider_pipe_stream(app_state, &url, req_headers, input_headers, item_type).await, None) } else { let buffer_stream_options = BufferStreamOptions::new(item_type, share_stream, &stream_options); let reconnect_flag = buffer_stream_options.get_reconnect_flag_clone(); diff --git a/src/api/endpoints/v1_api.rs b/src/api/endpoints/v1_api.rs index a7589ebb7..6c1e0fb53 100644 --- a/src/api/endpoints/v1_api.rs +++ b/src/api/endpoints/v1_api.rs @@ -190,6 +190,7 @@ async fn get_playlist(client: Arc, cfg_input: Option<&ConfigInp match input.input_type { InputType::M3u => m3u::get_m3u_playlist(client, cfg, input, &cfg.working_dir).await, InputType::Xtream => xtream::get_xtream_playlist(client, input, &cfg.working_dir).await, + InputType::M3uBatch => (vec![], vec![]) }; if result.is_empty() { let error_strings: Vec = errors.iter().map(std::string::ToString::to_string).collect(); diff --git a/src/api/main_api.rs b/src/api/main_api.rs index 0ca1ee4ab..c463537e7 100644 --- a/src/api/main_api.rs +++ b/src/api/main_api.rs @@ -27,6 +27,8 @@ use std::path::PathBuf; use std::sync::Arc; use tokio::sync::Mutex; use std::future::IntoFuture; +use std::time::Duration; +use reqwest::Client; use crate::api::model::hls_cache::HlsCache; fn get_web_dir_path(web_ui_enabled: bool, web_root: &str) -> Result { @@ -120,9 +122,20 @@ fn create_shared_data(cfg: &Arc) -> AppState { let active_users = Arc::new(ActiveUserManager::new()); let active_provider = Arc::new(ActiveProviderManager::new(cfg)); + let client = match cfg.reverse_proxy.as_ref().and_then(|r| r.stream.as_ref()).map(|stream| stream.connect_timeout_secs) { + Some(timeout) => { + Client::builder() + .connect_timeout(Duration::from_secs(u64::from(timeout))) + .build().unwrap_or_else(|_| Client::new()) + } + None => { + Client::new() + } + }; + AppState { config: Arc::clone(cfg), - http_client: Arc::new(reqwest::Client::new()), + http_client: Arc::new(client), downloads: Arc::new(DownloadQueue::new()), cache, hls_cache: HlsCache::garbage_collected(), diff --git a/src/api/model/streams/provider_stream.rs b/src/api/model/streams/provider_stream.rs index 223aa5a7e..ee84320a5 100644 --- a/src/api/model/streams/provider_stream.rs +++ b/src/api/model/streams/provider_stream.rs @@ -1,5 +1,5 @@ use std::collections::HashMap; -use crate::api::api_utils::{get_headers_from_request, HeaderFilter, StreamOptions}; +use crate::api::api_utils::{get_headers_from_request, HeaderFilter}; use crate::api::model::model_utils::get_response_headers; use crate::api::model::stream_error::StreamError; use crate::api::model::streams::custom_video_stream::CustomVideoStream; @@ -12,7 +12,6 @@ use futures::TryStreamExt; use log::{debug, error}; use reqwest::StatusCode; use std::sync::Arc; -use std::time::Duration; use axum::http::HeaderMap; use axum::response::IntoResponse; use url::Url; @@ -78,8 +77,7 @@ pub async fn get_provider_pipe_stream(app_state: &AppState, stream_url: &Url, req_headers: &HeaderMap, input_headers: Option>, - item_type: PlaylistItemType, - stream_options: &StreamOptions) -> ProviderStreamResponse { + item_type: PlaylistItemType) -> ProviderStreamResponse { let filter_header = get_header_filter_for_item_type(item_type); let req_headers = get_headers_from_request(req_headers, &filter_header); debug_if_enabled!("Stream requested with headers: {:?}", req_headers.iter().map(|header| (header.0, String::from_utf8_lossy(header.1))).collect::>()); @@ -87,12 +85,7 @@ pub async fn get_provider_pipe_stream(app_state: &AppState, // The stream url, we need to clone it because of move to async block. // We merge configured input headers with the headers from the request. let headers = get_request_headers(input_headers.as_ref(), Some(&req_headers)); - let client_builder = app_state.http_client.get(stream_url.clone()).headers(headers.clone()); - let client = if stream_options.stream_connect_timeout_secs > 0 { - client_builder.timeout(Duration::from_secs(u64::from(stream_options.stream_connect_timeout_secs))) - } else { - client_builder - }; + let client = app_state.http_client.get(stream_url.clone()).headers(headers.clone()); match client.send().await { Ok(response) => { let response_headers = get_response_headers(response.headers()); diff --git a/src/api/model/streams/provider_stream_factory.rs b/src/api/model/streams/provider_stream_factory.rs index bb55e74b6..23b7768fc 100644 --- a/src/api/model/streams/provider_stream_factory.rs +++ b/src/api/model/streams/provider_stream_factory.rs @@ -29,7 +29,6 @@ pub struct BufferStreamOptions { item_type: PlaylistItemType, reconnect_enabled: bool, force_reconnect_secs: u32, - connect_timeout_secs: u32, buffer_enabled: bool, buffer_size: usize, share_stream: bool, @@ -46,7 +45,6 @@ impl BufferStreamOptions { item_type, reconnect_enabled: stream_options.stream_retry, force_reconnect_secs: stream_options.stream_force_retry_secs, - connect_timeout_secs: stream_options.stream_connect_timeout_secs, buffer_enabled: stream_options.buffer_enabled, buffer_size: stream_options.buffer_size, share_stream, @@ -94,7 +92,6 @@ struct ProviderStreamOptions { url: Url, reconnect: bool, reconnect_force_secs: u32, - connect_timeout_secs: u32, headers: HeaderMap, range_bytes: Arc>, } @@ -178,7 +175,7 @@ fn get_request_range_start_bytes(req_headers: &HashMap>) -> Opti fn get_client_stream_request_params( req_headers: &HeaderMap, input_headers: Option>, - options: &BufferStreamOptions) -> (usize, Option, bool, u32, u32, HeaderMap) + options: &BufferStreamOptions) -> (usize, Option, bool, u32, HeaderMap) { let stream_buffer_size = if options.is_buffer_enabled() { options.get_stream_buffer_size() } else { 1 }; let filter_header = get_header_filter_for_item_type(options.item_type); @@ -191,31 +188,26 @@ fn get_client_stream_request_params( // We merge configured input headers with the headers from the request. let headers = get_request_headers(input_headers.as_ref(), Some(&req_headers)); - (stream_buffer_size, req_range_start_bytes, options.is_reconnect_enabled(), options.force_reconnect_secs, options.connect_timeout_secs, headers) + (stream_buffer_size, req_range_start_bytes, options.is_reconnect_enabled(), options.force_reconnect_secs, headers) } fn prepare_client(request_client: &Arc, stream_options: &ProviderStreamOptions) -> (reqwest::RequestBuilder, bool) { let url = stream_options.get_url(); let range_start = stream_options.get_total_bytes_send(); let headers = stream_options.get_headers(); - let mut client = request_client.get(url.clone()).headers(headers.clone()); + let mut request_builder = request_client.get(url.clone()).headers(headers.clone()); - let (client_builder, partial) = { + let (client, partial) = { if let Some(range) = range_start { // on reconnect send range header to avoid starting from beginning for vod let range = format!("bytes={range}-", ); - client = client.header(RANGE, range); - (client, true) // partial content + request_builder = request_builder.header(RANGE, range); + (request_builder, true) // partial content } else { - (client, false) + (request_builder, false) } }; - let client = if stream_options.connect_timeout_secs > 0 { - client_builder.timeout(Duration::from_secs(u64::from(stream_options.connect_timeout_secs))) - } else { - client_builder - }; (client, partial) } @@ -338,7 +330,7 @@ fn create_provider_stream_options(stream_url: &Url, req_headers: &HeaderMap, input_headers: Option>, options: &BufferStreamOptions) -> ProviderStreamOptions { - let (buffer_size, req_range_start_bytes, reconnect, reconnect_force_secs, connect_timeout_secs, headers) + let (buffer_size, req_range_start_bytes, reconnect, reconnect_force_secs, headers) = get_client_stream_request_params(req_headers, input_headers, options); let url = stream_url.clone(); let range_bytes = Arc::new(req_range_start_bytes.map(AtomicUsize::new)); @@ -349,7 +341,6 @@ fn create_provider_stream_options(stream_url: &Url, url, reconnect, reconnect_force_secs, - connect_timeout_secs, headers, range_bytes, } diff --git a/src/model/config.rs b/src/model/config.rs index a068a64d9..eb8d57e36 100644 --- a/src/model/config.rs +++ b/src/model/config.rs @@ -599,11 +599,14 @@ pub enum InputType { M3u, #[serde(rename = "xtream")] Xtream, + #[serde(rename = "m3u_batch")] + M3uBatch, } impl InputType { const M3U: &'static str = "m3u"; const XTREAM: &'static str = "xtream"; + const M3U_BATCH: &'static str = "m3u_batch"; } impl Display for InputType { @@ -611,6 +614,7 @@ impl Display for InputType { write!(f, "{}", match self { Self::M3u => Self::M3U, Self::Xtream => Self::XTREAM, + Self::M3uBatch => Self::M3U_BATCH, }) } } @@ -684,12 +688,22 @@ macro_rules! check_input_credentials { if $this.username.is_some() || $this.password.is_some() { debug!("for input type m3u: username and password are ignored"); } + if $this.username.is_none() && $this.password.is_none() { + let (username, password) = get_credentials_from_url_str(&$this.url); + $this.username = username; + $this.password = password; + } } InputType::Xtream => { if $this.username.is_none() || $this.password.is_none() { return Err(info_err!("for input type xtream: username and password are mandatory".to_string())); } } + InputType::M3uBatch => { + if $this.username.is_some() || $this.password.is_some() { + debug!("for input type m3u_batch: username and password are ignored"); + } + } } }; } @@ -725,6 +739,7 @@ impl ConfigInputAlias { self.username = get_trimmed_string(&self.username); self.password = get_trimmed_string(&self.password); check_input_credentials!(self, input_type); + Ok(()) } } @@ -780,11 +795,6 @@ impl ConfigInput { self.username = get_trimmed_string(&self.username); self.password = get_trimmed_string(&self.password); check_input_credentials!(self, self.input_type); - if self.username.is_none() && self.password.is_none() { - let (username, password) = get_credentials_from_url_str(&self.url); - self.username = username; - self.password = password; - } self.persist = get_trimmed_string(&self.persist); if let Some(aliases) = self.aliases.as_mut() { let input_type = &self.input_type; diff --git a/src/processing/processor/playlist.rs b/src/processing/processor/playlist.rs index d0700fe9b..3e4d29a7f 100644 --- a/src/processing/processor/playlist.rs +++ b/src/processing/processor/playlist.rs @@ -317,6 +317,7 @@ async fn process_source(client: Arc, cfg: Arc, source_i let (mut playlistgroups, mut error_list) = match input.input_type { InputType::M3u => m3u::get_m3u_playlist(Arc::clone(&client), &cfg, input, &cfg.working_dir).await, InputType::Xtream => xtream::get_xtream_playlist(Arc::clone(&client), input, &cfg.working_dir).await, + InputType::M3uBatch => (vec![], vec![]) }; let (tvguide, mut tvguide_errors) = if error_list.is_empty() { epg::get_xmltv(Arc::clone(&client), &cfg, input, &cfg.working_dir).await diff --git a/src/utils/file/config_reader.rs b/src/utils/file/config_reader.rs index 7123b3b35..6017ccfa5 100644 --- a/src/utils/file/config_reader.rs +++ b/src/utils/file/config_reader.rs @@ -210,6 +210,9 @@ fn csv_assign_config_input_column(config_input: &mut ConfigInputAlias, input_typ InputType::M3u => { config_input.url = url.to_string(); } + + InputType::M3uBatch => { + } }; }, FIELD_MAX_CON => {