euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-26 16:52:25 +01:00
Code Issues Packages Projects Releases 5 Wiki Activity

single Http Client reuse

Browse Source
This commit is contained in:
euzu
2025-01-09 15:58:38 +01:00
parent 1718e6d52c
commit d92a3bf568
16 changed files with 121 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/processing/playlist_processor.rs
+15 -13
View File
@@ -291,7 +291,7 @@ fn is_target_enabled(target: &ConfigTarget, user_targets: &ProcessTargets) -> bo
(!user_targets.enabled && target.enabled) || (user_targets.enabled && user_targets.has_target(target.id))
}
async fn process_source(cfg: Arc<Config>, source_idx: usize, user_targets: Arc<ProcessTargets>) -> (Vec<InputStats>, Vec<TargetStats>, Vec<M3uFilterError>) {
async fn process_source(client: Arc<reqwest::Client>, cfg: Arc<Config>, source_idx: usize, user_targets: Arc<ProcessTargets>) -> (Vec<InputStats>, Vec<TargetStats>, Vec<M3uFilterError>) {
let source = cfg.sources.get(source_idx).unwrap();
let mut errors = vec![];
let mut input_stats = HashMap::<u16, InputStats>::new();
@@ -304,11 +304,11 @@ async fn process_source(cfg: Arc<Config>, source_idx: usize, user_targets: Arc<P
let input_id = input.id;
if is_input_enabled(enabled_inputs, input.enabled, input_id, &user_targets) {
let (mut playlistgroups, mut error_list) = match input.input_type {
InputType::M3u => download::get_m3u_playlist(&cfg, input, &cfg.working_dir).await,
InputType::Xtream => download::get_xtream_playlist(input, &cfg.working_dir).await,
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,
};
let (tvguide, mut tvguide_errors) = if error_list.is_empty() {
download::get_xmltv(&cfg, input, &cfg.working_dir).await
download::get_xmltv(Arc::clone(&client), &cfg, input, &cfg.working_dir).await
} else {
(None, vec![])
};
@@ -344,7 +344,7 @@ async fn process_source(cfg: Arc<Config>, source_idx: usize, user_targets: Arc<P
debug_if_enabled!("Source has {} groups", source_playlists.iter().map(|fpl| fpl.playlistgroups.len()).sum::<usize>());
for target in &source.targets {
if is_target_enabled(target, &user_targets) {
match process_playlist_for_target(&mut source_playlists, target, &cfg, &mut input_stats, &mut errors).await {
match process_playlist_for_target(Arc::clone(&client), &mut source_playlists, target, &cfg, &mut input_stats, &mut errors).await {
Ok(()) => {
target_stats.push(TargetStats::success(&target.name));
}
@@ -376,7 +376,7 @@ fn create_input_stat(group_count: usize, channel_count: usize, error_count: usiz
}
}
async fn process_sources(config: Arc<Config>, user_targets: Arc<ProcessTargets>) -> (Vec<SourceStats>, Vec<M3uFilterError>) {
async fn process_sources(client: Arc<reqwest::Client>, config: Arc<Config>, user_targets: Arc<ProcessTargets>) -> (Vec<SourceStats>, Vec<M3uFilterError>) {
let mut handle_list = vec![];
let thread_num = config.threads;
let process_parallel = thread_num > 1 && config.sources.len() > 1;
@@ -398,10 +398,11 @@ async fn process_sources(config: Arc<Config>, user_targets: Arc<ProcessTargets>)
let cfg = config.clone();
let usr_trgts = user_targets.clone();
if process_parallel {
let http_client = Arc::clone(&client);
let handles = &mut handle_list;
let process = move || {
System::new().block_on(async {
let (input_stats, target_stats, mut res_errors) = process_source(cfg, index, usr_trgts).await;
let (input_stats, target_stats, mut res_errors) = process_source(Arc::clone(&http_client), cfg, index, usr_trgts).await;
shared_errors.lock().await.append(&mut res_errors);
let process_stats = SourceStats::new(input_stats, target_stats);
shared_stats.lock().await.push(process_stats);
@@ -412,7 +413,7 @@ async fn process_sources(config: Arc<Config>, user_targets: Arc<ProcessTargets>)
handles.drain(..).for_each(|handle| { let _ = handle.join(); });
}
} else {
let (input_stats, target_stats, mut res_errors) = process_source(cfg, index, usr_trgts).await;
let (input_stats, target_stats, mut res_errors) = process_source(Arc::clone(&client), cfg, index, usr_trgts).await;
shared_errors.lock().await.append(&mut res_errors);
let process_stats = SourceStats::new(input_stats, target_stats);
shared_stats.lock().await.push(process_stats);
@@ -485,7 +486,8 @@ fn flatten_groups(playlistgroups: Vec<PlaylistGroup>) -> Vec<PlaylistGroup> {
sort_order
}
async fn process_playlist_for_target(playlists: &mut [FetchedPlaylist<'_>],
async fn process_playlist_for_target(client: Arc<reqwest::Client>,
playlists: &mut [FetchedPlaylist<'_>],
target: &ConfigTarget,
cfg: &Config,
stats: &mut HashMap<u16, InputStats>,
@@ -497,8 +499,8 @@ async fn process_playlist_for_target(playlists: &mut [FetchedPlaylist<'_>],
let mut processed_fetched_playlists: Vec<FetchedPlaylist> = vec![];
for provider_fpl in playlists.iter_mut() {
let mut processed_fpl = execute_pipe(target, &pipe, provider_fpl, &mut duplicates);
playlist_resolve_series(cfg, target, errors, &pipe, provider_fpl, &mut processed_fpl).await;
playlist_resolve_vod(cfg, target, errors, &processed_fpl).await;
playlist_resolve_series(Arc::clone(&client), cfg, target, errors, &pipe, provider_fpl, &mut processed_fpl).await;
playlist_resolve_vod(Arc::clone(&client), cfg, target, errors, &processed_fpl).await;
// stats
let input_stats = stats.get_mut(&processed_fpl.input.id);
if let Some(stat) = input_stats {
@@ -560,9 +562,9 @@ fn process_watch(target: &ConfigTarget, cfg: &Config, new_playlist: &Vec<Playlis
}
}
pub async fn exec_processing(cfg: Arc<Config>, targets: Arc<ProcessTargets>) {
pub async fn exec_processing(client: Arc<reqwest::Client>, cfg: Arc<Config>, targets: Arc<ProcessTargets>) {
let start_time = Instant::now();
let (stats, errors) = process_sources(cfg.clone(), targets.clone()).await;
let (stats, errors) = process_sources(client, cfg.clone(), targets.clone()).await;
// log errors
for err in &errors {
error!("{}", err.message);
src/processing/xtream_processor.rs
+3 -2
View File
@@ -9,6 +9,7 @@ use std::collections::HashMap;
use std::fs::File;
use std::io::{BufWriter, Error, ErrorKind, Write};
use std::path::PathBuf;
use std::sync::Arc;
const FILE_SERIES_INFO: &str = "xtream_series_info";
const FILE_VOD_INFO: &str = "xtream_vod_info";
@@ -53,11 +54,11 @@ macro_rules! create_resolve_options_function_for_xtream_target {
};
}
pub(in crate::processing) async fn playlist_resolve_download_playlist_item(pli: &PlaylistItem, input: &ConfigInput, errors: &mut Vec<M3uFilterError>, resolve_delay: u16, cluster: XtreamCluster) -> Option<String> {
pub(in crate::processing) async fn playlist_resolve_download_playlist_item(client: Arc<reqwest::Client>, pli: &PlaylistItem, input: &ConfigInput, errors: &mut Vec<M3uFilterError>, resolve_delay: u16, cluster: XtreamCluster) -> Option<String> {
let mut result = None;
let provider_id = pli.get_provider_id()?;
if let Some(info_url) = download::get_xtream_player_api_info_url(input, cluster, provider_id) {
result = match download::get_xtream_stream_info_content(&info_url, input).await {
result = match download::get_xtream_stream_info_content(client, &info_url, input).await {
Ok(content) => Some(content),
Err(err) => {
errors.push(info_err!(format!("{err}")));
src/processing/xtream_processor_series.rs
+5 -4
View File
@@ -11,6 +11,7 @@ use crate::{create_resolve_options_function_for_xtream_target, handle_error, han
use std::collections::HashMap;
use std::fs::File;
use std::io::{BufWriter, Write};
use std::sync::Arc;
use std::time::Instant;
use log::{info, log_enabled, Level};
use crate::model::xtream::{XtreamSeriesEpisode, XtreamSeriesInfoEpisode};
@@ -52,7 +53,7 @@ fn should_update_series_info(pli: &PlaylistItem, processed_provider_ids: &HashMa
should_update_info(pli, processed_provider_ids, TAG_SERIES_INFO_LAST_MODIFIED)
}
async fn playlist_resolve_series_info(cfg: &Config, errors: &mut Vec<M3uFilterError>,
async fn playlist_resolve_series_info(client: Arc<reqwest::Client>, cfg: &Config, errors: &mut Vec<M3uFilterError>,
fpl: &mut FetchedPlaylist<'_>, resolve_delay: u16) -> bool {
let mut processed_info_ids = read_processed_series_info_ids(cfg, errors, fpl).await;
// we cant write to the indexed-document directly because of the write lock and time-consuming operation.
@@ -78,7 +79,7 @@ async fn playlist_resolve_series_info(cfg: &Config, errors: &mut Vec<M3uFilterEr
for pli in series_info_iter {
let (should_update, provider_id, ts) = should_update_series_info(pli, &processed_info_ids);
if should_update {
if let Some(content) = playlist_resolve_download_playlist_item(pli, fpl.input, errors, resolve_delay, XtreamCluster::Series).await {
if let Some(content) = playlist_resolve_download_playlist_item(Arc::clone(&client), pli, fpl.input, errors, resolve_delay, XtreamCluster::Series).await {
handle_error_and_return!(write_info_content_to_wal_file(&mut content_writer, provider_id, &content),
|err| errors.push(notify_err!(format!("Failed to resolve series, could not write to content wal file {err}"))));
processed_info_ids.insert(provider_id, ts);
@@ -208,7 +209,7 @@ async fn process_series_info(
}
pub async fn playlist_resolve_series(cfg: &Config, target: &ConfigTarget,
pub async fn playlist_resolve_series(client: Arc<reqwest::Client>, cfg: &Config, target: &ConfigTarget,
errors: &mut Vec<M3uFilterError>,
pipe: &ProcessingPipe,
provider_fpl: &mut FetchedPlaylist<'_>,
@@ -217,7 +218,7 @@ pub async fn playlist_resolve_series(cfg: &Config, target: &ConfigTarget,
let (resolve_series, resolve_delay) = get_resolve_series_options(target, processed_fpl);
if !resolve_series { return; }
if !playlist_resolve_series_info(cfg, errors, processed_fpl, resolve_delay).await { return; }
if !playlist_resolve_series_info(client, cfg, errors, processed_fpl, resolve_delay).await { return; }
let series_playlist = process_series_info(cfg, provider_fpl, errors).await;
if series_playlist.is_empty() { return; }
// original content saved into original list
src/processing/xtream_processor_vod.rs
+3 -2
View File
@@ -9,6 +9,7 @@ use serde_json::{Map, Value};
use std::collections::HashMap;
use std::fs::File;
use std::io::{BufWriter, Write};
use std::sync::Arc;
use std::time::Instant;
use log::{info, log_enabled, Level};
@@ -63,7 +64,7 @@ fn should_update_vod_info(pli: &PlaylistItem, processed_provider_ids: &HashMap<u
should_update_info(pli, processed_provider_ids, TAG_VOD_INFO_ADDED)
}
pub async fn playlist_resolve_vod(cfg: &Config, target: &ConfigTarget, errors: &mut Vec<M3uFilterError>, fpl: &FetchedPlaylist<'_>) {
pub async fn playlist_resolve_vod(client: Arc<reqwest::Client>, cfg: &Config, target: &ConfigTarget, errors: &mut Vec<M3uFilterError>, fpl: &FetchedPlaylist<'_>) {
let (resolve_movies, resolve_delay) = get_resolve_vod_options(target, fpl);
if !resolve_movies { return; }
@@ -92,7 +93,7 @@ pub async fn playlist_resolve_vod(cfg: &Config, target: &ConfigTarget, errors: &
for pli in vod_info_iter {
let (should_update, _provider_id, _ts) = should_update_vod_info(pli, &processed_info_ids);
if should_update {
if let Some(content) = playlist_resolve_download_playlist_item(pli, fpl.input, errors, resolve_delay, XtreamCluster::Video).await {
if let Some(content) = playlist_resolve_download_playlist_item(Arc::clone(&client), pli, fpl.input, errors, resolve_delay, XtreamCluster::Video).await {
if let Some((provider_id, info_record)) = extract_info_record_from_vod_info(&content) {
let ts = info_record.ts;
handle_error_and_return!(write_info_content_to_wal_file(&mut content_writer, provider_id, &content),