From c65e9181e4cd04a3c43703fc408726c9def28ed7 Mon Sep 17 00:00:00 2001 From: DarkBreakpoint <243206744+DarkBreakpoint@users.noreply.github.com> Date: Fri, 14 Nov 2025 08:58:11 -0600 Subject: [PATCH] Use tokio::fs for playlist persistence to avoid blocking async paths --- backend/src/utils/file/csv_input_reader.rs | 6 +- backend/src/utils/file/file_utils.rs | 11 +-- backend/src/utils/network/request.rs | 92 +++++++++++++--------- 3 files changed, 63 insertions(+), 46 deletions(-) diff --git a/backend/src/utils/file/csv_input_reader.rs b/backend/src/utils/file/csv_input_reader.rs index 622440654..2801342c9 100644 --- a/backend/src/utils/file/csv_input_reader.rs +++ b/backend/src/utils/file/csv_input_reader.rs @@ -1,5 +1,5 @@ use shared::error::{str_to_io_error, to_io_error}; -use crate::utils::request::{get_local_file_content}; +use crate::utils::request::get_local_file_content_blocking; use crate::utils::EnvResolvingReader; use crate::utils::{file_reader, resolve_relative_path}; use log::error; @@ -152,7 +152,7 @@ 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), Error> { let file_path = get_csv_file_path(file_uri)?; - match get_local_file_content(&file_path) { + match get_local_file_content_blocking(&file_path) { 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()); } -} \ No newline at end of file +} diff --git a/backend/src/utils/file/file_utils.rs b/backend/src/utils/file/file_utils.rs index fd1051215..a52092805 100644 --- a/backend/src/utils/file/file_utils.rs +++ b/backend/src/utils/file/file_utils.rs @@ -129,15 +129,12 @@ pub fn open_file(file_name: &Path) -> Result { File::open(file_name) } -pub fn persist_file(persist_file: Option, text: &str) { +pub async fn persist_file(persist_file: Option, 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}"), } } } diff --git a/backend/src/utils/network/request.rs b/backend/src/utils/network/request.rs index ab4024309..db2ad963c 100644 --- a/backend/src/utils/network/request.rs +++ b/backend/src/utils/network/request.rs @@ -70,12 +70,9 @@ pub async fn get_input_epg_content_as_file(client: Arc, 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 +114,13 @@ pub async fn get_input_text_content(client: Arc, 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,23 +205,36 @@ pub fn get_request_headers(request_header headers } -pub fn get_local_file_content(file_path: &PathBuf) -> Result { - // 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()); - } +fn decode_local_file_bytes(content: Vec) -> Result { + if content.len() >= 2 && is_gzip(&content[0..2]) { + let mut decoder = flate2::read::GzDecoder::new(&content[..]); + let mut decode_buffer = String::new(); + decoder + .read_to_string(&mut decode_buffer) + .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()) + } +} + +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 { + match tokio::fs::read(file_path).await { + Ok(content) => decode_local_file_bytes(content), + Err(_) => Err(local_file_not_found(file_path)), + } +} + +pub fn get_local_file_content_blocking(file_path: &PathBuf) -> Result { + match fs::read(file_path) { + Ok(content) => decode_local_file_bytes(content), + Err(_) => Err(local_file_not_found(file_path)), } - let file_str = file_path.to_str().unwrap_or("?"); - Err(Error::new(ErrorKind::InvalidData, format!("Cant find file {file_str}"))) } @@ -349,26 +356,39 @@ async fn download_epg_content_as_file(client: Arc, input: &Conf } } -pub async fn download_text_content(client: Arc, disabled_headers: Option<&ReverseProxyDisabledHeaderConfig>, input: &InputSource, headers: Option<&HeaderMap>, persist_filepath: Option) -> Result<(String, String), Error> { +pub async fn download_text_content( + client: Arc, + disabled_headers: Option<&ReverseProxyDisabledHeaderConfig>, + input: &InputSource, + headers: Option<&HeaderMap>, + persist_filepath: Option, +) -> Result<(String, String), Error> { if let Ok(url) = input.url.parse::() { 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) + ))) } }