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

Merge remote-tracking branch 'dark-breakpoint/async-file-copy' into feature/async_playlist_persistence

Browse Source
This commit is contained in:
euzu
2025-11-14 16:32:24 +01:00
parent 800ae6107e c65e9181e4
commit 66bba7b96b
11 changed files with 80 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+1
View File
@@ -2,6 +2,7 @@
# 3.3.0 (2025-11-xx)
- Avoid blocking the runtime when warming cache
- Normalize FileLockManager paths so aliases share the same lock
- Use async file operations for playlist persistence to avoid blocking async paths
# 3.2.0 (2025-11-14)
- Added `name` attribute to Staged Input.
Cargo.lock
Generated
+3 -3
View File
@@ -1102,7 +1102,7 @@ dependencies = [
[[package]]
name = "frontend"
version = "3.2.1"
version = "3.2.2"
dependencies = [
"anyhow",
"base64",
@@ -3782,7 +3782,7 @@ dependencies = [
[[package]]
name = "shared"
version = "3.2.1"
version = "3.2.2"
dependencies = [
"base64",
"bitflags 2.10.0",
@@ -4331,7 +4331,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
[[package]]
name = "tuliprox"
version = "3.2.1"
version = "3.2.2"
dependencies = [
"arc-swap",
"async-compression",
backend/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "tuliprox"
version = "3.2.1"
version = "3.2.2"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
backend/src/api/config_watch.rs
+1 -1
View File
@@ -82,7 +82,7 @@ impl ConfigFile {
let sources_file = paths.sources_file_path.as_str();
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&app_state.app_config.config);
let mut sources_dto = read_sources_file(sources_file, true, true, config.get_hdhr_device_overview().as_ref())?;
prepare_sources_batch(&mut sources_dto, true)?;
prepare_sources_batch(&mut sources_dto, true).await?;
let sources: SourcesConfig = SourcesConfig::try_from(sources_dto)?;
info!("Loaded sources file {sources_file}");
update_app_state_sources(app_state, sources).await?;
backend/src/api/endpoints/v1_api_config.rs
+1 -1
View File
@@ -93,7 +93,7 @@ async fn config(
let paths = app_state.app_config.paths.load();
match utils::read_app_config_dto(&paths, true, false) {
Ok(mut app_config) => {
if let Err(err) = prepare_sources_batch(&mut app_config.sources, false) {
if let Err(err) = prepare_sources_batch(&mut app_config.sources, false).await {
error!("Failed to prepare sources batch: {err}");
axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()
} else if let Err(err) = prepare_users(&mut app_config, &app_state.app_config).await {
backend/src/utils/file/config_reader.rs
+5 -5
View File
@@ -125,7 +125,7 @@ pub fn read_app_config_dto(paths: &ConfigPaths,
})
}
pub fn prepare_sources_batch(sources: &mut SourcesConfigDto, include_computed: bool) -> Result<(), TuliproxError> {
pub async fn prepare_sources_batch(sources: &mut SourcesConfigDto, include_computed: bool) -> Result<(), TuliproxError> {
let mut current_index = 0;
@@ -148,7 +148,7 @@ pub fn prepare_sources_batch(sources: &mut SourcesConfigDto, include_computed: b
for source in &mut sources.sources {
for input in &mut source.inputs {
match get_batch_aliases(input.input_type, input.url.as_str()) {
match get_batch_aliases(input.input_type, input.url.as_str()).await {
Ok(Some((_, aliases))) => {
if let Some(idx) = input.prepare_batch(aliases, current_index)? {
current_index = idx;
@@ -167,9 +167,9 @@ pub fn prepare_sources_batch(sources: &mut SourcesConfigDto, include_computed: b
Ok(())
}
pub fn get_batch_aliases(input_type: InputType, url: &str) -> Result<Option<(PathBuf, Vec<ConfigInputAliasDto>)>, TuliproxError> {
pub async fn get_batch_aliases(input_type: InputType, url: &str) -> Result<Option<(PathBuf, Vec<ConfigInputAliasDto>)>, TuliproxError> {
if input_type == InputType::M3uBatch || input_type == InputType::XtreamBatch {
return match utils::csv_read_inputs(input_type, url) {
return match utils::csv_read_inputs(input_type, url).await {
Ok((file_path, batch_aliases)) => {
Ok(Some((file_path, batch_aliases)))
}
@@ -213,7 +213,7 @@ pub async fn read_initial_app_config(paths: &mut ConfigPaths,
let config_dto = read_config_file(config_file, resolve_env, include_computed)?;
let mut sources_dto = read_sources_file(sources_file, resolve_env, include_computed, config_dto.get_hdhr_device_overview().as_ref())?;
prepare_sources_batch(&mut sources_dto, include_computed)?;
prepare_sources_batch(&mut sources_dto, include_computed).await?;
let sources: SourcesConfig = SourcesConfig::try_from(sources_dto)?;
let mut config: Config = Config::from(config_dto);
config.prepare(config_path)?;
backend/src/utils/file/csv_input_reader.rs
+4 -4
View File
@@ -1,5 +1,4 @@
use shared::error::{str_to_io_error, to_io_error};
use crate::utils::request::{get_local_file_content};
use crate::utils::EnvResolvingReader;
use crate::utils::{file_reader, resolve_relative_path};
use log::error;
@@ -9,6 +8,7 @@ use std::path::PathBuf;
use url::Url;
use shared::model::{ConfigInputAliasDto, InputType};
use shared::utils::{get_credentials_from_url, trim_last_slash};
use crate::utils::request::get_local_file_content;
const CSV_SEPARATOR: char = ';';
const HEADER_PREFIX: char = '#';
@@ -150,9 +150,9 @@ pub fn csv_read_inputs_from_reader(batch_input_type: InputType, reader: impl Buf
}
pub fn csv_read_inputs(input_type: InputType, file_uri: &str) -> Result<(PathBuf, Vec<ConfigInputAliasDto>), Error> {
pub async fn csv_read_inputs(input_type: InputType, file_uri: &str) -> Result<(PathBuf, Vec<ConfigInputAliasDto>), Error> {
let file_path = get_csv_file_path(file_uri)?;
match get_local_file_content(&file_path) {
match get_local_file_content(&file_path).await {
Ok(content) => {
Ok((file_path, csv_read_inputs_from_reader(input_type, EnvResolvingReader::new(file_reader(Cursor::new(content))))?))
}
@@ -251,4 +251,4 @@ input_2;de566567;de2345f43g5;http://provider_2.tv:8080;1
let resolved = resolve_env_var("${env:HOME}");
assert_eq!(resolved, std::env::var("HOME").unwrap());
}
}
}
backend/src/utils/file/file_utils.rs
+4 -7
View File
@@ -129,15 +129,12 @@ pub fn open_file(file_name: &Path) -> Result<File, std::io::Error> {
File::open(file_name)
}
pub fn persist_file(persist_file: Option<PathBuf>, text: &str) {
pub async fn persist_file(persist_file: Option<PathBuf>, text: &str) {
if let Some(path_buf) = persist_file {
let filename = &path_buf.to_str().unwrap_or("?");
match File::create(&path_buf) {
Ok(mut file) => match file.write_all(text.as_bytes()) {
Ok(()) => debug!("persisted: {filename}"),
Err(e) => error!("failed to persist file {filename}, {e}")
},
Err(e) => error!("failed to persist file {filename}, {e}")
match tokio::fs::write(&path_buf, text).await {
Ok(()) => debug!("persisted: {filename}"),
Err(e) => error!("failed to persist file {filename}, {e}"),
}
}
}
backend/src/utils/network/request.rs
+57 -38
View File
@@ -1,7 +1,6 @@
use std::collections::{HashMap, HashSet};
use std::fs;
use std::fs::File;
use std::io::{BufWriter, Error, ErrorKind, Read, Write};
use std::io::{BufWriter, Error, ErrorKind, Write};
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
@@ -70,12 +69,9 @@ pub async fn get_input_epg_content_as_file(client: Arc<reqwest::Client>, input:
if filepath.exists() {
if let Some(persist_file_value) = persist_filepath {
let to_file = &persist_file_value;
match fs::copy(&filepath, to_file) {
Ok(_) => {}
Err(e) => {
error!("cant persist to: {} => {}", to_file.to_str().unwrap_or("?"), e);
return create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed to persist: {} => {}", to_file.to_str().unwrap_or("?"), e);
}
if let Err(e) = tokio::fs::copy(&filepath, to_file).await {
error!("cant persist to: {} => {}", to_file.to_str().unwrap_or("?"), e);
return create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed to persist: {} => {}", to_file.to_str().unwrap_or("?"), e);
}
}
@@ -117,16 +113,13 @@ pub async fn get_input_text_content(client: Arc<reqwest::Client>, input: &InputS
if filepath.exists() {
if let Some(persist_file_value) = persist_filepath {
let to_file = &persist_file_value;
match fs::copy(&filepath, to_file) {
Ok(_) => {}
Err(e) => {
error!("cant persist to: {} => {}", to_file.to_str().unwrap_or("?"), e);
return create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed to persist: {} => {}", to_file.to_str().unwrap_or("?"), e);
}
if let Err(e) = tokio::fs::copy(&filepath, to_file).await {
error!("cant persist to: {} => {}", to_file.to_str().unwrap_or("?"), e);
return create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed to persist: {} => {}", to_file.to_str().unwrap_or("?"), e);
}
}
match get_local_file_content(&filepath) {
match get_local_file_content(&filepath).await {
Ok(content) => Some(content),
Err(err) => {
return create_tuliprox_error_result!(TuliproxErrorKind::Notify, "Failed : {}", err);
@@ -211,25 +204,38 @@ pub fn get_request_headers<S: ::std::hash::BuildHasher + Default>(request_header
headers
}
pub fn get_local_file_content(file_path: &PathBuf) -> Result<String, Error> {
// Check if the file is accessible
if file_path.exists() && file_path.is_file() {
if let Ok(content) = fs::read(file_path) {
if content.len() >= 2 && is_gzip(&content[0..2]) {
let mut decoder = flate2::read::GzDecoder::new(&content[..]);
let mut decode_buffer = String::new();
return match decoder.read_to_string(&mut decode_buffer) {
Ok(_) => Ok(decode_buffer),
Err(err) => Err(str_to_io_error(&format!("failed to decode gzip content {err}")))
};
}
return Ok(String::from_utf8_lossy(&content).parse().unwrap_or_default());
}
async fn decode_local_file_bytes(content: Vec<u8>) -> Result<String, Error> {
if content.len() >= 2 && is_gzip(&content[0..2]) {
let mut decoder = async_compression::tokio::bufread::GzipDecoder::new(&content[..]);
let mut decode_buffer = String::new();
decoder
.read_to_string(&mut decode_buffer).await
.map_err(|err| str_to_io_error(&format!("failed to decode gzip content {err}")))?;
Ok(decode_buffer)
} else {
Ok(String::from_utf8_lossy(&content).parse().unwrap_or_default())
}
let file_str = file_path.to_str().unwrap_or("?");
Err(Error::new(ErrorKind::InvalidData, format!("Cant find file {file_str}")))
}
fn local_file_not_found(path: &PathBuf) -> Error {
let file_str = path.to_str().unwrap_or("?");
Error::new(ErrorKind::InvalidData, format!("Cant find file {file_str}"))
}
pub async fn get_local_file_content(file_path: &PathBuf) -> Result<String, Error> {
match tokio::fs::read(file_path).await {
Ok(content) => decode_local_file_bytes(content).await,
Err(_) => Err(local_file_not_found(file_path)),
}
}
// pub fn get_local_file_content_blocking(file_path: &PathBuf) -> Result<String, Error> {
// match fs::read(file_path) {
// Ok(content) => decode_local_file_bytes(content).await,
// Err(_) => Err(local_file_not_found(file_path)),
// }
// }
async fn get_remote_content_as_file(client: Arc<reqwest::Client>, input: &ConfigInput, url: &Url, file_path: &Path) -> Result<PathBuf, std::io::Error> {
let start_time = Instant::now();
@@ -349,26 +355,39 @@ async fn download_epg_content_as_file(client: Arc<reqwest::Client>, input: &Conf
}
}
pub async fn download_text_content(client: Arc<reqwest::Client>, disabled_headers: Option<&ReverseProxyDisabledHeaderConfig>, input: &InputSource, headers: Option<&HeaderMap>, persist_filepath: Option<PathBuf>) -> Result<(String, String), Error> {
pub async fn download_text_content(
client: Arc<reqwest::Client>,
disabled_headers: Option<&ReverseProxyDisabledHeaderConfig>,
input: &InputSource,
headers: Option<&HeaderMap>,
persist_filepath: Option<PathBuf>,
) -> Result<(String, String), Error> {
if let Ok(url) = input.url.parse::<url::Url>() {
let result = if url.scheme() == "file" {
url.to_file_path().map_or_else(|()| Err(str_to_io_error(&format!("Unknown file {}", sanitize_sensitive_info(&input.url)))), |file_path|
get_local_file_content(&file_path).map(|c| (c, url.to_string())),
)
match url.to_file_path() {
Ok(file_path) => get_local_file_content(&file_path).await.map(|c| (c, url.to_string())),
Err(()) => Err(str_to_io_error(&format!(
"Unknown file {}",
sanitize_sensitive_info(&input.url)
))),
}
} else {
get_remote_content(client, input, headers, &url, disabled_headers).await
};
match result {
Ok((content, response_url)) => {
if persist_filepath.is_some() {
persist_file(persist_filepath, &content);
persist_file(persist_filepath, &content).await;
}
Ok((content, response_url))
}
Err(err) => Err(err)
Err(err) => Err(err),
}
} else {
Err(str_to_io_error(&format!("Malformed URL {}", sanitize_sensitive_info(&input.url))))
Err(str_to_io_error(&format!(
"Malformed URL {}",
sanitize_sensitive_info(&input.url)
)))
}
}
frontend/Cargo.toml
+2 -2
View File
@@ -1,10 +1,10 @@
[package]
name = "frontend"
version = "3.2.1"
version = "3.2.2"
edition = "2021"
[dependencies]
shared = { version = "3.2.1", path = "../shared" }
shared = { version = "3.2.2", path = "../shared" }
chrono = "0"
yew = "0.21"
yew-router = "0.18"
shared/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "shared"
version = "3.2.1"
version = "3.2.2"
edition = "2021"
[dependencies]