From a8d6946d80796ca2c60c6a9aa207bf0b3ccba72b Mon Sep 17 00:00:00 2001 From: euzu Date: Fri, 9 May 2025 22:25:44 +0200 Subject: [PATCH] Refactored stream handling --- src/api/model/model_utils.rs | 6 +- .../model/streams/provider_stream_factory.rs | 94 +++---------------- src/utils/constants.rs | 4 +- src/utils/network/request.rs | 14 +-- 4 files changed, 25 insertions(+), 93 deletions(-) diff --git a/src/api/model/model_utils.rs b/src/api/model/model_utils.rs index 849c5057b..cfe5e7be1 100644 --- a/src/api/model/model_utils.rs +++ b/src/api/model/model_utils.rs @@ -5,9 +5,10 @@ use reqwest::header::HeaderMap; use crate::utils::{filter_response_header}; pub fn get_response_headers(headers: &HeaderMap) -> Vec<(String, String)> { - let response_headers: Vec<(String, String)> = headers.iter() + let mut response_headers: Vec<(String, String)> = headers.iter() .filter(|(key, _)| filter_response_header(key.as_str())) .map(|(key, value)| (key.to_string(), value.to_str().unwrap().to_string())).collect(); + response_headers.push((axum::http::header::CONNECTION.as_str().to_string(), "keep-alive".to_string())); response_headers } @@ -27,7 +28,8 @@ pub fn get_stream_response_with_headers(custom: Option<(Vec<(String, String)>, S } let default_headers = vec![ - ("content-type", "application/octet-stream") + ("content-type", "application/octet-stream"), + ("connection", "keep-alive"), ]; for (key, value) in default_headers { diff --git a/src/api/model/streams/provider_stream_factory.rs b/src/api/model/streams/provider_stream_factory.rs index d4170e8c1..696aed976 100644 --- a/src/api/model/streams/provider_stream_factory.rs +++ b/src/api/model/streams/provider_stream_factory.rs @@ -155,82 +155,6 @@ impl ProviderStreamFactoryOptions { } } -// -// #[derive(Debug, Clone)] -// struct ProviderStreamOptions { -// buffer_size: usize, -// continue_flag: Arc, -// url: Url, -// pipe_stream: bool, -// reconnect: bool, -// reconnect_force_secs: u32, -// headers: HeaderMap, -// range_bytes: Arc>, -// } -// -// impl ProviderStreamOptions { -// #[inline] -// pub fn is_piped(&self) -> bool { -// self.pipe_stream -// } -// #[inline] -// pub fn is_buffered(&self) -> bool { -// self.buffer_size > 0 -// } -// #[inline] -// pub fn get_buffer_size(&self) -> usize { -// self.buffer_size -// } -// #[inline] -// pub fn get_continue_flag_clone(&self) -> Arc { -// Arc::clone(&self.continue_flag) -// } -// -// // #[inline] -// // pub fn get_continue_flag(&self) -> &Arc { -// // &self.continue_flag -// // } -// -// #[inline] -// pub fn cancel_reconnect(&self) { -// self.continue_flag.notify(); -// } -// -// #[inline] -// pub fn get_url(&self) -> &Url { -// &self.url -// } -// -// #[inline] -// pub fn should_reconnect(&self) -> bool { -// self.reconnect -// } -// -// #[inline] -// pub fn get_headers(&self) -> &HeaderMap { -// &self.headers -// } -// -// #[inline] -// pub fn get_total_bytes_send(&self) -> Option { -// self.range_bytes.as_ref().as_ref().map(|atomic| atomic.load(Ordering::SeqCst)) -// } -// -// // pub fn get_range_bytes(&self) -> &Arc> { -// // &self.range_bytes -// // } -// -// #[inline] -// pub fn get_range_bytes_clone(&self) -> Arc> { -// Arc::clone(&self.range_bytes) -// } -// -// #[inline] -// pub fn should_continue(&self) -> bool { -// self.continue_flag.is_active() -// } -// } - fn get_request_range_start_bytes(req_headers: &HashMap>) -> Option { // range header looks like bytes=1234-5566/2345345 or bytes=0- if let Some(req_range) = req_headers.get(axum::http::header::RANGE.as_str()) { @@ -248,13 +172,13 @@ fn get_request_range_start_bytes(req_headers: &HashMap>) -> Opti None } -fn get_host_and_optional_port(url: &Url) -> Option { - let host = url.host_str()?; - match url.port() { - Some(port) => Some(format!("{host}:{port}")), - None => Some(host.to_string()), - } -} +// fn get_host_and_optional_port(url: &Url) -> Option { +// let host = url.host_str()?; +// match url.port() { +// Some(port) => Some(format!("{host}:{port}")), +// None => Some(host.to_string()), +// } +// } fn prepare_client(request_client: &Arc, stream_options: &ProviderStreamFactoryOptions) -> (reqwest::RequestBuilder, bool) { @@ -280,6 +204,10 @@ fn prepare_client(request_client: &Arc, stream_options: &Provid // } // } + if !headers.contains_key(axum::http::header::CONNECTION) { + headers.insert(axum::http::header::CONNECTION, axum::http::header::HeaderValue::from_static("keep-alive")); + } + if !headers.contains_key(axum::http::header::USER_AGENT) { headers.insert(axum::http::header::USER_AGENT, axum::http::header::HeaderValue::from_static(DEFAULT_USER_AGENT)); } diff --git a/src/utils/constants.rs b/src/utils/constants.rs index 7f795f14f..fccc2cd8e 100644 --- a/src/utils/constants.rs +++ b/src/utils/constants.rs @@ -37,7 +37,7 @@ const SUPPORTED_RESPONSE_HEADERS: &[&str] = &[ "content-range", "vary", "transfer-encoding", - "connection", + //"connection", "access-control-allow-origin", "access-control-allow-credentials", "icy-metadata", @@ -54,7 +54,7 @@ pub fn filter_response_header(key: &str) -> bool { } pub fn filter_request_header(key: &str) -> bool { - if key == "host" { + if key == "host" || key == "connection" { return false; } true diff --git a/src/utils/network/request.rs b/src/utils/network/request.rs index 438b97fd0..19a44dcec 100644 --- a/src/utils/network/request.rs +++ b/src/utils/network/request.rs @@ -144,8 +144,8 @@ pub fn get_client_request(client: &Arc(request_headers: Option<&HashMap>, custom_headers: Option<&HashMap, S>>) -> HeaderMap { let mut headers = HeaderMap::default(); - if let Some(def_headers) = request_headers { - for (key, value) in def_headers { + if let Some(req_headers) = request_headers { + for (key, value) in req_headers { if let (Ok(key), Ok(value)) = (HeaderName::from_bytes(key.as_bytes()), HeaderValue::from_bytes(value.as_bytes())) { if filter_request_header(key.as_str()) { headers.insert(key, value); @@ -157,10 +157,12 @@ pub fn get_request_headers(request_header let header_keys: HashSet = headers.keys().map(|k| k.as_str().to_lowercase()).collect(); for (key, value) in custom { let key_lc = key.to_lowercase(); - if header_keys.contains(key_lc.as_str()) { - // debug_if_enabled!("Ignoring request header '{}={}'", key_lc, String::from_utf8_lossy(value)); - } else if let (Ok(key), Ok(value)) = (HeaderName::from_bytes(key.as_bytes()), HeaderValue::from_bytes(value)) { - headers.insert(key, value); + if filter_request_header(key_lc.as_str()) { + if header_keys.contains(key_lc.as_str()) { + // debug_if_enabled!("Ignoring request header '{}={}'", key_lc, String::from_utf8_lossy(value)); + } else if let (Ok(key), Ok(value)) = (HeaderName::from_bytes(key.as_bytes()), HeaderValue::from_bytes(value)) { + headers.insert(key, value); + } } } }