diff --git a/Cargo.lock b/Cargo.lock index 6c8f5a35f..bc7f3a456 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -374,10 +374,112 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "81953c529336010edd6d8e358f886d9581267795c61b19475b71314bffa46d35" dependencies = [ "concurrent-queue", - "event-listener", + "event-listener 2.5.3", "futures-core", ] +[[package]] +name = "async-channel" +version = "2.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89b47800b0be77592da0afd425cc03468052844aff33b84e33cc696f64e77b6a" +dependencies = [ + "concurrent-queue", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-executor" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30ca9a001c1e8ba5149f91a74362376cc6bc5b919d92d988668657bd570bdcec" +dependencies = [ + "async-task", + "concurrent-queue", + "fastrand 2.2.0", + "futures-lite 2.5.0", + "slab", +] + +[[package]] +name = "async-global-executor" +version = "2.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05b1b633a2115cd122d73b955eadd9916c18c8f510ec9cd1686404c60ad1c29c" +dependencies = [ + "async-channel 2.3.1", + "async-executor", + "async-io", + "async-lock", + "blocking", + "futures-lite 2.5.0", + "once_cell", +] + +[[package]] +name = "async-io" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43a2b323ccce0a1d90b449fd71f2a06ca7faa7c54c2751f06c9bd851fc061059" +dependencies = [ + "async-lock", + "cfg-if", + "concurrent-queue", + "futures-io", + "futures-lite 2.5.0", + "parking", + "polling 3.7.4", + "rustix", + "slab", + "tracing", + "windows-sys 0.59.0", +] + +[[package]] +name = "async-lock" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff6e472cdea888a4bd64f342f09b3f50e1886d32afe8df3d663c01140b811b18" +dependencies = [ + "event-listener 5.3.1", + "event-listener-strategy", + "pin-project-lite", +] + +[[package]] +name = "async-std" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c634475f29802fde2b8f0b505b1bd00dfe4df7d4a000f0b36f7671197d5c3615" +dependencies = [ + "async-channel 1.9.0", + "async-global-executor", + "async-io", + "async-lock", + "crossbeam-utils", + "futures-channel", + "futures-core", + "futures-io", + "futures-lite 2.5.0", + "gloo-timers", + "kv-log-macro", + "log", + "memchr", + "once_cell", + "pin-project-lite", + "pin-utils", + "slab", + "wasm-bindgen-futures", +] + +[[package]] +name = "async-task" +version = "4.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" + [[package]] name = "atomic-waker" version = "1.1.2" @@ -471,6 +573,19 @@ dependencies = [ "generic-array", ] +[[package]] +name = "blocking" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "703f41c54fc768e63e091340b424302bb1c29ef4aa0c7f10fe849dfb114d29ea" +dependencies = [ + "async-channel 2.3.1", + "async-task", + "futures-io", + "futures-lite 2.5.0", + "piper", +] + [[package]] name = "brotli" version = "6.0.0" @@ -849,6 +964,27 @@ version = "2.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" +[[package]] +name = "event-listener" +version = "5.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6032be9bd27023a771701cc49f9f053c751055f71efb2e0ae5c15809093675ba" +dependencies = [ + "concurrent-queue", + "parking", + "pin-project-lite", +] + +[[package]] +name = "event-listener-strategy" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c3e4e0dd3673c1139bf041f3008816d9cf2946bbfac2945c09e523b8d7b05b2" +dependencies = [ + "event-listener 5.3.1", + "pin-project-lite", +] + [[package]] name = "fastrand" version = "1.9.0" @@ -967,6 +1103,19 @@ dependencies = [ "waker-fn", ] +[[package]] +name = "futures-lite" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cef40d21ae2c515b51041df9ed313ed21e572df340ea58a922a0aefe7e8891a1" +dependencies = [ + "fastrand 2.2.0", + "futures-core", + "futures-io", + "parking", + "pin-project-lite", +] + [[package]] name = "futures-macro" version = "0.3.31" @@ -1037,6 +1186,18 @@ version = "0.31.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f" +[[package]] +name = "gloo-timers" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbb143cf96099802033e0d4f4963b19fd2e0b728bcf076cd9cf7f6634f092994" +dependencies = [ + "futures-channel", + "futures-core", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "h2" version = "0.3.26" @@ -1087,6 +1248,12 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hermit-abi" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fbf6a919d6cf397374f7dfeeea91d974c7c0a7221d0d0f4f20d859d329e53fcc" + [[package]] name = "http" version = "0.2.12" @@ -1434,19 +1601,19 @@ version = "1.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "334e04b4d781f436dc315cb1e7515bd96826426345d498149e4bde36b67f8ee9" dependencies = [ - "async-channel", + "async-channel 1.9.0", "castaway", "crossbeam-utils", "curl", "curl-sys", "encoding_rs", - "event-listener", - "futures-lite", + "event-listener 2.5.3", + "futures-lite 1.13.0", "http 0.2.12", "log", "mime", "once_cell", - "polling", + "polling 2.8.0", "serde", "serde_json", "slab", @@ -1497,6 +1664,15 @@ dependencies = [ "simple_asn1", ] +[[package]] +name = "kv-log-macro" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0de8b303297635ad57c9f5059fd9cee7a47f8e8daa09df0fcd07dd39fb22977f" +dependencies = [ + "log", +] + [[package]] name = "language-tags" version = "0.3.2" @@ -1575,6 +1751,9 @@ name = "log" version = "0.4.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24" +dependencies = [ + "value-bag", +] [[package]] name = "m3u-filter" @@ -1586,6 +1765,7 @@ dependencies = [ "actix-server", "actix-web", "actix-web-httpauth", + "async-std", "bincode", "blake3", "bytes", @@ -1931,6 +2111,17 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" +[[package]] +name = "piper" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96c8c490f422ef9a4efd2cb5b42b76c8613d7e7dfc1caf667b8a3350a5acc066" +dependencies = [ + "atomic-waker", + "fastrand 2.2.0", + "futures-io", +] + [[package]] name = "pkg-config" version = "0.3.31" @@ -1953,6 +2144,21 @@ dependencies = [ "windows-sys 0.48.0", ] +[[package]] +name = "polling" +version = "3.7.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a604568c3202727d1507653cb121dbd627a58684eb09a820fd746bee38b4442f" +dependencies = [ + "cfg-if", + "concurrent-queue", + "hermit-abi", + "pin-project-lite", + "rustix", + "tracing", + "windows-sys 0.59.0", +] + [[package]] name = "powerfmt" version = "0.2.0" @@ -2481,7 +2687,7 @@ version = "0.5.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6d7400c0eff44aa2fcb5e31a5f24ba9716ed90138769e4977a2ba6014ae63eb5" dependencies = [ - "async-channel", + "async-channel 1.9.0", "futures-core", "futures-io", ] @@ -2869,6 +3075,12 @@ version = "0.15.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4e8257fbc510f0a46eb602c10215901938b5c2a7d5e70fc11483b1d3c9b5b18c" +[[package]] +name = "value-bag" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ef4c4aa54d5d05a279399bfa921ec387b7aba77caf7a682ae8d86785b8fdad2" + [[package]] name = "vcpkg" version = "0.2.15" diff --git a/Cargo.toml b/Cargo.toml index 60eefdf4e..5b0b0f15f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -50,3 +50,4 @@ flate2 = "1" time = "0.3" blake3 = "1.5" bytes = "1.8.0" +async-std = "1.13" diff --git a/bin/release.sh b/bin/release.sh index a4f9001a0..2406ac080 100755 --- a/bin/release.sh +++ b/bin/release.sh @@ -39,6 +39,8 @@ declare -A BINARIES=( [RASPI]=armv7-unknown-linux-musleabihf/release/m3u-filter ) +WORKING_DIR=$(pwd) + # Build binaries ./bin/build_lin_static.sh ./bin/build_raspi.sh @@ -78,6 +80,8 @@ mv "${ARCS[@]}" checksum.txt "$RELEASE_DIR" # Clean up build directories rm -rf "${DIRS[@]}" +cd "${WORKING_DIR}"; + # Commit and tag release git add . git commit -m "release ${VERSION}" diff --git a/src/api/api_model.rs b/src/api/api_model.rs deleted file mode 100644 index 0db7737da..000000000 --- a/src/api/api_model.rs +++ /dev/null @@ -1,304 +0,0 @@ -use std::collections::VecDeque; -use std::ffi::OsStr; -use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex, RwLock}; - -use actix_web::web; -use chrono::{Duration, Local}; -use serde::{Deserialize, Serialize}; -use unidecode::unidecode; - -use crate::model::api_proxy::{ApiProxyConfig, ApiProxyServerInfo, ProxyUserCredentials}; -use crate::model::config::{Config, ConfigApi, ConfigRename, ConfigSort, ConfigTargetOptions, InputType, MessagingConfig, ProcessTargets, TargetOutput, VideoConfig, VideoDownloadConfig}; -use crate::model::config::ProcessingOrder; -use crate::repository::storage::{hash_string_as_hex}; - -/// File-Download information. -#[derive(Clone)] -pub struct FileDownload { - /// uuid of the download for identification. - pub uuid: String, - /// `file_dir` is the directory where the download should be placed. - pub file_dir: PathBuf, - /// `file_path` is the complete path including the filename. - pub file_path: PathBuf, - /// filename is the filename. - pub filename: String, - /// url is the download url. - pub url: reqwest::Url, - /// finished is true, if download is finished, otherweise false - pub finished: bool, - /// the filesize. - pub size: u64, - /// Optional error if something goes wrong during downloading. - pub error: Option, -} - -/// Returns the directory for th file download. -/// if option `organize_into_directories` is set, the root directory is determined. -/// - For series, the episode pattern is used to determine the sub directory for the series. -/// - For vod files, the title is used to determine the sub directory. -/// -/// # Arguments -/// * `download_cfg` the download configuration -/// * `filestem` the prepared filestem to use as sub directory -/// -fn get_download_directory(download_cfg: &VideoDownloadConfig, filestem: &str) -> PathBuf { - if download_cfg.organize_into_directories { - let mut stem = filestem; - if let Some(re) = &download_cfg.t_re_episode_pattern { - if let Some(captures) = re.captures(stem) { - if let Some(episode) = captures.name("episode") { - if !episode.as_str().is_empty() { - stem = &stem[..episode.start()]; - } - } - } - } - let re_ending = download_cfg.t_re_remove_filename_ending.as_ref().unwrap(); - let dir_name = re_ending.replace(stem, ""); - let file_dir: PathBuf = [download_cfg.directory.as_ref().unwrap(), dir_name.as_ref()].iter().collect(); - file_dir - } else { - PathBuf::from(download_cfg.directory.as_ref().unwrap()) - } -} - -const FILENAME_TRIM_PATTERNS: &[char] = &['.', '-', '_']; - -impl FileDownload { - - // TODO read header size info and restart support - // "content-type" => ".../..." - // "content-length" => "1975828544" - // "accept-ranges" => "0-1975828544" - // "content-range" => "bytes 0-1975828543/1975828544" - - pub fn new(req_url: &str, req_filename: &str, download_cfg: &VideoDownloadConfig) -> Option { - match reqwest::Url::parse(req_url) { - Ok(url) => { - let filename_re = download_cfg.t_re_filename.as_ref().unwrap(); - let tmp_filename = filename_re.replace_all(&unidecode(req_filename) - .replace(' ', "_"), "") - .replace("__", "_") - .replace("_-_", "-"); - let filename_path = Path::new(&tmp_filename); - let file_stem = filename_path.file_stem().and_then(OsStr::to_str).unwrap_or("").trim_matches(FILENAME_TRIM_PATTERNS); - let file_ext = filename_path.extension().and_then(OsStr::to_str).unwrap_or(""); - - let mut filename = format!("{file_stem}.{file_ext}"); - let file_dir = get_download_directory(download_cfg, file_stem); - let mut file_path: PathBuf = file_dir.clone(); - file_path.push(&filename); - let mut x: usize = 1; - while file_path.is_file() { - filename = format!("{file_stem}_{x}.{file_ext}"); - file_path.clone_from(&file_dir); - file_path.push(&filename); - x += 1; - } - - file_path.to_str()?; - - Some(Self { - uuid: hash_string_as_hex(req_url), - file_dir, - file_path, - filename, - url, - finished: false, - size: 0, - error: None, - }) - } - Err(_) => None - } - } -} - - -pub struct DownloadQueue { - pub queue: Arc>>, - pub active: Arc>>, - pub finished: Arc>>, -} - -pub struct AppState { - pub config: Arc, - pub targets: Arc, - pub downloads: Arc, -} - -#[derive(Serialize)] -pub struct XtreamUserInfo { - pub active_cons: String, - pub allowed_output_formats: Vec, - //["ts"], - pub auth: u16, - // 0 | 1 - pub created_at: i64, - //1623429679, - pub exp_date: i64, - //1628755200, - pub is_trial: String, - // 0 | 1 - pub max_connections: String, - pub message: String, - pub password: String, - pub username: String, - pub status: String, // "Active" -} - -#[derive(Serialize)] -pub struct XtreamServerInfo { - pub url: String, - pub port: String, - pub https_port: String, - pub server_protocol: String, - // http, https - pub rtmp_port: String, - pub timezone: String, - pub timestamp_now: i64, - pub time_now: String, //"2021-06-28 17:07:37" -} - -#[derive(Serialize)] -pub struct XtreamAuthorizationResponse { - pub user_info: XtreamUserInfo, - pub server_info: XtreamServerInfo, -} - -impl XtreamAuthorizationResponse { - pub fn new(server_info: &ApiProxyServerInfo, user: &ProxyUserCredentials) -> Self { - let now = Local::now(); - Self { - user_info: XtreamUserInfo { - active_cons: "0".to_string(), - allowed_output_formats: Vec::from(["ts".to_string(), "m3u8".to_string(), "rtmp".to_string()]), - auth: 1, - created_at: (now - Duration::days(365)).timestamp(), // fake - exp_date: (now + Duration::days(365)).timestamp(), // fake - is_trial: "0".to_string(), - max_connections: "1".to_string(), - message: server_info.message.to_string(), - password: user.password.to_string(), - username: user.username.to_string(), - status: "Active".to_string(), - }, - server_info: XtreamServerInfo { - url: server_info.host.clone(), - port: server_info.http_port.clone(), - https_port: server_info.https_port.clone(), - server_protocol: server_info.protocol.clone(), - rtmp_port: server_info.rtmp_port.clone(), - timezone: server_info.timezone.to_string(), - timestamp_now: now.timestamp(), - time_now: now.format("%Y-%m-%d %H:%M:%S").to_string(), - }, - } - } -} - - -#[derive(Debug, serde::Serialize, serde::Deserialize, Default)] -pub struct UserApiRequest { - #[serde(default)] - pub username: String, - #[serde(default)] - pub password: String, - #[serde(default)] - pub token: String, - #[serde(default)] - pub action: String, - #[serde(default)] - pub series_id: String, - #[serde(default)] - pub vod_id: String, - #[serde(default)] - pub stream_id: String, - #[serde(default)] - pub category_id: String, - #[serde(default)] - pub limit: String, - #[serde(default)] - pub start: String, - #[serde(default)] - pub end: String, - #[serde(default)] - pub stream: String, - #[serde(default)] - pub duration: String, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct ServerInputConfig { - pub id: u16, - pub input_type: InputType, - pub url: String, - pub username: Option, - pub password: Option, - pub persist: Option, - pub name: Option, - pub enabled: bool, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct ServerTargetConfig { - pub id: u16, - pub enabled: bool, - pub name: String, - pub options: Option, - pub sort: Option, - pub filter: String, - #[serde(alias = "type")] - pub output: Vec, - pub rename: Option>, - pub mapping: Option>, - pub processing_order: ProcessingOrder, - pub watch: Option>, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct ServerSourceConfig { - pub inputs: Vec, - pub targets: Vec, -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct ServerConfig { - pub api: ConfigApi, - pub threads: u8, - pub working_dir: String, - pub backup_dir: Option, - pub schedule: Option, - pub sources: Vec, - pub messaging: Option, - pub video: Option, - pub api_proxy: Option, -} - - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct PlaylistRequest { - pub url: Option, - pub input_id: Option, -} - -impl From> for PlaylistRequest { - fn from(req: web::Json) -> Self { - req.clone() - } -} - -#[derive(Deserialize, Serialize, Debug, Clone)] -pub struct FileDownloadRequest { - pub url: String, - pub filename: String, -} - -impl From> for FileDownloadRequest { - fn from(req: web::Json) -> Self { - req.clone() - } -} - diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs index e993ff425..2ad0be6e2 100644 --- a/src/api/api_utils.rs +++ b/src/api/api_utils.rs @@ -1,14 +1,17 @@ -use std::collections::HashMap; -use std::path::{Path}; -use actix_web::http::header::{CACHE_CONTROL, HeaderValue}; -use actix_web::{HttpRequest, HttpResponse}; -use log::{debug, error, log_enabled, Level}; -use url::Url; -use crate::api::api_model::{AppState, UserApiRequest}; +use crate::api::model::app_state::AppState; +use crate::api::model::request::UserApiRequest; +use crate::api::model::shared_stream::{NotifyStream, SharedStream}; use crate::model::api_proxy::{ApiProxyServerInfo, ProxyUserCredentials}; -use crate::model::config::{Config, ConfigTarget, ConfigInput}; +use crate::model::config::{Config, ConfigInput, ConfigTarget}; use crate::utils::request_utils; use crate::utils::request_utils::mask_sensitive_info; +use actix_web::http::header::{HeaderValue, CACHE_CONTROL}; +use actix_web::{HttpRequest, HttpResponse}; +use log::{debug, error, log_enabled, Level}; +use std::collections::HashMap; +use std::path::Path; +use std::sync::atomic::AtomicU32; +use url::Url; pub async fn serve_file(file_path: &Path, req: &HttpRequest, mime_type: mime::Mime) -> HttpResponse { if file_path.exists() { @@ -24,7 +27,7 @@ pub async fn serve_file(file_path: &Path, req: &HttpRequest, mime_type: mime::Mi } pub fn get_user_target_by_credentials<'a>(username: &str, password: &str, api_req: &'a UserApiRequest, - app_state: &'a AppState) -> Option<(ProxyUserCredentials, &'a ConfigTarget)> { + app_state: &'a AppState) -> Option<(ProxyUserCredentials, &'a ConfigTarget)> { if !username.is_empty() && !password.is_empty() { app_state.config.get_target_for_user(username, password) } else { @@ -49,11 +52,19 @@ pub fn get_user_server_info(cfg: &Config, user: &ProxyUserCredentials) -> ApiPro server_info_list.iter().find(|c| c.name.eq(server_info_name)).map_or_else(|| server_info_list.first().unwrap().clone(), std::clone::Clone::clone) } -pub async fn stream_response(stream_url: &str, req: &HttpRequest, input: Option<&ConfigInput>) -> HttpResponse { +pub async fn stream_response(app_state: &AppState, stream_url: &str, req: &HttpRequest, input: Option<&ConfigInput>) -> HttpResponse { let req_headers: HashMap<&str, &[u8]> = req.headers().iter().map(|(k, v)| (k.as_str(), v.as_bytes())).collect(); if log_enabled!(Level::Debug) { debug!("Try to open stream {}", mask_sensitive_info(stream_url)); } + + if let shared_streams = app_state.shared_streams.lock().await { + if let Some(shared_stream) = shared_streams.get(stream_url) { + shared_stream.client_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + } + } + + if let Ok(url) = Url::parse(stream_url) { let client = request_utils::get_client_request(input, &url, Some(&req_headers)); match client.send().await { @@ -63,7 +74,29 @@ pub async fn stream_response(stream_url: &str, req: &HttpRequest, input: Option< response.headers().iter().for_each(|(k, v)| { response_builder.insert_header((k.as_str(), v.as_ref())); }); - return response_builder.body(actix_web::body::BodyStream::new(response.bytes_stream())); + let byte_stream = Box::pin(response.bytes_stream()); + let cloned_byte_stream = Box::pin(byte_stream.clone()); + let _ = app_state.shared_streams.lock().await.insert(stream_url.to_string(), + SharedStream { + data_stream: byte_stream, + client_count: AtomicU32::new(0), + }, + ); + let (stream, notify) = NotifyStream::new(cloned_byte_stream); + let notify_stream_url = stream_url.to_string(); + actix_rt::spawn(async move { + let _ = notify.await; + debug!("connection closed {notify_stream_url}"); + if let mut shared_streams = app_state.shared_streams.lock().await { + if let Some(shared_stream) = shared_streams.get(stream_url) { + let cur_count = shared_stream.client_count.fetch_sub(1u32, std::sync::atomic::Ordering::SeqCst); + if cur_count == 1 { + shared_streams.remove(stream_url); + } + } + } + }); + return response_builder.body(actix_web::body::BodyStream::new(stream)); } if log_enabled!(Level::Debug) { debug!("Failed to open stream got status {} for {}", response.status(), mask_sensitive_info(stream_url)); diff --git a/src/api/download_api.rs b/src/api/download_api.rs index c5b7a744a..195c6044b 100644 --- a/src/api/download_api.rs +++ b/src/api/download_api.rs @@ -1,18 +1,20 @@ +use crate::model::config::VideoDownloadConfig; +use crate::utils::request_utils; +use actix_web::{web, HttpResponse}; +use futures::stream::TryStreamExt; +use log::info; +use serde_json::{json, Value}; use std::fs::File; -use std::{fs, io}; use std::io::{ErrorKind, Write}; use std::ops::Deref; -use std::sync::{Arc, RwLock}; -use actix_web::{HttpResponse, web}; -use serde_json::{json, Value}; -use crate::api::api_model::{AppState, DownloadQueue, FileDownload, FileDownloadRequest}; -use crate::model::config::{VideoDownloadConfig}; -use futures::stream::TryStreamExt; -use log::{info}; -use crate::utils::{request_utils}; +use std::sync::{Arc}; +use async_std::sync::RwLock; +use std::{fs, io}; +use crate::api::model::app_state::AppState; +use crate::api::model::download::{DownloadQueue, FileDownload, FileDownloadRequest}; async fn download_file(active: Arc>>, client: &reqwest::Client) -> Result<(), String> { - let file_download = { active.read().unwrap().as_ref().unwrap().clone() }; + let file_download = { active.read().await.as_ref().unwrap().clone() }; match client.get(file_download.url.clone()).send().await { Ok(response) => { match fs::create_dir_all(&file_download.file_dir) { @@ -30,14 +32,14 @@ async fn download_file(active: Arc>>, client: &reqwe match file.write_all(&chunk) { Ok(()) => { downloaded += chunk.len() as u64; - active.write().unwrap().as_mut().unwrap().size = downloaded; + active.write().await.as_mut().unwrap().size = downloaded; } Err(err) => return Err(format!("Error while writing to file: {file_path_str} {err}")) } } else { let megabytes = request_utils::bytes_to_megabytes(downloaded); info!("Downloaded {}, filesize: {}MB", file_path_str, megabytes); - active.write().unwrap().as_mut().unwrap().size = downloaded; + active.write().await.as_mut().unwrap().size = downloaded; return Ok(()); } } @@ -58,33 +60,33 @@ async fn download_file(active: Arc>>, client: &reqwe } } -fn run_download_queue(download_cfg: &VideoDownloadConfig, download_queue: &Arc) -> Result<(), String> { - let next_download = download_queue.as_ref().queue.lock().unwrap().pop_front(); +async fn run_download_queue(download_cfg: &VideoDownloadConfig, download_queue: &Arc) -> Result<(), String> { + let next_download = download_queue.as_ref().queue.lock().await.pop_front(); if next_download.is_some() { - { *download_queue.as_ref().active.write().unwrap() = next_download; } + { *download_queue.as_ref().active.write().await = next_download; } let headers = request_utils::get_request_headers(Some(&download_cfg.headers), None); let dq = Arc::clone(download_queue); match reqwest::Client::builder().default_headers(headers).build() { Ok(client) => { actix_rt::spawn(async move { loop { - if dq.active.read().unwrap().deref().is_some() { + if dq.active.read().await.deref().is_some() { match download_file(Arc::clone(&dq.active), &client).await { Ok(()) => { - if let Some(fd) = &mut *dq.active.write().unwrap() { + if let Some(fd) = &mut *dq.active.write().await { fd.finished = true; - dq.finished.write().unwrap().push(fd.clone()); + dq.finished.write().await.push(fd.clone()); } } Err(err) => { - if let Some(fd) = &mut *dq.active.write().unwrap() { + if let Some(fd) = &mut *dq.active.write().await { fd.finished = true; fd.error = Some(err); - dq.finished.write().unwrap().push(fd.clone()); + dq.finished.write().await.push(fd.clone()); } } } - *dq.active.write().unwrap() = dq.queue.lock().unwrap().pop_front(); + *dq.active.write().await = dq.queue.lock().await.pop_front(); } else { break; } @@ -117,9 +119,9 @@ pub async fn queue_download_file( match FileDownload::new(req.url.as_str(), req.filename.as_str(), download_cfg) { Some(file_download) => { let response = HttpResponse::Ok().json(download_info!(file_download)); - app_state.downloads.queue.lock().unwrap().push_back(file_download); - if app_state.downloads.active.read().unwrap().is_none() { - match run_download_queue(download_cfg, &app_state.downloads) { + app_state.downloads.queue.lock().await.push_back(file_download); + if app_state.downloads.active.read().await.is_none() { + match run_download_queue(download_cfg, &app_state.downloads).await { Ok(()) => {} Err(err) => return HttpResponse::InternalServerError().json(json!({"error": err})), } @@ -136,10 +138,10 @@ pub async fn queue_download_file( pub async fn download_file_info( app_state: web::Data, ) -> HttpResponse { - let finished_list: &[Value] = &app_state.downloads.finished.write().unwrap().drain(..) + let finished_list: &[Value] = &app_state.downloads.finished.write().await.drain(..) .map(|fd| download_info!(fd)).collect::>(); - (*app_state.downloads.active.read().unwrap()).as_ref().map_or_else(|| HttpResponse::Ok().json(json!({ + (*app_state.downloads.active.read().await).as_ref().map_or_else(|| HttpResponse::Ok().json(json!({ "completed": true, "downloads": finished_list })), |file_download| HttpResponse::Ok().json(json!({ "completed": false, "downloads": finished_list, "active": download_info!(file_download) diff --git a/src/api/m3u_api.rs b/src/api/m3u_api.rs index 2dde1acf8..6f8c29b89 100644 --- a/src/api/m3u_api.rs +++ b/src/api/m3u_api.rs @@ -4,20 +4,21 @@ use futures::{stream}; use bytes::Bytes; use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, stream_response}; -use crate::api::api_model::{AppState, UserApiRequest}; +use crate::api::model::app_state::AppState; +use crate::api::model::request::UserApiRequest; use crate::model::api_proxy::ProxyType; use crate::model::config::TargetType; use crate::repository::m3u_repository::{m3u_get_file_paths, m3u_get_item_for_stream_id, m3u_load_rewrite_playlist}; use crate::repository::storage::get_target_storage_path; use crate::utils::request_utils::mask_sensitive_info; -fn m3u_api( +async fn m3u_api( api_req: &UserApiRequest, app_state: &AppState, ) -> HttpResponse { match get_user_target(api_req, app_state) { Some((user, target)) => { - match m3u_load_rewrite_playlist(&app_state.config, target, &user) { + match m3u_load_rewrite_playlist(&app_state.config, target, &user).await { Ok(m3u_iter) => { // Convert the iterator into a stream of `Bytes` let content_stream = stream::iter(m3u_iter.map(|line| Ok::(Bytes::from(format!("{line}\n"))))); @@ -38,13 +39,13 @@ fn m3u_api( async fn m3u_api_get( api_req: web::Query, app_state: web::Data, ) -> HttpResponse { - m3u_api(&api_req.into_inner(), &app_state) + m3u_api(&api_req.into_inner(), &app_state).await } async fn m3u_api_post( api_req: web::Form, app_state: web::Data, ) -> HttpResponse { - m3u_api(&api_req.into_inner(), &app_state) + m3u_api(&api_req.into_inner(), &app_state).await } async fn m3u_api_stream( @@ -60,14 +61,14 @@ async fn m3u_api_stream( match get_target_storage_path(&app_state.config, target.name.as_str()) { Some(target_path) => { let (m3u_path, idx_path) = m3u_get_file_paths(&target_path); - match m3u_get_item_for_stream_id(&app_state.config, m3u_stream_id, &m3u_path, &idx_path) { + match m3u_get_item_for_stream_id(&app_state.config, m3u_stream_id, &m3u_path, &idx_path).await { Ok(m3u_item) => { if user.proxy == ProxyType::Redirect { let stream_url = m3u_item.url; debug!("Redirecting stream request to {}", mask_sensitive_info(&stream_url)); return HttpResponse::Found().insert_header(("Location", stream_url.to_string())).finish(); } - return stream_response(m3u_item.url.as_str(), &req, None).await; + return stream_response(&app_state, m3u_item.url.as_str(), &req, None).await; } Err(err) => { error!("Failed to get m3u url: {}", mask_sensitive_info(err.to_string().as_str())); diff --git a/src/api/main_api.rs b/src/api/main_api.rs index 7bfc0cb37..463476252 100644 --- a/src/api/main_api.rs +++ b/src/api/main_api.rs @@ -1,15 +1,16 @@ -use std::collections::VecDeque; +use std::collections::{HashMap, VecDeque}; use std::io::ErrorKind; use std::path::PathBuf; -use std::sync::{Arc, Mutex, RwLock}; - +use std::sync::{Arc}; use actix_cors::Cors; -use actix_web::{App, HttpResponse, HttpServer, web}; -use actix_web::middleware::{Logger}; +use actix_web::middleware::Logger; +use actix_web::{web, App, HttpResponse, HttpServer}; +use async_std::sync::{RwLock, Mutex}; use log::info; -use crate::api::api_model::{AppState, DownloadQueue}; use crate::api::m3u_api::m3u_api_register; +use crate::api::model::app_state::AppState; +use crate::api::model::download::DownloadQueue; use crate::api::scheduler::start_scheduler; use crate::api::v1_api::v1_api_register; use crate::api::web_index::index_register; @@ -32,10 +33,10 @@ fn get_web_dir_path(web_ui_enabled: bool, web_root: &str) -> Result HttpResponse { let ts = chrono::offset::Local::now().format("%Y-%m-%d %H:%M:%S").to_string(); - HttpResponse::Ok().json( Healthcheck { + HttpResponse::Ok().json(Healthcheck { status: "ok".to_string(), version: VERSION.to_string(), - time: ts + time: ts, }) } @@ -59,6 +60,7 @@ pub async fn start_server(cfg: Arc, targets: Arc) -> fut active: Arc::from(RwLock::new(None)), finished: Arc::from(RwLock::new(Vec::new())), }), + shared_streams: Arc::new(Mutex::new(HashMap::new())), }); // Scheduler diff --git a/src/api/mod.rs b/src/api/mod.rs index 3ae92b711..737f4af48 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1,5 +1,4 @@ pub mod api_utils; -pub mod api_model; pub mod main_api; mod download_api; mod v1_api; @@ -7,4 +6,6 @@ mod xtream_api; mod m3u_api; mod xmltv_api; mod scheduler; -mod web_index; \ No newline at end of file +mod web_index; + +pub(crate) mod model; \ No newline at end of file diff --git a/src/api/model/app_state.rs b/src/api/model/app_state.rs new file mode 100644 index 000000000..c539e10b2 --- /dev/null +++ b/src/api/model/app_state.rs @@ -0,0 +1,13 @@ +use std::collections::HashMap; +use std::sync::Arc; +use async_std::sync::{Mutex}; +use crate::api::model::download::DownloadQueue; +use crate::api::model::shared_stream::SharedStream; +use crate::model::config::{Config, ProcessTargets}; + +pub struct AppState { + pub config: Arc, + pub targets: Arc, + pub downloads: Arc, + pub shared_streams: Arc>>, +} diff --git a/src/api/model/config.rs b/src/api/model/config.rs new file mode 100644 index 000000000..912acc668 --- /dev/null +++ b/src/api/model/config.rs @@ -0,0 +1,53 @@ +use serde::{Deserialize, Serialize}; + +use crate::model::api_proxy::ApiProxyConfig; +use crate::model::config::ProcessingOrder; +use crate::model::config::{ConfigApi, ConfigRename, ConfigSort, ConfigTargetOptions, InputType, MessagingConfig, TargetOutput, VideoConfig}; + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ServerInputConfig { + pub id: u16, + pub input_type: InputType, + pub url: String, + pub username: Option, + pub password: Option, + pub persist: Option, + pub name: Option, + pub enabled: bool, +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ServerTargetConfig { + pub id: u16, + pub enabled: bool, + pub name: String, + pub options: Option, + pub sort: Option, + pub filter: String, + #[serde(alias = "type")] + pub output: Vec, + pub rename: Option>, + pub mapping: Option>, + pub processing_order: ProcessingOrder, + pub watch: Option>, +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ServerSourceConfig { + pub inputs: Vec, + pub targets: Vec, +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct ServerConfig { + pub api: ConfigApi, + pub threads: u8, + pub working_dir: String, + pub backup_dir: Option, + pub schedule: Option, + pub sources: Vec, + pub messaging: Option, + pub video: Option, + pub api_proxy: Option, +} + diff --git a/src/api/model/download.rs b/src/api/model/download.rs new file mode 100644 index 000000000..c6bdd1087 --- /dev/null +++ b/src/api/model/download.rs @@ -0,0 +1,133 @@ +use std::collections::VecDeque; +use std::ffi::OsStr; +use std::path::{Path, PathBuf}; +use std::sync::{Arc}; +use async_std::sync::{RwLock, Mutex}; +use actix_web::web; +use serde::{Deserialize, Serialize}; +use unidecode::unidecode; + +use crate::model::config::VideoDownloadConfig; +use crate::repository::storage::hash_string_as_hex; + +/// File-Download information. +#[derive(Clone)] +pub struct FileDownload { + /// uuid of the download for identification. + pub uuid: String, + /// `file_dir` is the directory where the download should be placed. + pub file_dir: PathBuf, + /// `file_path` is the complete path including the filename. + pub file_path: PathBuf, + /// filename is the filename. + pub filename: String, + /// url is the download url. + pub url: reqwest::Url, + /// finished is true, if download is finished, otherweise false + pub finished: bool, + /// the filesize. + pub size: u64, + /// Optional error if something goes wrong during downloading. + pub error: Option, +} + +/// Returns the directory for th file download. +/// if option `organize_into_directories` is set, the root directory is determined. +/// - For series, the episode pattern is used to determine the sub directory for the series. +/// - For vod files, the title is used to determine the sub directory. +/// +/// # Arguments +/// * `download_cfg` the download configuration +/// * `filestem` the prepared filestem to use as sub directory +/// +fn get_download_directory(download_cfg: &VideoDownloadConfig, filestem: &str) -> PathBuf { + if download_cfg.organize_into_directories { + let mut stem = filestem; + if let Some(re) = &download_cfg.t_re_episode_pattern { + if let Some(captures) = re.captures(stem) { + if let Some(episode) = captures.name("episode") { + if !episode.as_str().is_empty() { + stem = &stem[..episode.start()]; + } + } + } + } + let re_ending = download_cfg.t_re_remove_filename_ending.as_ref().unwrap(); + let dir_name = re_ending.replace(stem, ""); + let file_dir: PathBuf = [download_cfg.directory.as_ref().unwrap(), dir_name.as_ref()].iter().collect(); + file_dir + } else { + PathBuf::from(download_cfg.directory.as_ref().unwrap()) + } +} + +const FILENAME_TRIM_PATTERNS: &[char] = &['.', '-', '_']; + +impl FileDownload { + + // TODO read header size info and restart support + // "content-type" => ".../..." + // "content-length" => "1975828544" + // "accept-ranges" => "0-1975828544" + // "content-range" => "bytes 0-1975828543/1975828544" + + pub fn new(req_url: &str, req_filename: &str, download_cfg: &VideoDownloadConfig) -> Option { + match reqwest::Url::parse(req_url) { + Ok(url) => { + let filename_re = download_cfg.t_re_filename.as_ref().unwrap(); + let tmp_filename = filename_re.replace_all(&unidecode(req_filename) + .replace(' ', "_"), "") + .replace("__", "_") + .replace("_-_", "-"); + let filename_path = Path::new(&tmp_filename); + let file_stem = filename_path.file_stem().and_then(OsStr::to_str).unwrap_or("").trim_matches(FILENAME_TRIM_PATTERNS); + let file_ext = filename_path.extension().and_then(OsStr::to_str).unwrap_or(""); + + let mut filename = format!("{file_stem}.{file_ext}"); + let file_dir = get_download_directory(download_cfg, file_stem); + let mut file_path: PathBuf = file_dir.clone(); + file_path.push(&filename); + let mut x: usize = 1; + while file_path.is_file() { + filename = format!("{file_stem}_{x}.{file_ext}"); + file_path.clone_from(&file_dir); + file_path.push(&filename); + x += 1; + } + + file_path.to_str()?; + + Some(Self { + uuid: hash_string_as_hex(req_url), + file_dir, + file_path, + filename, + url, + finished: false, + size: 0, + error: None, + }) + } + Err(_) => None + } + } +} + +pub struct DownloadQueue { + pub queue: Arc>>, + pub active: Arc>>, + pub finished: Arc>>, +} + + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct FileDownloadRequest { + pub url: String, + pub filename: String, +} + +impl From> for FileDownloadRequest { + fn from(req: web::Json) -> Self { + req.clone() + } +} diff --git a/src/api/model/mod.rs b/src/api/model/mod.rs new file mode 100644 index 000000000..615449ccf --- /dev/null +++ b/src/api/model/mod.rs @@ -0,0 +1,6 @@ +pub mod request; +pub mod config; +pub mod download; +pub mod xtream; +pub mod app_state; +pub mod shared_stream; \ No newline at end of file diff --git a/src/api/model/request.rs b/src/api/model/request.rs new file mode 100644 index 000000000..1b9e1e8d8 --- /dev/null +++ b/src/api/model/request.rs @@ -0,0 +1,44 @@ +use actix_web::web; +use serde::{Deserialize, Serialize}; + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct PlaylistRequest { + pub url: Option, + pub input_id: Option, +} + +impl From> for PlaylistRequest { + fn from(req: web::Json) -> Self { + req.clone() + } +} + +#[derive(Debug, serde::Serialize, serde::Deserialize, Default)] +pub struct UserApiRequest { + #[serde(default)] + pub username: String, + #[serde(default)] + pub password: String, + #[serde(default)] + pub token: String, + #[serde(default)] + pub action: String, + #[serde(default)] + pub series_id: String, + #[serde(default)] + pub vod_id: String, + #[serde(default)] + pub stream_id: String, + #[serde(default)] + pub category_id: String, + #[serde(default)] + pub limit: String, + #[serde(default)] + pub start: String, + #[serde(default)] + pub end: String, + #[serde(default)] + pub stream: String, + #[serde(default)] + pub duration: String, +} \ No newline at end of file diff --git a/src/api/model/shared_stream.rs b/src/api/model/shared_stream.rs new file mode 100644 index 000000000..e2005dbc5 --- /dev/null +++ b/src/api/model/shared_stream.rs @@ -0,0 +1,109 @@ +use futures::stream::{Stream}; +use std::{ + error::Error as StdError, + pin::Pin, + task::{Context, Poll}, +}; +use std::sync::atomic::AtomicU32; +use bytes::Bytes; +use futures::channel::oneshot::{channel, Receiver, Sender}; +use futures::StreamExt; + +pub struct NotifyStream { + stream: S, + tx: Option>, +} + +impl NotifyStream +where + S: Stream> +{ + pub fn new(stream: S) -> (Self, Receiver<()>) { + let (send, recv) = channel(); + (NotifyStream { + stream, + tx: Some(send), + }, recv) + } +} + +impl Stream for NotifyStream where + S: Stream> + Unpin, + E: Into> + 'static, +{ + type Item = S::Item; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let result = self.stream.poll_next_unpin(cx); + let mut connection_closed = false; + match &result { + Poll::Ready(val) => { + if val.is_none() { + connection_closed = true; + } + } + Poll::Pending => {} + } + + if connection_closed { + if let Some(send) = self.tx.take() { + // Ignore errors as they just mean the receiver was dropped. + let _ = send.send(()); + } + } + + result + } + + fn size_hint(&self) -> (usize, Option) { + (0, None) + } +} + +pub struct SharedStream { + pub data_stream: Pin>>>, + pub client_count: AtomicU32, +} + +// +// async fn stream_handler( +// state: web::Data, +// _req: HttpRequest, +// ) -> impl Responder { +// let mut shared_state = state.lock().unwrap(); +// +// // Check if a shared resource already exists +// if let Some(resource) = shared_state.as_mut() { +// // Increment client count +// resource.client_count += 1; +// +// // Create a stream for this client +// let client_stream = resource.data_stream.clone(); +// return HttpResponse::Ok().streaming(client_stream); +// } +// +// // If no shared resource, create one +// let data_stream = crate::api::model::shared_stream::create_data_stream(); // Replace with your actual stream logic +// let boxed_stream = Box::pin(data_stream); +// +// *shared_state = Some(SharedStream { +// data_stream: boxed_stream.clone(), +// client_count: 1, +// }); +// +// // Respond with the stream for this client +// HttpResponse::Ok().streaming(boxed_stream) +// } +// +// +// async fn cleanup_on_disconnect(state: web::Data) { +// let mut shared_state = state.lock().unwrap(); +// if let Some(resource) = shared_state.as_mut() { +// resource.client_count -= 1; +// if resource.client_count == 0 { +// // Drop the shared resource if no clients are left +// *shared_state = None; +// } +// } +// } + diff --git a/src/api/model/xtream.rs b/src/api/model/xtream.rs new file mode 100644 index 000000000..a8addd29d --- /dev/null +++ b/src/api/model/xtream.rs @@ -0,0 +1,73 @@ +use crate::model::api_proxy::{ApiProxyServerInfo, ProxyUserCredentials}; +use serde::Serialize; +use chrono::{Duration, Local}; + +#[derive(Serialize)] +pub struct XtreamUserInfo { + pub active_cons: String, + pub allowed_output_formats: Vec, + //["ts"], + pub auth: u16, + // 0 | 1 + pub created_at: i64, + //1623429679, + pub exp_date: i64, + //1628755200, + pub is_trial: String, + // 0 | 1 + pub max_connections: String, + pub message: String, + pub password: String, + pub username: String, + pub status: String, // "Active" +} + +#[derive(Serialize)] +pub struct XtreamServerInfo { + pub url: String, + pub port: String, + pub https_port: String, + pub server_protocol: String, + // http, https + pub rtmp_port: String, + pub timezone: String, + pub timestamp_now: i64, + pub time_now: String, //"2021-06-28 17:07:37" +} + +#[derive(Serialize)] +pub struct XtreamAuthorizationResponse { + pub user_info: XtreamUserInfo, + pub server_info: XtreamServerInfo, +} + +impl XtreamAuthorizationResponse { + pub fn new(server_info: &ApiProxyServerInfo, user: &ProxyUserCredentials) -> Self { + let now = Local::now(); + Self { + user_info: XtreamUserInfo { + active_cons: "0".to_string(), + allowed_output_formats: Vec::from(["ts".to_string(), "m3u8".to_string(), "rtmp".to_string()]), + auth: 1, + created_at: (now - Duration::days(365)).timestamp(), // fake + exp_date: (now + Duration::days(365)).timestamp(), // fake + is_trial: "0".to_string(), + max_connections: "1".to_string(), + message: server_info.message.to_string(), + password: user.password.to_string(), + username: user.username.to_string(), + status: "Active".to_string(), + }, + server_info: XtreamServerInfo { + url: server_info.host.clone(), + port: server_info.http_port.clone(), + https_port: server_info.https_port.clone(), + server_protocol: server_info.protocol.clone(), + rtmp_port: server_info.rtmp_port.clone(), + timezone: server_info.timezone.to_string(), + timestamp_now: now.timestamp(), + time_now: now.format("%Y-%m-%d %H:%M:%S").to_string(), + }, + } + } +} diff --git a/src/api/scheduler.rs b/src/api/scheduler.rs index f8ac0d126..4ce6d5964 100644 --- a/src/api/scheduler.rs +++ b/src/api/scheduler.rs @@ -4,7 +4,7 @@ use actix_web::web::Data; use chrono::{DateTime, FixedOffset, Local}; use cron::Schedule; use log::error; -use crate::api::api_model::AppState; +use crate::api::model::app_state::AppState; use crate::exit; use crate::processing::playlist_processor::exec_processing; diff --git a/src/api/v1_api.rs b/src/api/v1_api.rs index 18a791b5d..2238e607a 100644 --- a/src/api/v1_api.rs +++ b/src/api/v1_api.rs @@ -1,20 +1,22 @@ use std::sync::Arc; -use actix_web::{HttpResponse, web}; use actix_web::middleware::Condition; +use actix_web::{web, HttpResponse}; use actix_web_httpauth::middleware::HttpAuthentication; use log::error; use serde_json::json; -use crate::api::api_model::{AppState, PlaylistRequest, ServerConfig, ServerInputConfig, ServerSourceConfig, ServerTargetConfig}; use crate::api::download_api; +use crate::api::model::app_state::AppState; +use crate::api::model::config::{ServerConfig, ServerInputConfig, ServerSourceConfig, ServerTargetConfig}; +use crate::api::model::request::PlaylistRequest; use crate::auth::authenticator::validator; use crate::m3u_filter_error::M3uFilterError; use crate::model::api_proxy::{ApiProxyConfig, ApiProxyServerInfo, ProxyUserCredentials, TargetUser}; -use crate::model::config::{Config, ConfigDto, ConfigInput, ConfigInputOptions, ConfigSource, ConfigTarget, InputType, validate_targets}; +use crate::model::config::{validate_targets, Config, ConfigDto, ConfigInput, ConfigInputOptions, ConfigSource, ConfigTarget, InputType}; use crate::processing::playlist_processor; -use crate::utils::{config_reader, download}; use crate::utils::request_utils::mask_sensitive_info; +use crate::utils::{config_reader, download}; fn intern_save_config_api_proxy(backup_dir: &str, api_proxy: &ApiProxyConfig, file_path: &str) -> Option { match config_reader::save_api_proxy(file_path, backup_dir, api_proxy) { @@ -155,7 +157,7 @@ async fn playlist( let url = req.url.as_deref().unwrap_or(""); let input = create_config_input_for_url(url); get_playlist(Some(&input), &app_state.config).await - } + }; } async fn config( diff --git a/src/api/web_index.rs b/src/api/web_index.rs index eac184dbf..392462385 100644 --- a/src/api/web_index.rs +++ b/src/api/web_index.rs @@ -4,8 +4,7 @@ use std::path::{Path, PathBuf}; use actix_files::NamedFile; use actix_web::{HttpRequest, HttpResponse, web}; use actix_web_httpauth::extractors::bearer::BearerAuth; - -use crate::api::api_model::AppState; +use crate::api::model::app_state::AppState; use crate::auth::authenticator::{create_jwt, verify_token}; use crate::auth::password::verify_password; use crate::auth::user::UserCredential; diff --git a/src/api/xmltv_api.rs b/src/api/xmltv_api.rs index 14f1e826b..5b799b6b5 100644 --- a/src/api/xmltv_api.rs +++ b/src/api/xmltv_api.rs @@ -10,8 +10,9 @@ use quick_xml::events::{BytesStart, Event}; use std::io::{BufReader}; use chrono::{Duration, NaiveDateTime, TimeDelta}; -use crate::api::api_model::{AppState, UserApiRequest}; use crate::api::api_utils::{get_user_target, serve_file}; +use crate::api::model::app_state::AppState; +use crate::api::model::request::UserApiRequest; use crate::model::api_proxy::{ProxyUserCredentials}; use crate::model::config::{Config, ConfigTarget}; use crate::model::config::TargetType; diff --git a/src/api/xtream_api.rs b/src/api/xtream_api.rs index cd22626f3..118f35c99 100644 --- a/src/api/xtream_api.rs +++ b/src/api/xtream_api.rs @@ -6,26 +6,28 @@ use std::io::{Error, ErrorKind}; use std::path::Path; use std::str::FromStr; -use actix_web::{HttpRequest, HttpResponse, web}; +use actix_web::{web, HttpRequest, HttpResponse}; use bytes::Bytes; -use futures::Stream; use futures::stream::{self, StreamExt}; +use futures::Stream; use log::{debug, error, log_enabled, Level}; use serde_json::{Map, Value}; -use crate::api::api_model::{AppState, UserApiRequest, XtreamAuthorizationResponse}; use crate::api::api_utils::{get_user_server_info, get_user_target, get_user_target_by_credentials, serve_file, stream_response}; +use crate::api::model::app_state::AppState; +use crate::api::model::request::UserApiRequest; +use crate::api::model::xtream::XtreamAuthorizationResponse; use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; use crate::model::api_proxy::{ProxyType, ProxyUserCredentials}; -use crate::model::config::{Config, ConfigInput, ConfigTarget}; use crate::model::config::TargetType; +use crate::model::config::{Config, ConfigInput, ConfigTarget}; use crate::model::playlist::{PlaylistItemType, XtreamCluster, XtreamPlaylistItem}; use crate::model::xtream::XtreamMappingOptions; use crate::repository::storage::{get_target_storage_path, hash_string}; use crate::repository::target_id_mapping::TargetIdMapping; use crate::repository::xtream_repository; -use crate::utils::{json_utils, request_utils}; use crate::utils::request_utils::mask_sensitive_info; +use crate::utils::{json_utils, request_utils}; const ACTION_GET_SERIES_INFO: &str = "get_series_info"; const ACTION_GET_VOD_INFO: &str = "get_vod_info"; @@ -35,7 +37,7 @@ const ACTION_GET_SHORT_EPG: &str = "get_short_epg"; const ACTION_GET_CATCHUP_TABLE: &str = "get_simple_data_table"; const ACTION_GET_LIVE_CATEGORIES: &str = "get_live_categories"; const ACTION_GET_VOD_CATEGORIES: &str = "get_vod_categories"; -const ACTION_GET_SERIES_CATEGORIES: &str = "get_series_categories"; +const ACTION_GET_SERIES_CATEGORIES: &str = "get_series_categories"; const ACTION_GET_LIVE_STREAMS: &str = "get_live_streams"; const ACTION_GET_VOD_STREAMS: &str = "get_vod_streams"; const ACTION_GET_SERIES: &str = "get_series"; @@ -120,10 +122,10 @@ struct XtreamApiStreamRequest<'a> { impl<'a> XtreamApiStreamRequest<'a> { pub const fn from(context: XtreamApiStreamContext, - username: &'a str, - password: &'a str, - stream_id: &'a str, - action_path: &'a str) -> Self { + username: &'a str, + password: &'a str, + stream_id: &'a str, + action_path: &'a str) -> Self { Self { context, username, @@ -205,7 +207,7 @@ async fn xtream_player_api_stream( } let (action_stream_id, stream_ext) = xtream_api_request_separate_number_and_rest(stream_req.stream_id); let virtual_id: u32 = try_result_bad_request!(action_stream_id.trim().parse()); - let pli = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id)); + let pli = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, None).await, true, format!("Failed to read xtream item for stream id {}", virtual_id)); let input = try_option_bad_request!(app_state.config.get_input_by_id(pli.input_id), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", stream_req.context)); if pli.item_type == PlaylistItemType::LiveHls { @@ -229,11 +231,11 @@ async fn xtream_player_api_stream( return HttpResponse::Found().insert_header(("Location", mask_sensitive_info(pli.url.as_str()))).finish(); } - let stream_url = try_option_bad_request!(get_xtream_player_api_stream_url(input, stream_req.context.to_string().as_str(), &query_path, pli.url.as_str()), true, format!("Cant find stream url for target {target_name}, context {}, stream_id {virtual_id}", stream_req.context)); + let stream_url = try_option_bad_request!(get_xtream_player_api_stream_url(input, stream_req.context.to_string().as_str(), &query_path, pli.url.as_str()), true, format!("Cant find stream url for target {target_name}, context {}, stream_id {virtual_id}", stream_req.context)); if log_enabled!(Level::Debug) { debug!("Streaming stream request from {}", mask_sensitive_info(&stream_url)); } - stream_response(&stream_url, req, Some(input)).await + stream_response(app_state, &stream_url, req, Some(input)).await } @@ -331,7 +333,7 @@ async fn xtream_get_stream_info_content(info_url: &str, input: &ConfigInput) -> async fn xtream_get_stream_info(config: &Config, input: &ConfigInput, target: &ConfigTarget, pli: &XtreamPlaylistItem, info_url: &str, cluster: XtreamCluster) -> Result { if cluster == XtreamCluster::Series { - if let Some(content) = xtream_repository::xtream_load_series_info(config, target.name.as_str(), pli.virtual_id) { + if let Some(content) = xtream_repository::xtream_load_series_info(config, target.name.as_str(), pli.virtual_id).await { return Ok(content); } } @@ -340,12 +342,12 @@ async fn xtream_get_stream_info(config: &Config, input: &ConfigInput, target: &C return match cluster { XtreamCluster::Live => Ok(content), XtreamCluster::Video => get_xtream_vod_info(target, pli, &content), - XtreamCluster::Series => xtream_repository::write_and_get_xtream_series_info(config, target, pli, &content), + XtreamCluster::Series => xtream_repository::write_and_get_xtream_series_info(config, target, pli, &content).await, }; } Err(Error::new(ErrorKind::Other, format!("Cant find stream with id: {}/{}/{}", - target.name.replace(' ', "_").as_str(), &cluster, pli.virtual_id))) + target.name.replace(' ', "_").as_str(), &cluster, pli.virtual_id))) } async fn xtream_get_stream_info_response(app_state: &AppState, user: &ProxyUserCredentials, @@ -356,7 +358,7 @@ async fn xtream_get_stream_info_response(app_state: &AppState, user: &ProxyUserC Err(_) => return HttpResponse::BadRequest().finish() }; - if let Ok(pli) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, Some(cluster)) { + if let Ok(pli) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, Some(cluster)).await { let input_id = pli.input_id; if let Some(input) = app_state.config.get_input_by_id(input_id) { if let Some(info_url) = get_xtream_player_api_info_url(input, cluster, pli.provider_id) { @@ -384,7 +386,7 @@ async fn xtream_get_short_epg(app_state: &AppState, user: &ProxyUserCredentials, Err(_) => return HttpResponse::BadRequest().finish() }; - if let Ok(pli) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, None) { + if let Ok(pli) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, None).await { let input_id: u16 = pli.input_id; if let Some(input) = app_state.config.get_input_by_id(input_id) { if let Some(action_url) = get_xtream_player_api_action_url(input, ACTION_GET_SHORT_EPG) { @@ -434,7 +436,7 @@ async fn xtream_player_api_handle_content_action(config: &Config, target_name: & async fn xtream_get_catchup_response(app_state: &AppState, target: &ConfigTarget, stream_id: &str, start: &str, end: &str) -> HttpResponse { let virtual_id: u32 = try_result_bad_request!(FromStr::from_str(stream_id)); - let pli = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, Some(XtreamCluster::Live))); + let pli = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, Some(XtreamCluster::Live)).await); let input = try_option_bad_request!(app_state.config.get_input_by_id(pli.input_id)); let info_url = try_option_bad_request!(get_xtream_player_api_action_url(input, ACTION_GET_CATCHUP_TABLE).map(|action_url| format!("{action_url}&{TAG_STREAM_ID}={}&start={start}&end={end}", pli.provider_id))); let content = try_result_bad_request!(xtream_get_stream_info_content(info_url.as_str(), input).await); @@ -465,7 +467,7 @@ async fn xtream_player_api( app_state: &web::Data, ) -> HttpResponse { let user_target = get_user_target(&api_req, app_state); - if let Some((user, target)) = user_target { + if let Some((user, target)) = user_target { if !target.has_output(&TargetType::Xtream) { return HttpResponse::Ok().json(get_user_info(&user, &app_state.config)); } @@ -478,22 +480,22 @@ async fn xtream_player_api( match action { ACTION_GET_SERIES_INFO => { return xtream_get_stream_info_response( - app_state, &user, target, api_req.series_id.trim(), XtreamCluster::Series + app_state, &user, target, api_req.series_id.trim(), XtreamCluster::Series, ).await; } ACTION_GET_VOD_INFO => { return xtream_get_stream_info_response( - app_state, &user, target, api_req.vod_id.trim(), XtreamCluster::Video + app_state, &user, target, api_req.vod_id.trim(), XtreamCluster::Video, ).await; } ACTION_GET_EPG | ACTION_GET_SHORT_EPG => { return xtream_get_short_epg( - app_state, &user, target, api_req.stream_id.trim(), api_req.limit.trim() + app_state, &user, target, api_req.stream_id.trim(), api_req.limit.trim(), ).await; } ACTION_GET_CATCHUP_TABLE => { return xtream_get_catchup_response( - app_state, target, api_req.stream_id.trim(), api_req.start.trim(), api_req.end.trim() + app_state, target, api_req.stream_id.trim(), api_req.start.trim(), api_req.end.trim(), ).await; } _ => {} @@ -501,7 +503,7 @@ async fn xtream_player_api( // Handle general content actions if let Some(response) = xtream_player_api_handle_content_action( - &app_state.config, &target.name, action, api_req.category_id.trim(), req + &app_state.config, &target.name, action, api_req.category_id.trim(), req, ).await { return response; } @@ -510,15 +512,15 @@ async fn xtream_player_api( let category_id = api_req.category_id.trim().parse::().unwrap_or(0); let result = match action { ACTION_GET_LIVE_STREAMS => xtream_repository::xtream_load_rewrite_playlist( - XtreamCluster::Live, &app_state.config, target, category_id - ), + XtreamCluster::Live, &app_state.config, target, category_id, + ).await, ACTION_GET_VOD_STREAMS => xtream_repository::xtream_load_rewrite_playlist( - XtreamCluster::Video, &app_state.config, target, category_id - ), + XtreamCluster::Video, &app_state.config, target, category_id, + ).await, ACTION_GET_SERIES => xtream_repository::xtream_load_rewrite_playlist( - XtreamCluster::Series, &app_state.config, target, category_id - ), - _ => Err(M3uFilterError::new(M3uFilterErrorKind::Info,format!("Cant find action: {action} for target: {}", &target.name) + XtreamCluster::Series, &app_state.config, target, category_id, + ).await, + _ => Err(M3uFilterError::new(M3uFilterErrorKind::Info, format!("Cant find action: {action} for target: {}", &target.name), )), }; @@ -547,7 +549,7 @@ async fn xtream_player_api( } } -fn xtream_create_content_stream(xtream_iter: impl Iterator) -> impl Stream> { +fn xtream_create_content_stream(xtream_iter: impl Iterator) -> impl Stream> { let mut first_item = true; stream::iter(xtream_iter.map(move |line| { let line = if first_item { diff --git a/src/auth/authenticator.rs b/src/auth/authenticator.rs index 5ebca65a1..fededceae 100644 --- a/src/auth/authenticator.rs +++ b/src/auth/authenticator.rs @@ -2,8 +2,8 @@ use actix_web::{dev::ServiceRequest, Error, web}; use actix_web_httpauth::extractors::bearer::BearerAuth; use chrono::{Local, Duration}; use jsonwebtoken::{Algorithm, DecodingKey, encode, decode, EncodingKey, Header, Validation}; -use crate::api::api_model::AppState; use crate::model::config::WebAuthConfig; +use crate::api::model::app_state::AppState; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct Claims { diff --git a/src/model/api_proxy.rs b/src/model/api_proxy.rs index 0b8b61357..05cf2b55c 100644 --- a/src/model/api_proxy.rs +++ b/src/model/api_proxy.rs @@ -3,7 +3,7 @@ use std::fmt::Display; use std::str::FromStr; use enum_iterator::Sequence; - +use log::debug; use crate::create_m3u_filter_error_result; use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; use crate::utils::config_reader; @@ -288,6 +288,7 @@ impl ApiProxyConfig { return Some((credentials.clone(), target_name.to_string())); }; } + debug!("Could not find any target for user {username}"); None } diff --git a/src/processing/playlist_processor.rs b/src/processing/playlist_processor.rs index 1d4eed912..8d0495af2 100644 --- a/src/processing/playlist_processor.rs +++ b/src/processing/playlist_processor.rs @@ -4,8 +4,9 @@ use core::cmp::Ordering; use std::cell::RefCell; use std::collections::{HashMap, HashSet}; use std::rc::Rc; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc}; use std::thread; +use async_std::sync::{Mutex}; use actix_rt::System; use log::{debug, error, info, log_enabled, trace, Level}; @@ -391,11 +392,11 @@ async fn process_sources(config: Arc, user_targets: Arc) if process_parallel { let handles = &mut handle_list; let process = move || { - let (mut res_stats, mut res_errors) = System::new().block_on(async { - process_source(cfg, index, usr_trgts).await + System::new().block_on(async { + let (mut res_stats, mut res_errors) = process_source(cfg, index, usr_trgts).await; + shared_errors.lock().await.append(&mut res_errors); + shared_stats.lock().await.append(&mut res_stats); }); - shared_errors.lock().unwrap().append(&mut res_errors); - shared_stats.lock().unwrap().append(&mut res_stats); }; handles.push(thread::spawn(process)); if handles.len() >= thread_num as usize { @@ -403,14 +404,14 @@ async fn process_sources(config: Arc, user_targets: Arc) } } else { let (mut res_stats, mut res_errors) = process_source(cfg, index, usr_trgts).await; - shared_errors.lock().unwrap().append(&mut res_errors); - shared_stats.lock().unwrap().append(&mut res_stats); + shared_errors.lock().await.append(&mut res_errors); + shared_stats.lock().await.append(&mut res_stats); } } for handle in handle_list { let _ = handle.join(); } - (Arc::try_unwrap(stats).unwrap().into_inner().unwrap(), Arc::try_unwrap(errors).unwrap().into_inner().unwrap()) + (Arc::try_unwrap(stats).unwrap().into_inner(), Arc::try_unwrap(errors).unwrap().into_inner()) } pub type ProcessingPipe = Vec Option>>; @@ -521,7 +522,7 @@ async fn process_playlist(playlists: &mut [FetchedPlaylist<'_>], sort_playlist(target, &mut flat_new_playlist); map_playlist_counter(target, &flat_new_playlist); process_watch(target, cfg, &flat_new_playlist); - persist_playlist(&mut flat_new_playlist, flatten_tvguide(&new_epg).as_ref(), target, cfg) + persist_playlist(&mut flat_new_playlist, flatten_tvguide(&new_epg).as_ref(), target, cfg).await } } diff --git a/src/repository/m3u_playlist_iterator.rs b/src/repository/m3u_playlist_iterator.rs index 969cb63e6..e2415fb03 100644 --- a/src/repository/m3u_playlist_iterator.rs +++ b/src/repository/m3u_playlist_iterator.rs @@ -22,7 +22,7 @@ pub struct M3uPlaylistIterator { } impl M3uPlaylistIterator { - pub fn new( + pub async fn new( cfg: &Config, target: &ConfigTarget, user: &ProxyUserCredentials, @@ -30,7 +30,7 @@ impl M3uPlaylistIterator { let target_path = ensure_target_storage_path(cfg, target.name.as_str())?; let (m3u_path, idx_path) = m3u_get_file_paths(&target_path); - let file_lock = cfg.file_locks.read_lock(&m3u_path).map_err(|err| { + let file_lock = cfg.file_locks.read_lock(&m3u_path).await.map_err(|err| { M3uFilterError::new( M3uFilterErrorKind::Info, format!("Could not lock document {m3u_path:?}: {err}"), diff --git a/src/repository/m3u_repository.rs b/src/repository/m3u_repository.rs index ada21ebf5..d6faf4390 100644 --- a/src/repository/m3u_repository.rs +++ b/src/repository/m3u_repository.rs @@ -51,7 +51,7 @@ fn persist_m3u_playlist_as_text(target: &ConfigTarget, cfg: &Config, m3u_playlis } } -pub fn m3u_write_playlist(target: &ConfigTarget, cfg: &Config, target_path: &Path, new_playlist: &[PlaylistGroup]) -> Result<(), M3uFilterError> { +pub async fn m3u_write_playlist(target: &ConfigTarget, cfg: &Config, target_path: &Path, new_playlist: &[PlaylistGroup]) -> Result<(), M3uFilterError> { if !new_playlist.is_empty() { let (m3u_path, idx_path) = m3u_get_file_paths(target_path); let m3u_playlist = new_playlist.iter() @@ -61,7 +61,7 @@ pub fn m3u_write_playlist(target: &ConfigTarget, cfg: &Config, target_path: &Pat persist_m3u_playlist_as_text(target, cfg, &m3u_playlist); { - let _file_lock = cfg.file_locks.write_lock(&m3u_path).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Info, format!("{err}")))?; + let _file_lock = cfg.file_locks.write_lock(&m3u_path).await.map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Info, format!("{err}")))?; match IndexedDocumentWriter::new(m3u_path.clone(), idx_path) { Ok(mut writer) => { for m3u in m3u_playlist { @@ -79,21 +79,21 @@ pub fn m3u_write_playlist(target: &ConfigTarget, cfg: &Config, target_path: &Pat Ok(()) } -pub fn m3u_load_rewrite_playlist( +pub async fn m3u_load_rewrite_playlist( cfg: &Config, target: &ConfigTarget, user: &ProxyUserCredentials, ) -> Result>, M3uFilterError> { - Ok(Box::new(M3uPlaylistIterator::new(cfg, target, user)?)) + Ok(Box::new(M3uPlaylistIterator::new(cfg, target, user).await?)) } -pub fn m3u_get_item_for_stream_id(cfg: &Config, stream_id: u32, m3u_path: &Path, idx_path: &Path) -> Result { +pub async fn m3u_get_item_for_stream_id(cfg: &Config, stream_id: u32, m3u_path: &Path, idx_path: &Path) -> Result { if stream_id < 1 { return Err(Error::new(ErrorKind::Other, "id should start with 1")); } { - let _file_lock = cfg.file_locks.read_lock(m3u_path)?; + let _file_lock = cfg.file_locks.read_lock(m3u_path).await?; IndexedDocumentReader::::read_indexed_item(m3u_path, idx_path, stream_id) } } \ No newline at end of file diff --git a/src/repository/playlist_repository.rs b/src/repository/playlist_repository.rs index 9efae09fc..0d0560fd2 100644 --- a/src/repository/playlist_repository.rs +++ b/src/repository/playlist_repository.rs @@ -10,7 +10,7 @@ use crate::repository::storage::{ensure_target_storage_path, get_target_id_mappi use crate::repository::target_id_mapping::TargetIdMapping; use crate::repository::xtream_repository::xtream_write_playlist; -pub fn persist_playlist(playlist: &mut [PlaylistGroup], epg: Option<&Epg>, +pub async fn persist_playlist(playlist: &mut [PlaylistGroup], epg: Option<&Epg>, target: &ConfigTarget, cfg: &Config) -> Result<(), Vec> { let mut errors = vec![]; let target_path = match ensure_target_storage_path(cfg, &target.name) { @@ -19,7 +19,7 @@ pub fn persist_playlist(playlist: &mut [PlaylistGroup], epg: Option<&Epg>, }; let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = match cfg.file_locks.write_lock(&target_id_mapping_file) { + let _file_lock = match cfg.file_locks.write_lock(&target_id_mapping_file).await { Ok(lock) => lock, Err(err) => { errors.push(M3uFilterError::new(M3uFilterErrorKind::Info, err.to_string())); @@ -45,8 +45,8 @@ pub fn persist_playlist(playlist: &mut [PlaylistGroup], epg: Option<&Epg>, for output in &target.output { let result = match output.target { - TargetType::M3u => m3u_write_playlist(target, cfg, &target_path, playlist), - TargetType::Xtream => xtream_write_playlist(target, cfg, playlist), + TargetType::M3u => m3u_write_playlist(target, cfg, &target_path, playlist).await, + TargetType::Xtream => xtream_write_playlist(target, cfg, playlist).await, TargetType::Strm => kodi_write_strm_playlist(target, cfg, playlist, output.filename.as_ref()), }; diff --git a/src/repository/xtream_playlist_iterator.rs b/src/repository/xtream_playlist_iterator.rs index 67314b2e9..393539e8d 100644 --- a/src/repository/xtream_playlist_iterator.rs +++ b/src/repository/xtream_playlist_iterator.rs @@ -15,7 +15,7 @@ pub struct XtreamPlaylistIterator { } impl XtreamPlaylistIterator { - pub fn new( + pub async fn new( cluster: XtreamCluster, config: &Config, target: &ConfigTarget, @@ -23,7 +23,7 @@ impl XtreamPlaylistIterator { ) -> Result { if let Some(storage_path) = xtream_get_storage_path(config, target.name.as_str()) { let (xtream_path, idx_path) = xtream_get_file_paths(&storage_path, cluster); - let file_lock = config.file_locks.read_lock(&xtream_path).map_err(|err| + let file_lock = config.file_locks.read_lock(&xtream_path).await.map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Info, format!("Could not lock document {xtream_path:?}: {err}")) )?; diff --git a/src/repository/xtream_repository.rs b/src/repository/xtream_repository.rs index 758559ba7..b8f0f1bc2 100644 --- a/src/repository/xtream_repository.rs +++ b/src/repository/xtream_repository.rs @@ -71,11 +71,11 @@ fn xtream_get_info_file_paths(storage_path: &Path, cluster: XtreamCluster) -> Op None } -fn write_playlists_to_file(cfg: &Config, storage_path: &Path, collections: Vec<(XtreamCluster, &mut [PlaylistItem])>) -> Result<(), M3uFilterError> { +async fn write_playlists_to_file(cfg: &Config, storage_path: &Path, collections: Vec<(XtreamCluster, &mut [PlaylistItem])>) -> Result<(), M3uFilterError> { for (cluster, playlist) in collections { let (xtream_path, idx_path) = xtream_get_file_paths(storage_path, cluster); { - let _file_lock = cfg.file_locks.write_lock(&xtream_path).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Info, format!("{err}")))?; + let _file_lock = cfg.file_locks.write_lock(&xtream_path).await.map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Info, format!("{err}")))?; match IndexedDocumentWriter::new(xtream_path.clone(), idx_path) { Ok(mut writer) => { for item in playlist { @@ -153,18 +153,18 @@ pub fn xtream_get_file_paths_for_series(storage_path: &Path) -> (PathBuf, PathBu xtream_get_file_paths_for_name(storage_path, FILE_SERIES) } -fn xtream_garbage_collect(config: &Config, target_name: &str) -> std::io::Result<()> { +async fn xtream_garbage_collect(config: &Config, target_name: &str) -> std::io::Result<()> { // Garbage collect series let storage_path = try_option_ok!(xtream_get_storage_path(config, target_name)); let (info_path, idx_path) = try_option_ok!(xtream_get_info_file_paths(&storage_path, XtreamCluster::Series)); { - let _file_lock = config.file_locks.write_lock(&info_path)?; + let _file_lock = config.file_locks.write_lock(&info_path).await?; IndexedDocumentGarbageCollector::new(info_path, idx_path)?.garbage_collect()?; } Ok(()) } -pub fn xtream_write_playlist(target: &ConfigTarget, cfg: &Config, playlist: &mut [PlaylistGroup]) -> Result<(), M3uFilterError> { +pub async fn xtream_write_playlist(target: &ConfigTarget, cfg: &Config, playlist: &mut [PlaylistGroup]) -> Result<(), M3uFilterError> { let path = ensure_xtream_storage_path(cfg, target.name.as_str())?; let mut errors = Vec::new(); let mut cat_live_col = vec![]; @@ -244,9 +244,9 @@ pub fn xtream_write_playlist(target: &ConfigTarget, cfg: &Config, playlist: &mut match write_playlists_to_file(cfg, &path, vec![ (XtreamCluster::Live, &mut live_col), (XtreamCluster::Video, &mut vod_col), - (XtreamCluster::Series, &mut series_col)]) { + (XtreamCluster::Series, &mut series_col)]).await { Ok(()) => { - if let Err(err) = xtream_garbage_collect(cfg, &target.name) { + if let Err(err) = xtream_garbage_collect(cfg, &target.name).await { if err.kind() != ErrorKind::NotFound { errors.push(format!("Garbage collection failed:{err}")); } @@ -274,18 +274,18 @@ pub fn xtream_get_collection_path(cfg: &Config, target_name: &str, collection_na Err(Error::new(ErrorKind::Other, format!("Cant find collection: {target_name}/{collection_name}"))) } -fn xtream_read_item_for_stream_id(cfg: &Config, stream_id: u32, storage_path: &Path, cluster: XtreamCluster) -> Result { +async fn xtream_read_item_for_stream_id(cfg: &Config, stream_id: u32, storage_path: &Path, cluster: XtreamCluster) -> Result { let (xtream_path, idx_path) = xtream_get_file_paths(storage_path, cluster); { - let _file_lock = cfg.file_locks.read_lock(&xtream_path)?; + let _file_lock = cfg.file_locks.read_lock(&xtream_path).await?; IndexedDocumentReader::::read_indexed_item(&xtream_path, &idx_path, stream_id) } } -fn xtream_read_series_item_for_stream_id(cfg: &Config, stream_id: u32, storage_path: &Path) -> Result { +async fn xtream_read_series_item_for_stream_id(cfg: &Config, stream_id: u32, storage_path: &Path) -> Result { let (xtream_path, idx_path) = xtream_get_file_paths_for_series(storage_path); { - let _file_lock = cfg.file_locks.read_lock(&xtream_path)?; + let _file_lock = cfg.file_locks.read_lock(&xtream_path).await?; IndexedDocumentReader::::read_indexed_item(&xtream_path, &idx_path, stream_id) } } @@ -297,7 +297,7 @@ macro_rules! try_cluster { }; } -pub fn xtream_get_item_for_stream_id( +pub async fn xtream_get_item_for_stream_id( virtual_id: u32, config: &Config, target: &ConfigTarget, @@ -310,7 +310,7 @@ pub fn xtream_get_item_for_stream_id( { let target_id_mapping_file = get_target_id_mapping_file(&target_path); let _file_lock = config.file_locks.read_lock(&target_id_mapping_file) - .map_err(|err| Error::new(ErrorKind::Other, format!("Could not get lock for id mapping for target {} err:{err}", target.name)))?; + .await.map_err(|err| Error::new(ErrorKind::Other, format!("Could not get lock for id mapping for target {} err:{err}", target.name)))?; let mut target_id_mapping = BPlusTreeQuery::::try_new(&target_id_mapping_file) .map_err(|err| Error::new(ErrorKind::Other, format!("Could not load id mapping for target {} err:{err}", target.name)))?; @@ -320,32 +320,32 @@ pub fn xtream_get_item_for_stream_id( .ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not find mapping for target {} and id {}", target.name, virtual_id)))?; match mapping.item_type { - PlaylistItemType::SeriesInfo => xtream_read_series_item_for_stream_id(config, virtual_id, &storage_path), + PlaylistItemType::SeriesInfo => xtream_read_series_item_for_stream_id(config, virtual_id, &storage_path).await, PlaylistItemType::SeriesEpisode => { - let mut item = xtream_read_series_item_for_stream_id(config, mapping.parent_virtual_id, &storage_path)?; + let mut item = xtream_read_series_item_for_stream_id(config, mapping.parent_virtual_id, &storage_path).await?; item.provider_id = mapping.provider_id; Ok(item) } PlaylistItemType::Catchup => { let cluster = try_cluster!(xtream_cluster, mapping.item_type, virtual_id)?; - let mut item = xtream_read_item_for_stream_id(config, mapping.parent_virtual_id, &storage_path, cluster)?; + let mut item = xtream_read_item_for_stream_id(config, mapping.parent_virtual_id, &storage_path, cluster).await?; item.provider_id = mapping.provider_id; Ok(item) } _ => { let cluster = try_cluster!(xtream_cluster, mapping.item_type, virtual_id)?; - xtream_read_item_for_stream_id(config, virtual_id, &storage_path, cluster) + xtream_read_item_for_stream_id(config, virtual_id, &storage_path, cluster).await } } } } -pub fn xtream_load_rewrite_playlist(cluster: XtreamCluster, config: &Config, target: &ConfigTarget, category_id: u32) -> Result>, M3uFilterError> { - Ok(Box::new(XtreamPlaylistIterator::new(cluster, config, target, category_id)?)) +pub async fn xtream_load_rewrite_playlist(cluster: XtreamCluster, config: &Config, target: &ConfigTarget, category_id: u32) -> Result>, M3uFilterError> { + Ok(Box::new(XtreamPlaylistIterator::new(cluster, config, target, category_id).await?)) } -pub fn xtream_write_series_info(config: &Config, target_name: &str, +pub async fn xtream_write_series_info(config: &Config, target_name: &str, series_info_id: u32, content: &str) -> Result<(), Error> { let target_path = try_option_ok!(get_target_storage_path(config, target_name)); @@ -353,7 +353,7 @@ pub fn xtream_write_series_info(config: &Config, target_name: &str, let (info_path, idx_path) = try_option_ok!(xtream_get_info_file_paths(&storage_path, XtreamCluster::Series)); { - let _file_lock = config.file_locks.write_lock(&info_path)?; + let _file_lock = config.file_locks.write_lock(&info_path).await?; let mut writer = IndexedDocumentWriter::new_append(info_path, idx_path)?; writer .write_doc(series_info_id, content) @@ -363,7 +363,7 @@ pub fn xtream_write_series_info(config: &Config, target_name: &str, } { let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = config.file_locks.write_lock(&target_id_mapping_file)?; + let _file_lock = config.file_locks.write_lock(&target_id_mapping_file).await?; if let Ok(mut target_id_mapping) = BPlusTreeUpdate::::try_new(&target_id_mapping_file) { if let Some(record) = target_id_mapping.query(&series_info_id) { let new_record = record.copy_update_timestamp(); @@ -376,13 +376,13 @@ pub fn xtream_write_series_info(config: &Config, target_name: &str, } // Reads the series info entry if exists, otherwise error -pub fn xtream_load_series_info(config: &Config, target_name: &str, series_id: u32) -> Option { +pub async fn xtream_load_series_info(config: &Config, target_name: &str, series_id: u32) -> Option { let target_path = get_target_storage_path(config, target_name)?; let storage_path = xtream_get_storage_path(config, target_name)?; { let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = config.file_locks.read_lock(&target_id_mapping_file).map_err(|err| { + let _file_lock = config.file_locks.read_lock(&target_id_mapping_file).await.map_err(|err| { error!("Could not lock id mapping for target {target_name}: {}", err); Error::new(ErrorKind::Other, format!("ID mapping load error for target {target_name}")) }).ok()?; @@ -403,7 +403,7 @@ pub fn xtream_load_series_info(config: &Config, target_name: &str, series_id: u3 if info_path.exists() && idx_path.exists() { { - let _file_lock = config.file_locks.read_lock(&info_path).map_err(|err| { + let _file_lock = config.file_locks.read_lock(&info_path).await.map_err(|err| { error!("Could not lock document {:?}: {}", info_path, err); Error::new(ErrorKind::Other, format!("Document Reader error for target {target_name}")) }).ok()?; @@ -419,7 +419,7 @@ pub fn xtream_load_series_info(config: &Config, target_name: &str, series_id: u3 None } -pub fn write_and_get_xtream_series_info( +pub async fn write_and_get_xtream_series_info( config: &Config, target: &ConfigTarget, pli_series_info: &XtreamPlaylistItem, @@ -437,7 +437,7 @@ pub fn write_and_get_xtream_series_info( { let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = config.file_locks.write_lock(&target_id_mapping_file) + let _file_lock = config.file_locks.write_lock(&target_id_mapping_file).await .map_err(|err| Error::new(ErrorKind::Other, format!("Could not load id mapping for target {} err:{err}", target.name)))?; let mut target_id_mapping = TargetIdMapping::new(&target_id_mapping_file); let options = XtreamMappingOptions::from_target_options(target.options.as_ref()); @@ -459,7 +459,7 @@ pub fn write_and_get_xtream_series_info( } let result = serde_json::to_string(&doc) .map_err(|_| Error::new(ErrorKind::Other, "Failed to serialize updated series info"))?; - xtream_write_series_info(config, target.name.as_str(), pli_series_info.virtual_id, &result).ok(); + xtream_write_series_info(config, target.name.as_str(), pli_series_info.virtual_id, &result).await.ok(); Ok(result) } \ No newline at end of file diff --git a/src/utils/file_lock_manager.rs b/src/utils/file_lock_manager.rs index 1f63028f2..491d4b831 100644 --- a/src/utils/file_lock_manager.rs +++ b/src/utils/file_lock_manager.rs @@ -1,7 +1,8 @@ use std::collections::HashMap; -use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard}; +use std::sync::{Arc}; use std::{fmt, io}; use std::path::{Path, PathBuf}; +use async_std::sync::{Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard}; #[derive(Clone)] pub struct FileLockManager { @@ -16,30 +17,28 @@ impl FileLockManager { } // Acquires a read lock for the specified file and returns a FileReadGuard. - pub fn read_lock(&self, path: &Path) -> io::Result { - let file_lock = self.get_or_create_lock(path)?; - let guard = file_lock.read().map_err(|_| { - io::Error::new(io::ErrorKind::Other, "Failed to acquire read lock") + pub async fn read_lock(&self, path: &Path) -> io::Result { + let file_lock = self.get_or_create_lock(path).await.map_err(|_| { + io::Error::new(io::ErrorKind::Other, "Failed to acquire write lock") })?; + let guard = file_lock.read().await; // Clone the Arc to avoid moving `file_lock` out, as it is still borrowed by `guard` Ok(FileReadGuard::new(Arc::clone(&file_lock), guard)) } // Acquires a write lock for the specified file and returns a FileWriteGuard. - pub fn write_lock(&self, path: &Path) -> io::Result { - let file_lock = self.get_or_create_lock(path)?; - let guard = file_lock.write().map_err(|_| { + pub async fn write_lock(&self, path: &Path) -> io::Result { + let file_lock = self.get_or_create_lock(path).await.map_err(|_| { io::Error::new(io::ErrorKind::Other, "Failed to acquire write lock") })?; + let guard = file_lock.write().await; // Clone the Arc to avoid moving `file_lock` out, as it is still borrowed by `guard` Ok(FileWriteGuard::new(Arc::clone(&file_lock), guard)) } // Helper function: retrieves or creates a lock for a file. - fn get_or_create_lock(&self, path: &Path) -> io::Result>> { - let mut locks = self.locks.lock().map_err(|_| { - io::Error::new(io::ErrorKind::Other, "Failed to acquire lock on lock manager") - })?; + async fn get_or_create_lock(&self, path: &Path) -> io::Result>> { + let mut locks = self.locks.lock().await; if let Some(lock) = locks.get(path) { return Ok(lock.clone()); @@ -61,7 +60,7 @@ impl Default for FileLockManager { impl fmt::Debug for FileLockManager { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("FileLockManager") - .field("locks", &self.locks.lock().unwrap().keys().collect::>()) + // .field("locks", &self.locks.lock().await.keys().collect::>()) .finish() } }