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

fixed connection timeout issue

Browse Source
This commit is contained in:
euzu
2025-03-21 17:53:21 +01:00
parent ec19df8695
commit d3124d1464
8 changed files with 50 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/api_utils.rs
+5 -6
View File
@@ -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();
src/api/endpoints/v1_api.rs
+1
View File
@@ -190,6 +190,7 @@ async fn get_playlist(client: Arc<reqwest::Client>, 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<String> = errors.iter().map(std::string::ToString::to_string).collect();
src/api/main_api.rs
+14 -1
View File
@@ -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<PathBuf, std::io::Error> {
@@ -120,9 +122,20 @@ fn create_shared_data(cfg: &Arc<Config>) -> 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(),
src/api/model/streams/provider_stream.rs
+3 -10
View File
@@ -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<HashMap<String, String>>,
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::<Vec<_>>());
@@ -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());
src/api/model/streams/provider_stream_factory.rs
+8 -17
View File
@@ -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<Option<AtomicUsize>>,
}
@@ -178,7 +175,7 @@ fn get_request_range_start_bytes(req_headers: &HashMap<String, Vec<u8>>) -> Opti
fn get_client_stream_request_params(
req_headers: &HeaderMap,
input_headers: Option<HashMap<String, String>>,
options: &BufferStreamOptions) -> (usize, Option<usize>, bool, u32, u32, HeaderMap)
options: &BufferStreamOptions) -> (usize, Option<usize>, 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<reqwest::Client>, 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<HashMap<String, String>>,
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,
}