diff --git a/Cargo.lock b/Cargo.lock index 1c6d02e1c..9b2071734 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -368,18 +368,6 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" -[[package]] -name = "async-broadcast" -version = "0.7.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "435a87a52755b8f27fcf321ac4f04b2802e337c8c4872923137471ec39c37532" -dependencies = [ - "event-listener 5.4.0", - "event-listener-strategy", - "futures-core", - "pin-project-lite", -] - [[package]] name = "async-channel" version = "1.9.0" @@ -692,9 +680,9 @@ dependencies = [ [[package]] name = "clap" -version = "4.5.27" +version = "4.5.28" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "769b0145982b4b48713e01ec42d61614425f27b7058bda7180a3a41f30104796" +checksum = "3e77c3243bd94243c03672cb5154667347c457ca271254724f9f393aee1c05ff" dependencies = [ "clap_builder", "clap_derive", @@ -714,9 +702,9 @@ dependencies = [ [[package]] name = "clap_derive" -version = "4.5.24" +version = "4.5.28" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "54b755194d6389280185988721fffba69495eed5ee9feeee9a599b53db80318c" +checksum = "bf4ced95c6f4a675af3da73304b9ac4ed991640c36374e4b46795c49e17cf1ed" dependencies = [ "heck", "proc-macro2", @@ -1013,6 +1001,18 @@ version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" +[[package]] +name = "filetime" +version = "0.2.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35c0522e981e68cbfa8c3f978441a5f34b30b96e146b33cd3359176b50fe8586" +dependencies = [ + "cfg-if", + "libc", + "libredox", + "windows-sys 0.59.0", +] + [[package]] name = "flate2" version = "1.0.35" @@ -1720,6 +1720,17 @@ dependencies = [ "libc", ] +[[package]] +name = "libredox" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0ff37bd590ca25063e35af745c343cb7a0271906fb7b37e4813e8f79f00268d" +dependencies = [ + "bitflags 2.8.0", + "libc", + "redox_syscall", +] + [[package]] name = "libz-sys" version = "1.1.21" @@ -1790,7 +1801,6 @@ dependencies = [ "actix-server", "actix-web", "actix-web-httpauth", - "async-broadcast", "async-std", "bincode", "blake3", @@ -1800,6 +1810,7 @@ dependencies = [ "cron", "enum-iterator", "env_logger", + "filetime", "flate2", "futures", "jsonwebtoken", diff --git a/Cargo.toml b/Cargo.toml index 563e65be2..86d67d2be 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,7 +56,7 @@ tokio = "1.43" paste = "1.0" tempfile = "3.15" ruzstd = "0" -async-broadcast = "0" +filetime = "0.2" #[cfg(target_os = "macos")] libc = "0" #[cfg(target_os = "windows")] diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs index 4475becd8..6d08cfdfc 100644 --- a/src/api/api_utils.rs +++ b/src/api/api_utils.rs @@ -5,7 +5,6 @@ use crate::api::model::provider_stream; use crate::api::model::provider_stream::get_provider_pipe_stream; use crate::api::model::provider_stream_factory::BufferStreamOptions; use crate::api::model::request::UserApiRequest; -use crate::api::model::shared_stream::SharedStream; use crate::api::model::stream_error::StreamError; use crate::{debug_if_enabled, trace_if_enabled}; use crate::model::api_proxy::ProxyUserCredentials; @@ -20,8 +19,6 @@ use actix_web::body::{BodyStream, SizedStream}; use actix_web::http::header::{HeaderValue, CACHE_CONTROL}; use actix_web::{HttpRequest, HttpResponse}; use async_std::sync::Mutex; -use bytes::Bytes; -use futures::stream::BoxStream; use futures::{TryStreamExt}; use log::{error, log_enabled, trace}; use reqwest::StatusCode; @@ -30,6 +27,7 @@ use std::path::Path; use std::sync::Arc; use url::Url; use crate::api::model::active_client_stream::ActiveClientStream; +use crate::api::model::shared_stream_manager::SharedStreamManager; #[macro_export] macro_rules! try_option_bad_request { @@ -102,21 +100,6 @@ pub fn get_user_target<'a>(api_req: &'a UserApiRequest, app_state: &'a AppState) get_user_target_by_credentials(username, password, api_req, app_state) } -/// Creates a broadcast notify stream for the given URL if a shared stream exists. -async fn create_broadcast_stream( - app_state: &AppState, - stream_url: &str, -) -> Option>> { - let notify_stream_url = stream_url.to_string(); - // Acquire lock and check for existing stream - let shared_streams = app_state.shared_streams.lock().await; - if let Some((_, shared_stream)) = shared_streams.get(¬ify_stream_url) { - Some(shared_stream.get_receiver()) - } else { - None - } -} - fn get_stream_options(app_state: &AppState) -> (bool, bool, usize, bool, bool) { let (stream_retry, buffer_enabled, buffer_size) = app_state .config @@ -173,8 +156,8 @@ pub async fn stream_response(app_state: &AppState, stream_url: &str, let stream = ActiveClientStream::new(stream, active_clients, log_active_clients); let stream_resp = if share_stream { let shared_headers = provider_response.as_ref().map_or_else(Vec::new, |(h, _)| h.clone()); - SharedStream::register(app_state, stream_url, stream, shared_stream_use_own_buffer, shared_headers).await; - if let Some(broadcast_stream) = create_broadcast_stream(app_state, stream_url).await { + SharedStreamManager::subscribe(app_state, stream_url, stream, shared_stream_use_own_buffer, shared_headers).await; + if let Some(broadcast_stream) = SharedStreamManager::subscribe_shared_stream(app_state, stream_url).await { let mut response_builder = get_stream_response_with_headers(provider_response, stream_url); if content_length > 0 { response_builder.body(SizedStream::new(content_length, broadcast_stream)) } else { response_builder.body(BodyStream::new(broadcast_stream)) } } else { @@ -193,9 +176,9 @@ pub async fn stream_response(app_state: &AppState, stream_url: &str, } async fn shared_stream_response(app_state: &AppState, stream_url: &str) -> Option { - if let Some(stream) = create_broadcast_stream(app_state, stream_url).await { + if let Some(stream) = SharedStreamManager::subscribe_shared_stream(app_state, stream_url).await { debug_if_enabled!("Using shared channel {}", sanitize_sensitive_info(stream_url)); - if let Some((headers, _)) = app_state.shared_streams.lock().await.get(stream_url) { + if let Some(headers) = app_state.shared_stream_manager.lock().await.get_shared_state_headers(stream_url).await { let mut response_builder = get_stream_response_with_headers(Some((headers.clone(), StatusCode::OK)), stream_url); return Some(response_builder.body(BodyStream::new(stream))); } @@ -204,7 +187,7 @@ async fn shared_stream_response(app_state: &AppState, stream_url: &str) -> Optio } pub fn is_stream_share_enabled(item_type: PlaylistItemType, target: &ConfigTarget) -> bool { - item_type == PlaylistItemType::Live && target.options.as_ref().is_some_and(|opt| opt.share_live_streams) + (item_type == PlaylistItemType::Live || item_type == PlaylistItemType::LiveHls) && target.options.as_ref().is_some_and(|opt| opt.share_live_streams) } pub type HeaderFilter = Option bool>>; diff --git a/src/api/main_api.rs b/src/api/main_api.rs index 38e470df9..e893ec346 100644 --- a/src/api/main_api.rs +++ b/src/api/main_api.rs @@ -4,7 +4,7 @@ use actix_web::web::Data; use actix_web::{web, App, HttpResponse, HttpServer}; use async_std::sync::{Mutex, RwLock}; use log::{error, info}; -use std::collections::{HashMap, VecDeque}; +use std::collections::{VecDeque}; use std::io::ErrorKind; use std::path::{PathBuf}; use std::sync::Arc; @@ -13,6 +13,7 @@ use crate::api::hls_api::hls_api_register; use crate::api::m3u_api::m3u_api_register; use crate::api::model::app_state::AppState; use crate::api::model::download::DownloadQueue; +use crate::api::model::shared_stream_manager::SharedStreamManager; use crate::api::scheduler::start_scheduler; use crate::api::v1_api::v1_api_register; use crate::api::web_index::index_register; @@ -69,7 +70,7 @@ fn create_shared_data(cfg: &Arc) -> Data { finished: Arc::from(RwLock::new(Vec::new())), }), active_clients: Arc::new(AtomicUsize::new(0)), - shared_streams: Arc::new(Mutex::new(HashMap::new())), + shared_stream_manager: Arc::new(Mutex::new(SharedStreamManager::new())), http_client: Arc::new(reqwest::Client::new()), cache, }) diff --git a/src/api/model/app_state.rs b/src/api/model/app_state.rs index e2d4e9660..b80570e25 100644 --- a/src/api/model/app_state.rs +++ b/src/api/model/app_state.rs @@ -1,18 +1,15 @@ -use std::collections::HashMap; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use async_std::sync::{Mutex}; use crate::api::model::download::DownloadQueue; -use crate::api::model::shared_stream::SharedStream; +use crate::api::model::shared_stream_manager::SharedStreamManager; use crate::model::config::{Config}; use crate::utils::lru_cache::LRUResourceCache; -type SharedStreamState = (Vec<(String, String)>, SharedStream); - pub struct AppState { pub config: Arc, pub downloads: Arc, - pub shared_streams: Arc>>, + pub shared_stream_manager: Arc>, pub active_clients: Arc, pub http_client: Arc, pub cache: Arc>> diff --git a/src/api/model/broadcast_stream.rs b/src/api/model/broadcast_stream.rs deleted file mode 100644 index 70bcc65fe..000000000 --- a/src/api/model/broadcast_stream.rs +++ /dev/null @@ -1,43 +0,0 @@ -use std::pin::Pin; -use std::task::{Context, Poll}; -use async_broadcast::{Receiver}; -use bytes::Bytes; -use futures::{Stream}; -use log::error; -use crate::api::model::stream_error::StreamError; - -#[derive(Debug)] -pub struct BroadcastStream { - inner: Receiver -} - -impl BroadcastStream { - pub fn new(mut recv: Receiver) -> Self { - recv.set_overflow(true); - Self { inner: recv } - } -} - -impl Stream for BroadcastStream { - type Item = Result; - - fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - if self.inner.is_closed() { - return Poll::Ready(None); - } - match Pin::new(&mut self.inner).poll_recv(cx) { - Poll::Pending => Poll::Pending, - Poll::Ready(Some(item)) => { - match item { - Ok(data) => Poll::Ready(Some(Ok(data))), - Err(err) => { - error!("Shared stream client error {err}"); - Poll::Pending - } - } - } - Poll::Ready(None) => Poll::Ready(Some(Err(StreamError::ReceiverClosed))), - - } - } -} diff --git a/src/api/model/mod.rs b/src/api/model/mod.rs index a67fe05c9..b136434ac 100644 --- a/src/api/model/mod.rs +++ b/src/api/model/mod.rs @@ -3,13 +3,12 @@ pub mod config; pub mod download; pub mod xtream; pub mod app_state; -pub mod shared_stream; pub mod provider_stream; pub mod persist_pipe_stream; pub mod provider_stream_factory; mod buffered_stream; +pub mod shared_stream_manager; pub mod model_utils; mod client_stream; pub mod stream_error; -mod broadcast_stream; -pub mod active_client_stream; +pub mod active_client_stream; \ No newline at end of file diff --git a/src/api/model/model_utils.rs b/src/api/model/model_utils.rs index 0e3b6807b..b505975b1 100644 --- a/src/api/model/model_utils.rs +++ b/src/api/model/model_utils.rs @@ -41,7 +41,7 @@ pub fn get_stream_response_with_headers(custom: Option<(Vec<(String, String)>, S headers.push((actix_web::http::header::DATE, HeaderValue::from_str(&chrono::Utc::now().to_rfc2822()).unwrap())); let mut response_builder = actix_web::HttpResponse::build(actix_web::http::StatusCode::from_u16(status).unwrap()); - debug_if_enabled!("Responding stream {} with status {status}, headers {headers:?}", sanitize_sensitive_info(stream_url)); + debug_if_enabled!("Responding stream request {} with status {status}, headers {headers:?}", sanitize_sensitive_info(stream_url)); for header in headers { response_builder.insert_header(header); } diff --git a/src/api/model/shared_stream.rs b/src/api/model/shared_stream.rs deleted file mode 100644 index 28814209b..000000000 --- a/src/api/model/shared_stream.rs +++ /dev/null @@ -1,76 +0,0 @@ -use crate::api::model::app_state::AppState; -use crate::api::model::provider_stream_factory::STREAM_QUEUE_SIZE; -use crate::debug_if_enabled; -use crate::utils::request_utils::sanitize_sensitive_info; -use bytes::Bytes; -use std::sync::Arc; -use std::time::Duration; -use futures::stream::BoxStream; -use futures::{Stream, StreamExt}; -use log::error; -use crate::api::model::broadcast_stream::BroadcastStream; -use crate::api::model::stream_error::StreamError; - -const MIN_STREAM_QUEUE_SIZE: usize = 1024; - -pub struct SharedStream { - pub receiver: async_broadcast::Receiver, -} - -impl SharedStream { - - pub(crate)fn get_receiver(&self) -> BoxStream<'static, Result> { - BroadcastStream::new(self.receiver.clone()).boxed() - } - - /// Creates a shared stream and stores it in the shared state. - pub(crate) async fn register( - app_state: &AppState, - stream_url: &str, - bytes_stream: S, - use_buffer: bool, - headers: Vec<(String, String)>, - ) where - S: Stream> + Unpin + 'static, - { - // Create a broadcast channel for the shared stream - let (tx, rx) = async_broadcast::broadcast(if use_buffer { STREAM_QUEUE_SIZE } else { MIN_STREAM_QUEUE_SIZE }); - let sender = tx.clone(); - - // Insert the shared stream into the shared state - app_state - .shared_streams - .lock() - .await - .insert( - stream_url.to_string(), - (headers, SharedStream {receiver: rx, }), - ); - - let shared_streams_map = Arc::clone(&app_state.shared_streams); - let mut source_stream = Box::pin(bytes_stream); - let streaming_url = stream_url.to_string(); - // Spawn a task to forward items from the source stream to the broadcast channel - actix_rt::spawn(async move { - while let Some(item) = source_stream.next().await { - if let Ok(data) = item { - if sender.receiver_count() > 0 { - match sender.broadcast(data).await { - Ok(_) => {} - Err(err) => { - error!("Broadcast channel send {err}"); - } - } - actix_web::rt::time::sleep(Duration::from_millis(20)).await; - } else { - debug_if_enabled!("No active subscribers. Closing stream {}", sanitize_sensitive_info(&streaming_url)); - // Cleanup for removing unused shared streams - let mut shared_streams = shared_streams_map.lock().await; - shared_streams.remove(&streaming_url); - return; - } - } - } - }); - } -} diff --git a/src/api/model/shared_stream_manager.rs b/src/api/model/shared_stream_manager.rs new file mode 100644 index 000000000..ae9194528 --- /dev/null +++ b/src/api/model/shared_stream_manager.rs @@ -0,0 +1,165 @@ +use crate::api::model::app_state::AppState; +use crate::api::model::provider_stream_factory::STREAM_QUEUE_SIZE; +use crate::api::model::stream_error::StreamError; +use crate::debug_if_enabled; +use crate::utils::request_utils::sanitize_sensitive_info; +use async_std::sync::Mutex; +use bytes::Bytes; +use futures::stream::BoxStream; +use futures::{Stream, StreamExt}; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::mpsc; +use tokio::sync::mpsc::error::TrySendError; +use tokio::sync::mpsc::{Sender}; +use tokio_stream::wrappers::ReceiverStream; + +use std::pin::Pin; +use std::task::{Context, Poll}; + +const MIN_STREAM_QUEUE_SIZE: usize = 1024; + +/// +/// Wraps a `ReceiverStream` as Stream> +/// +struct ReceiverStreamWrapper { + stream: S, +} + +impl Stream for ReceiverStreamWrapper +where + S: Stream + Unpin, +{ + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match Pin::new(&mut self.stream).poll_next(cx) { + Poll::Ready(Some(bytes)) => Poll::Ready(Some(Ok(bytes))), + Poll::Ready(None) => Poll::Ready(None), + Poll::Pending => Poll::Pending, + } + } +} + +fn convert_stream(stream: BoxStream) -> BoxStream> { + Box::pin(ReceiverStreamWrapper { stream }.boxed()) +} + +/// Represents the state of a shared provider URL. +/// +/// - `headers`: The initial connection headers used during the setup of the shared stream. +/// - `subscribers`: A list of clients that have subscribed to the shared stream. +struct SharedStreamState { + headers: Vec<(String, String)>, + buf_size: usize, + subscribers: Arc>>>, +} + +impl SharedStreamState { + fn new(headers: Vec<(String, String)>, + buf_size: usize) -> Self { + Self { + headers, + buf_size, + subscribers: Arc::new(Mutex::new(Vec::new())), + } + } + + async fn subscribe(&self) -> BoxStream<'static, Result> { + let (tx, rx) = mpsc::channel(self.buf_size); + self.subscribers.lock().await.push(tx); + convert_stream(ReceiverStream::new(rx).boxed()) + } + + fn broadcast(&self, stream_url: &str, bytes_stream: S, shared_streams: Arc>) + where + S: Stream> + Unpin + 'static, + { + let mut source_stream = Box::pin(bytes_stream); + let subscriber = Arc::clone(&self.subscribers); + let streaming_url = stream_url.to_string(); + // Spawn a task to forward items from the source stream to the broadcast channel + actix_rt::spawn(async move { + while let Some(item) = source_stream.next().await { + if let Ok(data) = item { + let mut subs = subscriber.lock().await; + if subs.len() > 0 { + (*subs).retain(|sender| { + match sender.try_send(data.clone()) { + Err(TrySendError::Closed(_)) => false, + Ok(()) | Err(_) => true, + // Err(TrySendError::Full(_)) => false, // Drop slow consumers + } + }); + } else { + debug_if_enabled!("No active subscribers. Closing shared provider stream {}", sanitize_sensitive_info(&streaming_url)); + // Cleanup for removing unused shared streams + shared_streams.lock().await.unregister(&streaming_url).await; + return; + } + } + } + }); + } +} + +type SharedStreamRegister = Arc>>; + +pub struct SharedStreamManager { + shared_streams: SharedStreamRegister, +} + +impl SharedStreamManager { + pub(crate) fn new() -> Self { + Self { + shared_streams: Arc::new(Mutex::new(HashMap::new())), + } + } + + pub async fn get_shared_state_headers(&self, stream_url: &str) -> Option> { + self.shared_streams.lock().await.get(stream_url).map(|s| s.headers.clone()) + } + + async fn unregister(&self, stream_url: &str) { + self.shared_streams.lock().await.remove(stream_url); + } + + pub(crate) async fn subscribe( + app_state: &AppState, + stream_url: &str, + bytes_stream: S, + use_buffer: bool, + headers: Vec<(String, String)>, ) + where + S: Stream> + Unpin + 'static, + { + let buf_size = if use_buffer { STREAM_QUEUE_SIZE } else { MIN_STREAM_QUEUE_SIZE }; + let shared_state = SharedStreamState::new(headers, buf_size); + shared_state.broadcast(stream_url, bytes_stream, Arc::clone(&app_state.shared_stream_manager)); + app_state + .shared_stream_manager + .lock().await + .shared_streams + .lock().await + .insert(stream_url.to_string(), shared_state); + debug_if_enabled!("Created shared provider stream {}", sanitize_sensitive_info(stream_url)); + } + + /// Creates a broadcast notify stream for the given URL if a shared stream exists. + pub async fn subscribe_shared_stream( + app_state: &AppState, + stream_url: &str, + ) -> Option>> { + if let Some(shared_stream) = app_state + .shared_stream_manager + .lock().await + .shared_streams + .lock().await + .get(stream_url) { + debug_if_enabled!("Responding existing shared client stream {}", sanitize_sensitive_info(stream_url)); + Some(shared_stream.subscribe().await) + } else { + None + } + } +} \ No newline at end of file diff --git a/src/api/model/stream_error.rs b/src/api/model/stream_error.rs index 95a163856..d170da183 100644 --- a/src/api/model/stream_error.rs +++ b/src/api/model/stream_error.rs @@ -3,7 +3,7 @@ pub enum StreamError { Reqwest(String), // StdIo(std::io::Error), - ReceiverClosed, + // ReceiverClosed, // ReceiverError(RecvError), } @@ -23,7 +23,7 @@ impl std::fmt::Display for StreamError { match self { StreamError::Reqwest(e) => write!(f, "Reqwest error: {e}"), // StreamError::StdIo(e) => write!(f, "IO error: {e}"), - StreamError::ReceiverClosed => write!(f, "Receiver closed"), + // StreamError::ReceiverClosed => write!(f, "Receiver closed"), // StreamError::ReceiverError(e) => write!(f, "Receiver error {e}"), } } diff --git a/src/api/xtream_api.rs b/src/api/xtream_api.rs index 43cddb2bf..7d646af05 100644 --- a/src/api/xtream_api.rs +++ b/src/api/xtream_api.rs @@ -538,7 +538,7 @@ async fn xtream_get_catchup_response(app_state: &AppState, target: &ConfigTarget for epg_list_item in epg_listings.iter_mut().filter_map(Value::as_object_mut) { // TODO epg_id if let Some(catchup_provider_id) = epg_list_item.get(TAG_ID).and_then(Value::as_str).and_then(|id| id.parse::().ok()) { - let uuid = generate_playlist_uuid(&hex_encode(&pli.get_uuid()), &catchup_provider_id.to_string(), &pli.url); + let uuid = generate_playlist_uuid(&hex_encode(&pli.get_uuid()), &catchup_provider_id.to_string(), pli.item_type, &pli.url); let virtual_id = target_id_mapping.get_and_update_virtual_id(uuid, catchup_provider_id, PlaylistItemType::Catchup, pli.provider_id); epg_list_item.insert(TAG_ID.to_string(), Value::String(virtual_id.to_string())); } diff --git a/src/model/playlist.rs b/src/model/playlist.rs index c085b475f..21db2f7e9 100644 --- a/src/model/playlist.rs +++ b/src/model/playlist.rs @@ -165,7 +165,7 @@ pub struct PlaylistItemHeader { impl PlaylistItemHeader { pub fn gen_uuid(&mut self) { - self.uuid = Rc::new(generate_playlist_uuid(&self.input_name, &self.id, &self.url)); + self.uuid = Rc::new(generate_playlist_uuid(&self.input_name, &self.id, self.item_type, &self.url)); } pub const fn get_uuid(&self) -> &Rc { &self.uuid @@ -338,7 +338,7 @@ impl PlaylistEntry for M3uPlaylistItem { } fn get_uuid(&self) -> UUIDType { - generate_playlist_uuid(&self.input_name, &self.provider_id, &self.url) + generate_playlist_uuid(&self.input_name, &self.provider_id, self.item_type, &self.url) } } @@ -416,7 +416,7 @@ impl PlaylistEntry for XtreamPlaylistItem { #[inline] fn get_uuid(&self) -> UUIDType { - generate_playlist_uuid(&self.input_name, &self.provider_id.to_string(), &self.url) + generate_playlist_uuid(&self.input_name, &self.provider_id.to_string(), self.item_type, &self.url) } } @@ -589,7 +589,7 @@ impl PlaylistEntry for PlaylistItem { #[inline] fn get_uuid(&self) -> UUIDType { let header = self.header.borrow(); - generate_playlist_uuid(&header.input_name, &header.id, &header.url) + generate_playlist_uuid(&header.input_name, &header.id, header.item_type, &header.url) } } diff --git a/src/processing/xtream_parser.rs b/src/processing/xtream_parser.rs index 675284e79..cccb52d9a 100644 --- a/src/processing/xtream_parser.rs +++ b/src/processing/xtream_parser.rs @@ -54,7 +54,7 @@ pub fn parse_xtream_series_info(info: &Value, group_title: &str, series_name: &s PlaylistItem { header: RefCell::new(PlaylistItemHeader { id: Rc::new(episode.id.to_string()), - uuid: Rc::new(generate_playlist_uuid(&input.name, &episode.id, &episode_url)), + uuid: Rc::new(generate_playlist_uuid(&input.name, &episode.id, PlaylistItemType::Series, &episode_url)), name: Rc::new(series_name.to_string()), logo: Rc::new(episode.info.as_ref().map_or_else(String::new, |info| info.movie_image.to_string())), group: Rc::new(group_title.to_string()), @@ -132,17 +132,18 @@ pub fn parse_xtream(input: &ConfigInput, let mut grp = group.borrow_mut(); let category_name = &grp.category_name; let stream_url = create_xtream_url(xtream_cluster, url, username, password, &stream, live_stream_use_prefix, live_stream_without_extension); + let item_type = PlaylistItemType::from(xtream_cluster); let item = PlaylistItem { header: RefCell::new(PlaylistItemHeader { id: Rc::new(stream.get_stream_id().to_string()), - uuid: Rc::new(generate_playlist_uuid(&input_name, &stream.get_stream_id().to_string(), &stream_url)), + uuid: Rc::new(generate_playlist_uuid(&input_name, &stream.get_stream_id().to_string(), item_type, &stream_url)), name: Rc::clone(&stream.name), logo: Rc::clone(&stream.stream_icon), group: Rc::clone(category_name), title: Rc::clone(&stream.name), url: stream_url, epg_channel_id: stream.epg_channel_id.clone(), - item_type: PlaylistItemType::from(xtream_cluster), + item_type, xtream_cluster, additional_properties: stream.get_additional_properties(), category_id: 0, diff --git a/src/repository/indexed_document.rs b/src/repository/indexed_document.rs index a2f4d1645..b7906023e 100644 --- a/src/repository/indexed_document.rs +++ b/src/repository/indexed_document.rs @@ -585,8 +585,8 @@ mod tests { #[test] fn test_read_xt() -> io::Result<()> { - let main_path = PathBuf::from("/tmp/live.db"); - let index_path = PathBuf::from("/tmp/live.idx"); + let main_path = PathBuf::from("../m3u-test/settings/m3u-silver/data/xt_m3u/xtream/live.db"); + let index_path = PathBuf::from("../m3u-test/settings/m3u-silver/data/xt_m3u/xtream/live.idx"); let reader = IndexedDocumentIterator::::new(&main_path, &index_path)?; for doc in reader { println!("{doc:?}"); diff --git a/src/repository/kodi_repository.rs b/src/repository/kodi_repository.rs index 655613406..9de05fd59 100644 --- a/src/repository/kodi_repository.rs +++ b/src/repository/kodi_repository.rs @@ -21,6 +21,7 @@ use serde::Serialize; use std::collections::{HashMap, HashSet}; use std::rc::Rc; use std::sync::LazyLock; +use filetime::{set_file_times, FileTime}; const FILE_STRM: &str = "strm"; @@ -280,6 +281,13 @@ struct StrmItemInfo { release_date: Option, season: Option, episode: Option, + added: Option, +} + +impl StrmItemInfo { + pub(crate) fn get_file_ts(&self) -> Option { + self.added + } } fn extract_item_info(pli: &PlaylistItem) -> StrmItemInfo { @@ -291,19 +299,28 @@ fn extract_item_info(pli: &PlaylistItem) -> StrmItemInfo { let virtual_id = header.virtual_id; let input_name = Rc::clone(&header.input_name); let url = Rc::clone(&header.url); - let (series_name, release_date, season, episode) = if header.item_type == PlaylistItemType::Series { - let series_name = match header.get_field("name") { - Some(name) if !name.is_empty() => Some(name.to_string()), - _ => header.get_additional_property_as_str("series_name"), - }; - let release_date = header.get_additional_property_as_str("series_release_date") - .or_else(|| header.get_additional_property_as_str("release_date")); - let season = header.get_additional_property_as_str("season"); - let episode = header.get_additional_property_as_str("episode"); - (series_name, release_date, season, episode) - } else { (None, None, None, None) }; - - StrmItemInfo { group, title, item_type, provider_id, virtual_id, input_name, url, series_name, release_date, season, episode } + let (series_name, release_date, added, season, episode) = match header.item_type { + PlaylistItemType::Series => { + let series_name = match header.get_field("name") { + Some(name) if !name.is_empty() => Some(name.to_string()), + _ => header.get_additional_property_as_str("series_name"), + }; + let release_date = header.get_additional_property_as_str("series_release_date") + .or_else(|| header.get_additional_property_as_str("release_date")); + let season = header.get_additional_property_as_str("season"); + let episode = header.get_additional_property_as_str("episode"); + let added = header.get_additional_property_as_u64("added"); + (series_name, release_date, added, season, episode) + }, + PlaylistItemType::Video => { + let name = header.get_field("name").map(|v| v.to_string()); + let release_date = header.get_additional_property_as_str("release_date"); + let added = header.get_additional_property_as_u64("added"); + (name, release_date, added, None, None) + }, + _ => { (None, None, None, None, None) } + }; + StrmItemInfo { group, title, item_type, provider_id, virtual_id, input_name, url, series_name, release_date, season, episode, added } } async fn prepare_strm_output_directory(path: &Path) -> Result<(), M3uFilterError> { @@ -490,7 +507,7 @@ pub async fn kodi_write_strm_playlist(target: &ConfigTarget, cfg: &Config, new_p // if we cant create the directory skip this entry if !ensure_strm_file_directory(&mut failed, &output_path).await { continue; } - match write_strm_file(&file_path, content_as_bytes).await { + match write_strm_file(&file_path, content_as_bytes, str_item_info.get_file_ts()).await { Ok(()) => { processed_strm.insert(relative_file_path); } Err(err) => { failed.push(err); } }; @@ -537,12 +554,20 @@ async fn ensure_strm_file_directory(failed: &mut Vec, output_path: &Path true } -async fn write_strm_file(file_path: &Path, content_as_bytes: &[u8]) -> Result<(), String> { +async fn write_strm_file(file_path: &Path, content_as_bytes: &[u8], timestamp: Option) -> Result<(), String> { File::create(file_path).await .map_err(|err| format!("failed to create strm file: {err}"))? .write_all(content_as_bytes).await .map_err(|err| format!("failed to write strm playlist: {err}"))?; + if let Some(ts) = timestamp { + #[allow(clippy::cast_possible_wrap)] + let mtime = FileTime::from_unix_time(ts as i64, 0); // Unix-Timestamp: 01.01.2023 00:00:00 UTC + #[allow(clippy::cast_possible_wrap)] + let atime = FileTime::from_unix_time(ts as i64, 0); // access time + let _ = set_file_times(file_path, mtime, atime); + } + Ok(()) } diff --git a/src/repository/target_id_mapping.rs b/src/repository/target_id_mapping.rs index 5b7054dbf..227067174 100644 --- a/src/repository/target_id_mapping.rs +++ b/src/repository/target_id_mapping.rs @@ -89,14 +89,16 @@ impl TargetIdMapping { None => { self.dirty = true; self.virtual_id_counter += 1; - let record = VirtualIdRecord::new(provider_id, self.virtual_id_counter, item_type, parent_virtual_id, uuid); - self.by_virtual_id.insert(self.virtual_id_counter, record); + let virtual_id = self.virtual_id_counter; + let record = VirtualIdRecord::new(provider_id, virtual_id, item_type, parent_virtual_id, uuid); + self.by_virtual_id.insert(virtual_id, record); self.virtual_id_counter } Some(virtual_id) => { if let Some(record) = self.by_virtual_id.query(virtual_id) { - if record.provider_id != provider_id || record.item_type != item_type || record.parent_virtual_id != parent_virtual_id { + if record.provider_id == provider_id && (record.item_type != item_type || record.parent_virtual_id != parent_virtual_id) { let new_record = VirtualIdRecord::new(provider_id, *virtual_id, item_type, parent_virtual_id, uuid); + println!("updating record {virtual_id} {record:?} {new_record:?} "); self.by_virtual_id.insert(*virtual_id, new_record); self.dirty = true; } @@ -131,10 +133,12 @@ mod tests { #[test] fn test_id_mapping() { - let path = PathBuf::from("/home/euzuner/projects/m3u-test/settings/m3u-catbox/data/m3u/id_mapping.db"); + let path = PathBuf::from("../m3u-test/settings/m3u-silver/data/xt_m3u/id_mapping.db"); let mapping = BPlusTree::::load(&path); mapping.unwrap().traverse(|keys, values| { - println!("{keys:?} {values:?}"); + for (key, value) in keys.iter().zip(values.iter()) { + println!("{key:?} {value:?}\n"); + } }); } -} \ No newline at end of file +} diff --git a/src/repository/xtream_repository.rs b/src/repository/xtream_repository.rs index 965bcca77..5ec21c058 100644 --- a/src/repository/xtream_repository.rs +++ b/src/repository/xtream_repository.rs @@ -705,7 +705,7 @@ async fn rewrite_xtream_series_info

( for episode in episode_list.iter_mut().filter_map(Value::as_object_mut) { if let Some(episode_provider_id) = episode.get(TAG_ID).and_then(get_u32_from_serde_value) { - let uuid = generate_playlist_uuid(&hex_encode(&pli.get_uuid()), &episode_provider_id.to_string(), &provider_url); + let uuid = generate_playlist_uuid(&hex_encode(&pli.get_uuid()), &episode_provider_id.to_string(), PlaylistItemType::Series, &provider_url); let episode_virtual_id = target_id_mapping.get_and_update_virtual_id( uuid, episode_provider_id, diff --git a/src/utils/hash_utils.rs b/src/utils/hash_utils.rs index dd07e1a2c..7c5904745 100644 --- a/src/utils/hash_utils.rs +++ b/src/utils/hash_utils.rs @@ -1,4 +1,4 @@ -use crate::model::playlist::UUIDType; +use crate::model::playlist::{PlaylistItemType, UUIDType}; use crate::repository::storage::hash_string; pub fn extract_id_from_url(url: &str) -> Option { @@ -9,24 +9,15 @@ pub fn extract_id_from_url(url: &str) -> Option { } pub fn get_provider_id(provider_id: &str, url: &str) -> Option { - match provider_id.parse::() { - Ok(id) => Some(id), - Err(_) => match extract_id_from_url(url) { - Some(id) => match id.parse::() { - Ok(newid) => { - Some(newid) - } - Err(_) => None, - }, - None => None, - } - } + provider_id.parse::().ok().or_else(|| { + extract_id_from_url(url)?.parse::().ok() + }) } -pub fn generate_playlist_uuid(key: &str, provider_id: &str, url: &str) -> UUIDType { +pub fn generate_playlist_uuid(key: &str, provider_id: &str, item_type: PlaylistItemType, url: &str) -> UUIDType { if let Some(id) = get_provider_id(provider_id, url) { if id > 0 { - return hash_string(&format!("{key}{id}")); + return hash_string(&format!("{key}{id}{item_type}")); } } hash_string(url)