euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-24 00:32:30 +02:00
Code Issues Packages Projects Releases 5 Wiki Activity

Refactored stream handling

Browse Source
This commit is contained in:
euzu
2025-05-09 22:25:44 +02:00
parent 8d36c9c00e
commit a8d6946d80
4 changed files with 25 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/model/model_utils.rs
+4 -2
View File
@@ -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 {
src/api/model/streams/provider_stream_factory.rs
+11 -83
View File
@@ -155,82 +155,6 @@ impl ProviderStreamFactoryOptions {
}
}
//
// #[derive(Debug, Clone)]
// struct ProviderStreamOptions {
// buffer_size: usize,
// continue_flag: Arc<AtomicOnceFlag>,
// url: Url,
// pipe_stream: bool,
// reconnect: bool,
// reconnect_force_secs: u32,
// headers: HeaderMap,
// range_bytes: Arc<Option<AtomicUsize>>,
// }
//
// 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<AtomicOnceFlag> {
// Arc::clone(&self.continue_flag)
// }
//
// // #[inline]
// // pub fn get_continue_flag(&self) -> &Arc<AtomicFlag> {
// // &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<usize> {
// self.range_bytes.as_ref().as_ref().map(|atomic| atomic.load(Ordering::SeqCst))
// }
//
// // pub fn get_range_bytes(&self) -> &Arc<Option<AtomicUsize>> {
// // &self.range_bytes
// // }
//
// #[inline]
// pub fn get_range_bytes_clone(&self) -> Arc<Option<AtomicUsize>> {
// 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<String, Vec<u8>>) -> Option<usize> {
// 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<String, Vec<u8>>) -> Opti
None
}
fn get_host_and_optional_port(url: &Url) -> Option<String> {
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<String> {
// 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<reqwest::Client>, stream_options: &ProviderStreamFactoryOptions) -> (reqwest::RequestBuilder, bool) {
@@ -280,6 +204,10 @@ fn prepare_client(request_client: &Arc<reqwest::Client>, 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));
}
src/utils/constants.rs
+2 -2
View File
@@ -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
src/utils/network/request.rs
+8 -6
View File
@@ -144,8 +144,8 @@ pub fn get_client_request<S: ::std::hash::BuildHasher + Default>(client: &Arc<re
pub fn get_request_headers<S: ::std::hash::BuildHasher + Default>(request_headers: Option<&HashMap<String, String, S>>, custom_headers: Option<&HashMap<String, Vec<u8>, 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<S: ::std::hash::BuildHasher + Default>(request_header
let header_keys: HashSet<String> = 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);
}
}
}
}