From 6c238c607f2d6a3ca8deebd093ce7520163dcffc Mon Sep 17 00:00:00 2001 From: euzu Date: Tue, 22 Jul 2025 11:32:46 +0200 Subject: [PATCH] refactored unwrap calls to make it more stable --- backend/src/api/api_utils.rs | 48 +++--- .../src/api/endpoints/api_playlist_utils.rs | 69 ++++---- backend/src/api/endpoints/download_api.rs | 105 +++++++------ backend/src/api/endpoints/hdhomerun_api.rs | 18 +-- backend/src/api/endpoints/hls_api.rs | 49 +++--- backend/src/api/endpoints/m3u_api.rs | 3 +- backend/src/api/endpoints/user_api.rs | 22 ++- backend/src/api/endpoints/v1_api.rs | 23 +-- backend/src/api/endpoints/web_index.rs | 22 ++- backend/src/api/endpoints/websocket_api.rs | 38 +++-- backend/src/api/endpoints/xmltv_api.rs | 26 ++-- backend/src/api/endpoints/xtream_api.rs | 40 ++--- backend/src/api/main_api.rs | 32 ++-- .../src/api/model/active_provider_manager.rs | 11 +- backend/src/api/model/download.rs | 4 +- backend/src/api/model/model_utils.rs | 5 +- backend/src/api/model/provider_config.rs | 2 +- .../api/model/streams/active_client_stream.rs | 45 +++--- .../src/api/model/streams/provider_stream.rs | 3 +- backend/src/model/config/api_proxy.rs | 8 +- backend/src/model/config/app.rs | 3 +- backend/src/model/config/base.rs | 10 +- backend/src/model/config/video_download.rs | 9 +- backend/src/model/xtream.rs | 6 +- backend/src/processing/parser/m3u.rs | 66 ++++---- backend/src/processing/parser/xmltv.rs | 63 +++++--- backend/src/processing/processor/epg.rs | 7 +- backend/src/processing/processor/playlist.rs | 147 ++++++++++-------- .../src/processing/processor/xtream_series.rs | 11 +- backend/src/repository/bplustree.rs | 34 ++-- backend/src/repository/indexed_document.rs | 6 +- backend/src/repository/m3u_repository.rs | 4 +- backend/src/repository/strm_repository.rs | 10 +- backend/src/repository/xtream_repository.rs | 4 +- shared/src/foundation/mapper.rs | 29 ++-- shared/src/model/config/api_user.rs | 3 +- shared/src/model/config/base.rs | 6 +- shared/src/model/config/cache.rs | 16 +- shared/src/model/config/video_download.rs | 4 +- shared/src/utils/directed_graph.rs | 7 +- .../app/components/dashboard/stats_view.rs | 2 +- webui/src/app/components/home.rs | 10 +- webui/src/services/requests.rs | 3 +- webui/src/services/websocket_service.rs | 6 +- 44 files changed, 560 insertions(+), 479 deletions(-) diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index a746fc21e..e7ec0ed3b 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -37,6 +37,8 @@ use std::path::Path; use std::sync::Arc; use tokio::sync::Mutex; use url::Url; +use crate::api::model::active_user_manager::UserSession; +use crate::api::model::provider_config::ProviderConfig; #[macro_export] macro_rules! try_option_bad_request { @@ -76,10 +78,19 @@ macro_rules! try_result_bad_request { }; } +#[macro_export] +macro_rules! try_unwrap_body { + ($body:expr) => { + $body.map_or_else( + |_| axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response(), + |resp| resp.into_response(), + ) + }; +} + pub use try_option_bad_request; pub use try_result_bad_request; -use crate::api::model::active_user_manager::UserSession; -use crate::api::model::provider_config::ProviderConfig; +pub use try_unwrap_body; pub fn get_server_time() -> String { chrono::offset::Local::now().with_timezone(&chrono::Local).format("%Y-%m-%d %H:%M:%S %Z").to_string() @@ -103,13 +114,11 @@ pub async fn serve_file(file_path: &Path, mime_type: mime::Mime) -> impl IntoRes let stream = tokio_util::io::ReaderStream::new(reader); let body = axum::body::Body::from_stream(stream); - axum::response::Response::builder() + try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime_type.to_string()) .header(axum::http::header::CACHE_CONTROL, axum::http::header::HeaderValue::from_static("no-cache")) - .body(body) - .unwrap() - .into_response() + .body(body)) } Err(_) => axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response(), }; @@ -295,8 +304,6 @@ async fn resolve_streaming_strategy(app_state: &AppState, stream_url: &str, addr None => app_state.active_provider.acquire_connection(&input.name, addr).await }; - error!("{:?}", app_state.active_provider.active_connections().await); - let stream_response_params = match &**provider_connection_guard { ProviderAllocation::Exhausted => { debug!("Input {} is exhausted. No connections allowed.", input.name); @@ -556,14 +563,14 @@ pub async fn force_provider_stream_response(addr: &str, let body_stream = prepare_body_stream(app_state, item_type, stream); debug_if_enabled!("Streaming provider forced stream request from {}", sanitize_sensitive_info(&user_session.stream_url)); - return response.body(body_stream).unwrap().into_response(); + return try_unwrap_body!(response.body(body_stream)); } drop(stream_details.provider_connection_guard.take()); if let (Some(stream), _stream_info) = create_channel_unavailable_stream(&app_state.app_config, &[], axum::http::StatusCode::BAD_GATEWAY) { debug!("Streaming custom stream"); - axum::response::Response::builder().status(axum::http::StatusCode::OK).body(axum::body::Body::from_stream(stream)).unwrap().into_response() + try_unwrap_body!(axum::response::Response::builder().status(axum::http::StatusCode::OK).body(axum::body::Body::from_stream(stream))) } else { axum::http::StatusCode::BAD_REQUEST.into_response() } @@ -611,7 +618,7 @@ pub async fn stream_response(addr: &str, for (key, value) in &header_map { response = response.header(key, value); } - response.body(axum::body::Body::from_stream(broadcast_stream)).unwrap().into_response() + try_unwrap_body!(response.body(axum::body::Body::from_stream(broadcast_stream))) } else { axum::http::StatusCode::BAD_REQUEST.into_response() } @@ -637,7 +644,7 @@ pub async fn stream_response(addr: &str, } let body_stream = prepare_body_stream(app_state, item_type, stream); - response.body(body_stream).unwrap().into_response() + try_unwrap_body!(response.body(body_stream)) }; return stream_resp.into_response(); @@ -666,7 +673,7 @@ fn shared_stream_response(app_state: &AppState, stream_url: &str, addr: &str, us for (key, value) in &header_map { response = response.header(key, value); } - return Some(response.body(axum::body::Body::from_stream(stream)).unwrap()); + return response.body(axum::body::Body::from_stream(stream)).ok(); } } None @@ -743,10 +750,10 @@ pub async fn resource_response(app_state: &AppState, resource_url: &str, req_hea let writer = BufWriter::new(file); let add_cache_content = get_add_cache_content(resource_url, &app_state.cache); let stream = PersistPipeStream::new(byte_stream, writer, add_cache_content); - return response_builder.body(axum::body::Body::from_stream(stream)).unwrap().into_response(); + return try_unwrap_body!(response_builder.body(axum::body::Body::from_stream(stream))); } } - return response_builder.body(axum::body::Body::from_stream(byte_stream)).unwrap().into_response(); + return try_unwrap_body!(response_builder.body(axum::body::Body::from_stream(byte_stream))); } debug_if_enabled!("Failed to open resource got status {} for {}", status, sanitize_sensitive_info(resource_url)); } @@ -770,12 +777,10 @@ pub fn separate_number_and_remainder(input: &str) -> (String, Option) { /// # Panics pub fn empty_json_list_response() -> impl IntoResponse + Send { - axum::response::Response::builder() + try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header("Content-Type", mime::APPLICATION_JSON.to_string()) - .body("[]".to_string()) - .unwrap() - .into_response() + .body("[]".to_string())) } pub fn get_username_from_auth_header( @@ -797,11 +802,10 @@ pub fn get_username_from_auth_header( /// # Panics pub fn redirect(url: &str) -> impl IntoResponse { - axum::response::Response::builder() + try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::FOUND) .header("Location", url) - .body(axum::body::Body::empty()) - .unwrap() + .body(axum::body::Body::empty())) } pub async fn is_seek_request( diff --git a/backend/src/api/endpoints/api_playlist_utils.rs b/backend/src/api/endpoints/api_playlist_utils.rs index 59f6b0a03..4a2f40047 100644 --- a/backend/src/api/endpoints/api_playlist_utils.rs +++ b/backend/src/api/endpoints/api_playlist_utils.rs @@ -1,13 +1,13 @@ use crate::model::{AppConfig, Config, ConfigInput, ConfigTarget}; -use shared::model::{InputType, M3uPlaylistItem, PlaylistGroup, PlaylistItemType, TargetType, XtreamCluster}; use crate::repository::{m3u_repository, xtream_repository}; use crate::utils::{m3u, xtream}; +use crate::utils; use axum::response::IntoResponse; +use indexmap::IndexMap; use serde::Serialize; use serde_json::{json, Value}; +use shared::model::{InputType, M3uPlaylistItem, PlaylistGroup, PlaylistItemType, TargetType, XtreamCluster}; use std::sync::Arc; -use indexmap::IndexMap; -use crate::utils; #[derive(serde::Serialize, serde::Deserialize)] struct PlaylistResponseGroup { @@ -57,35 +57,36 @@ where fn group_playlist_items_by_cluster(params: Option<(utils::FileReadGuard, impl Iterator)>) -> (Vec, Vec, Vec) { - if params.is_none() { - return (vec![], vec![], vec![]); - } - let mut live = Vec::new(); - let mut video = Vec::new(); - let mut series = Vec::new(); - let (guard, iter) = params.unwrap(); - for (item, _) in iter { - match item.item_type { - PlaylistItemType::Live - | PlaylistItemType::LiveUnknown - | PlaylistItemType::LiveHls - | PlaylistItemType::LiveDash => { - live.push(item); - } - PlaylistItemType::Catchup - | PlaylistItemType::Video => { - video.push(item); - } - PlaylistItemType::Series - | PlaylistItemType::SeriesInfo => { - series.push(item); + match params { + None => (vec![], vec![], vec![]), + Some((guard, iter)) => { + let mut live = Vec::new(); + let mut video = Vec::new(); + let mut series = Vec::new(); + for (item, _) in iter { + match item.item_type { + PlaylistItemType::Live + | PlaylistItemType::LiveUnknown + | PlaylistItemType::LiveHls + | PlaylistItemType::LiveDash => { + live.push(item); + } + PlaylistItemType::Catchup + | PlaylistItemType::Video => { + video.push(item); + } + PlaylistItemType::Series + | PlaylistItemType::SeriesInfo => { + series.push(item); + } + } } + + drop(guard); + + (live, video, series) } } - - drop(guard); - - (live, video, series) } fn group_playlist_groups_by_cluster(playlist: Vec, input_type: InputType) -> (Vec, Vec, Vec) { @@ -93,7 +94,15 @@ fn group_playlist_groups_by_cluster(playlist: Vec, input_type: In let mut video = Vec::new(); let mut series = Vec::new(); for group in playlist { - let channels = group.channels.iter().map(|item| if input_type == InputType::M3u { serde_json::to_value(item.to_m3u()).unwrap() } else { serde_json::to_value(item.to_xtream()).unwrap() }).collect(); + let channels = group.channels.iter() + .filter_map(|item| { + if input_type == InputType::M3u { + serde_json::to_value(item.to_m3u()) + } else { + serde_json::to_value(item.to_xtream()) + }.ok() + }) + .collect::(); let grp = PlaylistResponseGroup { id: group.id, title: group.title, diff --git a/backend/src/api/endpoints/download_api.rs b/backend/src/api/endpoints/download_api.rs index 8f452109a..e04c146b5 100644 --- a/backend/src/api/endpoints/download_api.rs +++ b/backend/src/api/endpoints/download_api.rs @@ -19,49 +19,56 @@ use shared::error::to_io_error; use crate::utils::request::create_client; async fn download_file(active: Arc>>, client: &reqwest::Client) -> Result<(), String> { - 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) { - Ok(()) => { - if let Some(file_path_str) = file_download.file_path.to_str() { - info!("Downloading {file_path_str}"); - match File::create(&file_download.file_path) { - Ok(mut file) => { - let mut downloaded: u64 = 0; - let mut stream = response.bytes_stream().map_err(to_io_error); - loop { - match stream.try_next().await { - Ok(item) => { - if let Some(chunk) = item { - match file.write_all(&chunk) { - Ok(()) => { - downloaded += chunk.len() as u64; - active.write().await.as_mut().unwrap().size = downloaded; + if let Some(file_download) = active.read().await.as_ref().as_ref() { + match client.get(file_download.url.clone()).send().await { + Ok(response) => { + match fs::create_dir_all(&file_download.file_dir) { + Ok(()) => { + if let Some(file_path_str) = file_download.file_path.to_str() { + info!("Downloading {file_path_str}"); + match File::create(&file_download.file_path) { + Ok(mut file) => { + let mut downloaded: u64 = 0; + let mut stream = response.bytes_stream().map_err(to_io_error); + loop { + match stream.try_next().await { + Ok(item) => { + if let Some(chunk) = item { + match file.write_all(&chunk) { + Ok(()) => { + downloaded += chunk.len() as u64; + if let Some(lock) = active.write().await.as_mut() { + lock.size = downloaded; + } + } + Err(err) => return Err(format!("Error while writing to file: {file_path_str} {err}")) } - Err(err) => return Err(format!("Error while writing to file: {file_path_str} {err}")) + } else { + let megabytes = bytes_to_megabytes(downloaded); + info!("Downloaded {file_path_str}, filesize: {megabytes}MB"); + if let Some(lock) = active.write().await.as_mut() { + lock.size = downloaded; + } + return Ok(()); } - } else { - let megabytes = bytes_to_megabytes(downloaded); - info!("Downloaded {file_path_str}, filesize: {megabytes}MB"); - active.write().await.as_mut().unwrap().size = downloaded; - return Ok(()); } + Err(err) => return Err(format!("Error while writing to file: {file_path_str} {err}")) } - Err(err) => return Err(format!("Error while writing to file: {file_path_str} {err}")) } } + Err(err) => Err(format!("Error while writing to file: {file_path_str} {err}")) } - Err(err) => Err(format!("Error while writing to file: {file_path_str} {err}")) + } else { + Err("Error file-download file-path unknown".to_string()) } - } else { - Err("Error file-download file-path unknown".to_string()) } + Err(err) => Err(format!("Error while creating directory for file: {} {}", &file_download.file_dir.to_str().unwrap_or("?"), err)) } - Err(err) => Err(format!("Error while creating directory for file: {} {}", &file_download.file_dir.to_str().unwrap_or("?"), err)) } + Err(err) => Err(format!("Error while opening url: {} {}", &file_download.url, err)) } - Err(err) => Err(format!("Error while opening url: {} {}", &file_download.url, err)) + } else { + Err("No active file download".to_string()) } } @@ -121,25 +128,29 @@ pub async fn queue_download_file( let app_config = &*app_state.app_config; let config = > as Access>::load(&app_config.config); - if let Some(download_cfg) = config.video.as_ref().unwrap().download.as_ref() { - if download_cfg.directory.is_none() { - return (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Server config missing video.download.directory configuration"}))).into_response(); - } - match FileDownload::new(req.url.as_str(), req.filename.as_str(), download_cfg) { - Some(file_download) => { - app_state.downloads.queue.lock().await.push_back(file_download.clone()); - if app_state.downloads.active.read().await.is_none() { - match run_download_queue(&app_state.app_config, download_cfg, &app_state.downloads).await { - Ok(()) => {} - Err(err) => return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err}))).into_response(), - } - } - axum::Json(download_info!(&file_download)).into_response() + if let Some(video_cfg) = config.video.as_ref() { + if let Some(download_cfg) = video_cfg.download.as_ref() { + if download_cfg.directory.is_empty() { + return (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Server config missing video.download.directory configuration"}))).into_response(); } - None => (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Invalid Arguments"}))).into_response(), + match FileDownload::new(req.url.as_str(), req.filename.as_str(), download_cfg) { + Some(file_download) => { + app_state.downloads.queue.lock().await.push_back(file_download.clone()); + if app_state.downloads.active.read().await.is_none() { + match run_download_queue(&app_state.app_config, download_cfg, &app_state.downloads).await { + Ok(()) => {} + Err(err) => return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err}))).into_response(), + } + } + axum::Json(download_info!(&file_download)).into_response() + } + None => (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Invalid Arguments"}))).into_response(), + } + } else { + (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Server config missing video.download configuration"}))).into_response() } } else { - (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Server config missing video.download configuration"}))).into_response() + (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Server config missing video configuration"}))).into_response() } } diff --git a/backend/src/api/endpoints/hdhomerun_api.rs b/backend/src/api/endpoints/hdhomerun_api.rs index defdaab79..23aedb0fc 100644 --- a/backend/src/api/endpoints/hdhomerun_api.rs +++ b/backend/src/api/endpoints/hdhomerun_api.rs @@ -13,6 +13,7 @@ use serde::{Deserialize, Serialize}; use serde_json::json; use std::sync::Arc; use shared::utils::get_string_from_serde_value; +use crate::api::api_utils::try_unwrap_body; // https://info.hdhomerun.com/info/http_api // const DISCOVERY_BYTES: &[u8] = &[0, 2, 0, 12, 1, 4, 255, 255, 255, 255, 2, 4, 255, 255, 255, 255, 115, 204, 125, 143]; @@ -184,12 +185,10 @@ fn create_device(app_state: &Arc) -> Option { async fn device_xml(axum::extract::State(app_state): axum::extract::State>) -> impl IntoResponse { if let Some(device) = create_device(&app_state) { - axum::response::Response::builder() + try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, "application/xml") - .body(axum::body::Body::from(device.as_xml())) - .unwrap() - .into_response() + .body(axum::body::Body::from(device.as_xml()))) } else { axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response() } @@ -231,11 +230,10 @@ async fn lineup(app_state: &Arc, cfg: &Arc, creden let body_stream = stream::once(async { Ok(Bytes::from("[")) }) .chain(stream) .chain(stream::once(async { Ok(Bytes::from("]")) })); - return axum::response::Response::builder() + return try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) - .body(axum::body::Body::from_stream(body_stream)) - .unwrap().into_response(); + .body(axum::body::Body::from_stream(body_stream))); } else if (use_all || use_xtream) && target.has_output(&TargetType::Xtream) { let server_info = app_state.app_state.app_config.get_user_server_info(credentials); let base_url = server_info.get_base_url(); @@ -262,12 +260,10 @@ async fn lineup(app_state: &Arc, cfg: &Arc, creden .chain(comma_stream) .chain(vod_stream_peek) .chain(stream::once(async { Ok(Bytes::from("]")) })); - return axum::response::Response::builder() + return try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) - .body(axum::body::Body::from_stream(body_stream)) - .unwrap() - .into_response(); + .body(axum::body::Body::from_stream(body_stream))); } axum::http::StatusCode::NOT_FOUND.into_response() } diff --git a/backend/src/api/endpoints/hls_api.rs b/backend/src/api/endpoints/hls_api.rs index c1a4680be..28c5eab53 100644 --- a/backend/src/api/endpoints/hls_api.rs +++ b/backend/src/api/endpoints/hls_api.rs @@ -1,19 +1,20 @@ use crate::api::api_utils::{force_provider_stream_response, get_stream_alternative_url, is_seek_request}; -use crate::api::api_utils::{try_option_bad_request}; +use crate::api::api_utils::try_option_bad_request; +use crate::api::model::active_user_manager::UserSession; use crate::api::model::app_state::AppState; use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, CustomVideoStreamType}; -use crate::model::{ProxyUserCredentials}; +use crate::auth::Fingerprint; use crate::model::ConfigInput; -use shared::model::{PlaylistItemType, UserConnectionPermission, XtreamCluster}; +use crate::model::ProxyUserCredentials; use crate::processing::parser::hls::{get_hls_session_token_and_url_from_token, rewrite_hls, RewriteHlsProps}; -use shared::utils::{is_hls_url, replace_url_extension, sanitize_sensitive_info, HLS_EXT}; +use crate::api::api_utils::try_unwrap_body; use crate::utils::request; use axum::response::IntoResponse; use log::{debug, error}; use serde::Deserialize; +use shared::model::{PlaylistItemType, UserConnectionPermission, XtreamCluster}; +use shared::utils::{is_hls_url, replace_url_extension, sanitize_sensitive_info, HLS_EXT}; use std::sync::Arc; -use crate::api::model::active_user_manager::UserSession; -use crate::auth::Fingerprint; #[derive(Debug, Deserialize)] struct HlsApiPathParams { @@ -25,24 +26,22 @@ struct HlsApiPathParams { } fn hls_response(hls_content: String) -> impl IntoResponse + Send { - let builder = axum::response::Response::builder() + try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) - .header(axum::http::header::CONTENT_TYPE, "application/x-mpegurl"); - builder.body(hls_content) - .unwrap() - .into_response() + .header(axum::http::header::CONTENT_TYPE, "application/x-mpegurl") + .body(hls_content)) } #[allow(clippy::too_many_arguments)] pub(in crate::api) async fn handle_hls_stream_request( - fingerprint: &str, addr: &str, - app_state: &Arc, - user: &ProxyUserCredentials, - user_session: Option<&UserSession>, - hls_url: &str, - virtual_id: u32, - input: &ConfigInput, - connection_permission: UserConnectionPermission) -> impl IntoResponse + Send { + fingerprint: &str, addr: &str, + app_state: &Arc, + user: &ProxyUserCredentials, + user_session: Option<&UserSession>, + hls_url: &str, + virtual_id: u32, + input: &ConfigInput, + connection_permission: UserConnectionPermission) -> impl IntoResponse + Send { let url = replace_url_extension(hls_url, HLS_EXT); let server_info = app_state.app_config.get_user_server_info(user); @@ -52,18 +51,18 @@ pub(in crate::api) async fn handle_hls_stream_request( Some(provider_cfg) => { let stream_url = get_stream_alternative_url(&url, input, &provider_cfg); (stream_url, Some(session.token.to_string())) - }, + } None => (url, None), } - }, + } None => { match app_state.active_provider.get_next_provider(&input.name).await { Some(provider_cfg) => { let stream_url = get_stream_alternative_url(&url, input, &provider_cfg); let user_session_token = format!("{fingerprint}{virtual_id}"); - let session_token= app_state.active_users.create_user_session(user, &user_session_token, virtual_id, &provider_cfg.name, &stream_url, addr, connection_permission); + let session_token = app_state.active_users.create_user_session(user, &user_session_token, virtual_id, &provider_cfg.name, &stream_url, addr, connection_permission); (stream_url, Some(session_token)) - }, + } None => (url, None), } } @@ -110,7 +109,7 @@ async fn hls_api_stream( let user_session_token = format!("{fingerprint}{virtual_id}"); let mut user_session = app_state.active_users.get_user_session(&user.username, &user_session_token); - if let Some(session) = &mut user_session { + if let Some(session) = &mut user_session { if session.permission == UserConnectionPermission::Exhausted { return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response(); } @@ -128,7 +127,7 @@ async fn hls_api_stream( if session.virtual_id == virtual_id { if is_seek_request(XtreamCluster::Live, &req_headers).await { // partial request means we are in reverse proxy mode, seek happened - return force_provider_stream_response(&addr, &app_state, session, PlaylistItemType::LiveHls, &req_headers, &input, &user).await.into_response() + return force_provider_stream_response(&addr, &app_state, session, PlaylistItemType::LiveHls, &req_headers, &input, &user).await.into_response(); } } else { return axum::http::StatusCode::BAD_REQUEST.into_response(); diff --git a/backend/src/api/endpoints/m3u_api.rs b/backend/src/api/endpoints/m3u_api.rs index 5778825be..a176804a0 100644 --- a/backend/src/api/endpoints/m3u_api.rs +++ b/backend/src/api/endpoints/m3u_api.rs @@ -14,6 +14,7 @@ use futures::stream; use log::{debug, error}; use std::sync::Arc; use crate::auth::Fingerprint; +use crate::api::api_utils::try_unwrap_body; async fn m3u_api( api_req: &UserApiRequest, @@ -32,7 +33,7 @@ async fn m3u_api( if api_req.content_type == "m3u_plus" { builder = builder.header("Content-Disposition", "attachment; filename=\"playlist.m3u\""); } - builder.body(axum::body::Body::from_stream(content_stream)).unwrap().into_response() + try_unwrap_body!(builder.body(axum::body::Body::from_stream(content_stream))) } Err(err) => { error!("{}", sanitize_sensitive_info(err.to_string().as_str())); diff --git a/backend/src/api/endpoints/user_api.rs b/backend/src/api/endpoints/user_api.rs index 405851b8a..e21972604 100644 --- a/backend/src/api/endpoints/user_api.rs +++ b/backend/src/api/endpoints/user_api.rs @@ -15,6 +15,7 @@ use std::collections::HashSet; use std::sync::Arc; use axum::response::IntoResponse; use crate::auth::AuthBearer; +use crate::api::api_utils::try_unwrap_body; fn get_categories_from_xtream(categories: Option>) -> Vec { let mut groups: Vec = Vec::new(); @@ -86,13 +87,10 @@ async fn playlist_categories( .chain(m3u_stream) .chain(stream::once(async { Ok::(Bytes::from("}")) })); - - return axum::response::Response::builder() + return try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header("Content-Type", mime::APPLICATION_JSON.to_string()) - .body(axum::body::Body::from_stream(json_stream)) - .unwrap() - .into_response(); + .body(axum::body::Body::from_stream(json_stream))); } } axum::http::StatusCode::BAD_REQUEST.into_response() @@ -134,20 +132,18 @@ async fn playlist_bouquet( let config = &app_state.app_config.config.load(); let xtream = load_user_bouquet_as_json(config, &username, TargetType::Xtream).await; let m3u = load_user_bouquet_as_json(config, &username, TargetType::M3u).await; - return axum::response::Response::builder() + return try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header("Content-Type", mime::APPLICATION_JSON.to_string()) - .body(axum::body::Body::from(format!(r#"{{"xtream": {}, "m3u": {} }}"#, xtream.unwrap_or("null".to_string()), m3u.unwrap_or("null".to_string())))) - .unwrap() - .into_response(); + .body(axum::body::Body::from(format!(r#"{{"xtream": {}, "m3u": {} }}"#, + xtream.unwrap_or("null".to_string()), + m3u.unwrap_or("null".to_string()))))); } } - axum::response::Response::builder() + try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header("Content-Type", mime::APPLICATION_JSON.to_string()) - .body(axum::body::Body::from("{}")) - .unwrap() - .into_response() + .body(axum::body::Body::from("{}"))) } pub fn user_api_register(app_state: Arc) -> axum::Router> { diff --git a/backend/src/api/endpoints/v1_api.rs b/backend/src/api/endpoints/v1_api.rs index 6451a9fd9..f73144d60 100644 --- a/backend/src/api/endpoints/v1_api.rs +++ b/backend/src/api/endpoints/v1_api.rs @@ -11,6 +11,7 @@ use crate::processing::processor::playlist; use crate::repository::user_repository::store_api_user; use crate::utils::ip_checker::get_ips; use crate::{utils, VERSION}; +use crate::api::api_utils::try_unwrap_body; use axum::response::IntoResponse; use log::error; use serde_json::json; @@ -79,9 +80,9 @@ async fn save_config_api_proxy_user( } } else { let config = app_state.app_config.config.load(); - let backup_dir = config.backup_dir.as_ref().unwrap().as_str(); + let backup_dir = config.get_backup_dir(); let paths = app_state.app_config.paths.load(); - if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(&*new_api_proxy), paths.api_proxy_file_path.as_str()) { + if let Some(err) = intern_save_config_api_proxy(backup_dir.as_ref(), &ApiProxyConfigDto::from(&*new_api_proxy), paths.api_proxy_file_path.as_str()) { return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response(); } } @@ -97,8 +98,8 @@ async fn save_config_main( let paths = app_state.app_config.paths.load(); let file_path = paths.config_file_path.as_str(); let config = app_state.app_config.config.load(); - let backup_dir = config.backup_dir.as_ref().unwrap().as_str(); - if let Some(err) = intern_save_config_main(file_path, backup_dir, &cfg) { + let backup_dir = config.get_backup_dir(); + if let Some(err) = intern_save_config_main(file_path, backup_dir.as_ref(), &cfg) { return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response(); } axum::http::StatusCode::OK.into_response() @@ -117,16 +118,16 @@ async fn save_config_api_proxy_config( } } - // TODO wenn hot reload an ist wird doppelt geladen + // TODO if hot reload is on, loaded twice if let Some(old_api_proxy) = app_state.app_config.api_proxy.load().clone() { let mut api_proxy = (*old_api_proxy).clone(); api_proxy.server = req_api_proxy.iter().map(Into::into).collect(); let new_api_proxy = Arc::new(api_proxy); app_state.app_config.api_proxy.store(Some(Arc::clone(&new_api_proxy))); let config = app_state.app_config.config.load(); - let backup_dir = config.backup_dir.as_ref().unwrap().as_str(); + let backup_dir = config.get_backup_dir(); let paths = app_state.app_config.paths.load(); - if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(new_api_proxy.as_ref()), paths.api_proxy_file_path.as_str()) { + if let Some(err) = intern_save_config_api_proxy(backup_dir.as_ref(), &ApiProxyConfigDto::from(new_api_proxy.as_ref()), paths.api_proxy_file_path.as_str()) { return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response(); } } @@ -318,8 +319,8 @@ pub async fn create_status_check(app_state: &Arc) -> StatusCheck { async fn status(axum::extract::State(app_state): axum::extract::State>) -> axum::response::Response { let status = create_status_check(&app_state).await; match serde_json::to_string_pretty(&status) { - Ok(pretty_json) => axum::response::Response::builder().status(axum::http::StatusCode::OK) - .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()).body(pretty_json).unwrap().into_response(), + Ok(pretty_json) => try_unwrap_body!(axum::response::Response::builder().status(axum::http::StatusCode::OK) + .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()).body(pretty_json)), Err(_) => axum::Json(status).into_response(), } } @@ -331,8 +332,8 @@ async fn ipinfo(axum::extract::State(app_state): axum::extract::State axum::response::Response::builder().status(axum::http::StatusCode::OK) - .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()).body(json).unwrap().into_response(), + Ok(json) => try_unwrap_body!(axum::response::Response::builder().status(axum::http::StatusCode::OK) + .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()).body(json)), Err(_) => axum::Json(ipcheck).into_response(), }; } diff --git a/backend/src/api/endpoints/web_index.rs b/backend/src/api/endpoints/web_index.rs index 7c46b85aa..d38bb3ef0 100644 --- a/backend/src/api/endpoints/web_index.rs +++ b/backend/src/api/endpoints/web_index.rs @@ -1,14 +1,15 @@ use crate::api::api_utils::serve_file; +use crate::api::api_utils::try_unwrap_body; use crate::api::model::app_state::AppState; -use crate::auth::{AuthBearer, verify_password, create_jwt_admin, create_jwt_user, is_admin, verify_token}; +use crate::auth::{create_jwt_admin, create_jwt_user, is_admin, verify_password, verify_token, AuthBearer}; use axum::response::IntoResponse; use log::error; use serde_json::json; -use std::path::{Path, PathBuf}; -use std::sync::{Arc}; -use tower::Service; use shared::model::{TokenResponse, UserCredential}; use shared::utils::CONSTANTS; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use tower::Service; fn no_web_auth_token() -> impl axum::response::IntoResponse + Send { axum::Json(TokenResponse { @@ -114,10 +115,9 @@ async fn index( new_content.replace_range(pos..pos + 6, &base_href); } - return axum::response::Response::builder() + return try_unwrap_body!(axum::response::Response::builder() .header("Content-Type", mime::TEXT_HTML_UTF_8.as_ref()) - .body(new_content.into()) - .unwrap(); + .body(new_content)); } Err(err) => { error!("Failed to read web ui index.hml: {err}"); @@ -151,10 +151,9 @@ async fn index_config( } } if let Ok(json_content) = serde_json::to_string(&json_data) { - return axum::response::Response::builder() + return try_unwrap_body!(axum::response::Response::builder() .header("Content-Type", mime::APPLICATION_JSON.as_ref()) - .body(axum::body::Body::from(json_content)) - .unwrap(); + .body(axum::body::Body::from(json_content))); } } } @@ -208,8 +207,7 @@ pub fn index_register_with_path(web_dir_path: &Path, web_ui_path: &str) -> axum: let new_req = axum::http::Request::builder() .method(req.method()) .uri(new_uri) - .body(req.into_body()) - .unwrap(); + .body(req.into_body()).unwrap(); serve_dir.call(new_req) } diff --git a/backend/src/api/endpoints/websocket_api.rs b/backend/src/api/endpoints/websocket_api.rs index 0182881c1..37061469a 100644 --- a/backend/src/api/endpoints/websocket_api.rs +++ b/backend/src/api/endpoints/websocket_api.rs @@ -1,14 +1,14 @@ -use std::sync::Arc; -use axum::{ - extract::ws::{WebSocketUpgrade, WebSocket, Message}, - response::IntoResponse, -}; -use axum::extract::ws::CloseFrame; -use log::{error, info}; -use shared::model::{ProtocolHandler, ProtocolMessage, WsCloseCode, PROTOCOL_VERSION}; use crate::api::endpoints::v1_api::create_status_check; use crate::api::model::app_state::AppState; -use crate::auth::verify_token; +use crate::auth::verify_token_admin; +use axum::extract::ws::CloseFrame; +use axum::{ + extract::ws::{Message, WebSocket, WebSocketUpgrade}, + response::IntoResponse, +}; +use log::{error, info}; +use shared::model::{ProtocolHandler, ProtocolMessage, WsCloseCode, PROTOCOL_VERSION}; +use std::sync::Arc; // WebSocket upgrade handler async fn websocket_handler( @@ -34,8 +34,17 @@ pub fn ws_api_register(web_auth_enabled: bool, web_ui_path: &str) -> axum::Route } } +#[inline] +fn verify_auth_admin_token(auth_token: &str, secret_key: Option<&Vec>) -> bool { + match secret_key.as_ref() { + None => false, + Some(key) => verify_token_admin(auth_token, key.as_slice()) + } +} + // WebSocket communication logic +#[allow(clippy::too_many_lines)] async fn handle_socket(mut socket: WebSocket, app_state: Arc, auth: bool) { let secret_key = if auth { if let Some(web_auth_config) = &app_state.app_config.config.load().web_ui.as_ref().and_then(|c| c.auth.as_ref()) { @@ -48,13 +57,8 @@ async fn handle_socket(mut socket: WebSocket, app_state: Arc, auth: bo None }; - let verify_auth_token = |auth_token: &str| { - secret_key.as_ref().map(|key| verify_token(auth_token, key.as_slice())) - }; - - let mut active_user_change_rx = app_state.active_users.get_active_user_change_channel(); - let mut active_provider_change_rx = app_state.active_provider.get_active_provider_change_channel(); - + let mut active_user_change_rx = app_state.active_users.get_active_user_change_channel(); + let mut active_provider_change_rx = app_state.active_provider.get_active_provider_change_channel(); let mut handler = ProtocolHandler::Version(PROTOCOL_VERSION); @@ -94,7 +98,7 @@ async fn handle_socket(mut socket: WebSocket, app_state: Arc, auth: bo if let Message::Binary(bytes) = msg { match ProtocolMessage::from_bytes(bytes) { Ok(ProtocolMessage::StatusRequest(auth_token)) => { - if !auth || verify_auth_token(&auth_token).is_some() { + if !auth || verify_auth_admin_token(&auth_token, secret_key.as_ref()) { let status = create_status_check(&app_state).await; if let Ok(response) = ProtocolMessage::StatusResponse(status).to_bytes() { if socket.send(Message::Binary(response)).await.is_err() { diff --git a/backend/src/api/endpoints/xmltv_api.rs b/backend/src/api/endpoints/xmltv_api.rs index da55c662c..6c86874dc 100644 --- a/backend/src/api/endpoints/xmltv_api.rs +++ b/backend/src/api/endpoints/xmltv_api.rs @@ -12,19 +12,19 @@ use std::sync::Arc; 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::Config; use crate::model::{ConfigTarget, ProxyUserCredentials, TargetOutput}; -use crate::model::{Config}; use crate::repository::m3u_repository::m3u_get_epg_file_path; use crate::repository::storage::get_target_storage_path; use crate::repository::xtream_repository::{xtream_get_epg_file_path, xtream_get_storage_path}; -use crate::utils; +use crate::{utils}; +use crate::api::api_utils::try_unwrap_body; pub fn get_empty_epg_response() -> impl axum::response::IntoResponse + Send { - axum::response::Response::builder() + try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) // Entspricht `HttpResponse::Ok()` .header(axum::http::header::CONTENT_TYPE, axum::http::HeaderValue::from_static("text/xml")) - .body(axum::body::Body::from(r#""#)) // Setzt den Body der Antwort - .unwrap() + .body(axum::body::Body::from(r#""#))) } fn time_correct(date_time: &str, correction: &TimeDelta) -> String { @@ -159,14 +159,14 @@ fn serve_epg_with_timeshift(epg_file: File, offset_minutes: i32) -> impl axum::r buf.clear(); } - - let compressed_data = xml_writer.into_inner().finish().unwrap(); - axum::response::Response::builder() - .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_OCTET_STREAM.to_string()) - .header(axum::http::header::CONTENT_ENCODING, "gzip") // Set Content-Encoding header - .body(axum::body::Body::from(compressed_data)) - .unwrap() - .into_response() + match xml_writer.into_inner().finish() { + Ok(compressed_data) => + try_unwrap_body!(axum::response::Response::builder() + .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_OCTET_STREAM.to_string()) + .header(axum::http::header::CONTENT_ENCODING, "gzip") // Set Content-Encoding header + .body(axum::body::Body::from(compressed_data))), + Err(err) => (axum::http::StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response() + } } /// Handles XMLTV EPG API requests, serving the appropriate EPG file with optional time-shifting based on user configuration. diff --git a/backend/src/api/endpoints/xtream_api.rs b/backend/src/api/endpoints/xtream_api.rs index 21d26d258..8d24c1d3e 100644 --- a/backend/src/api/endpoints/xtream_api.rs +++ b/backend/src/api/endpoints/xtream_api.rs @@ -38,6 +38,7 @@ use std::path::Path; use std::str::FromStr; use std::sync::Arc; use shared::model::{PlaylistItemType, XtreamCluster, FieldGetAccessor, PlaylistEntry, TargetType, UserConnectionPermission, ProxyType, get_backdrop_path_value, XtreamPlaylistItem}; +use crate::api::api_utils::try_unwrap_body; #[derive(Serialize, Deserialize, Debug, Copy, Clone, Eq, PartialEq)] pub enum ApiStreamContext { @@ -626,12 +627,10 @@ async fn xtream_get_stream_info_response(app_state: &AppState, user: &ProxyUserC if user.proxy == ProxyType::Redirect && cluster == XtreamCluster::Live { return redirect(&info_url).into_response(); } else if let Ok(content) = xtream::get_xtream_stream_info(Arc::clone(&app_state.http_client.load()), &app_state.app_config, user, &input, target, &pli, info_url.as_str(), cluster).await { - return axum::response::Response::builder() + return try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) - .body(axum::body::Body::from(content)) - .unwrap() - .into_response() + .body(axum::body::Body::from(content))) } } } @@ -640,27 +639,21 @@ async fn xtream_get_stream_info_response(app_state: &AppState, user: &ProxyUserC return match cluster { XtreamCluster::Video => { let content = create_vod_info_from_item(target, user, &pli, virtual_record.last_updated); - axum::response::Response::builder() + try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) - .body(axum::body::Body::from(content)) - .unwrap() - .into_response() + .body(axum::body::Body::from(content))) } - XtreamCluster::Live | XtreamCluster::Series => axum::response::Response::builder() + XtreamCluster::Live | XtreamCluster::Series => try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) - .body(axum::body::Body::from("{}".as_bytes())) - .unwrap() - .into_response(), + .body(axum::body::Body::from("{}".as_bytes()))), }; } - axum::response::Response::builder() + try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) - .body(axum::body::Body::from("{}".as_bytes())) - .unwrap() - .into_response() + .body(axum::body::Body::from("{}".as_bytes()))) } async fn xtream_get_short_epg(app_state: &AppState, user: &ProxyUserCredentials, target: &ConfigTarget, stream_id: &str, limit: &str) -> impl IntoResponse + Send { @@ -721,9 +714,9 @@ async fn xtream_player_api_handle_content_action(config: &Config, target_name: & } return Some(serve_file(&file_path, mime::APPLICATION_JSON).await.into_response()); } else if let Some(payload) = content { - return Some(axum::response::Response::builder() + return Some(try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) - .body(payload).unwrap().into_response()); + .body(payload))); } return Some(api_utils::empty_json_list_response().into_response()); } @@ -758,10 +751,10 @@ async fn xtream_get_catchup_response(app_state: &AppState, target: &ConfigTarget serde_json::to_string(&doc) .map_or_else( |_| axum::http::StatusCode::BAD_REQUEST.into_response(), - |result| axum::response::Response::builder() + |result| try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) - .body(result).unwrap().into_response()) + .body(result))) } macro_rules! skip_json_response_if_flag_set { @@ -848,8 +841,7 @@ async fn xtream_player_api( skip_flag_optional!(skip_vod, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Video, &app_state.app_config, &target, category_id, &user).await), crate::model::XC_ACTION_GET_SERIES => skip_flag_optional!(skip_series, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Series, &app_state.app_config, &target, category_id, &user).await), - _ => Some(Err(info_err!(format!("Cant find content: {action} for target: {}", &target.name)) - )), + _ => Some(Err(info_err!(format!("Cant find content: {action} for target: {}", &target.name)))), }; match result { @@ -858,10 +850,10 @@ async fn xtream_player_api( Ok(xtream_iter) => { // Convert the iterator into a stream of `Bytes` let content_stream = xtream_create_content_stream(xtream_iter); - axum::response::Response::builder() + try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()) - .body(axum::body::Body::from_stream(content_stream)).unwrap().into_response() + .body(axum::body::Body::from_stream(content_stream))) } Err(err) => { error!("Failed response for xtream target: {} action: {} error: {}", &target.name, action, err); diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index c915ea2f6..e770dd38f 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -1,8 +1,11 @@ +use crate::api::api_utils::{get_build_time, get_server_time}; +use crate::api::config_watch::exec_config_watch; use crate::api::endpoints::hdhomerun_api::hdhr_api_register; use crate::api::endpoints::hls_api::hls_api_register; use crate::api::endpoints::m3u_api::m3u_api_register; use crate::api::endpoints::v1_api::v1_api_register; use crate::api::endpoints::web_index::{index_register_with_path, index_register_without_path}; +use crate::api::endpoints::websocket_api::ws_api_register; use crate::api::endpoints::xmltv_api::xmltv_api_register; use crate::api::endpoints::xtream_api::xtream_api_register; use crate::api::model::active_provider_manager::ActiveProviderManager; @@ -10,23 +13,20 @@ use crate::api::model::active_user_manager::ActiveUserManager; use crate::api::model::app_state::{create_cache, create_http_client, AppState, CancelTokens, HdHomerunAppState}; use crate::api::model::download::DownloadQueue; use crate::api::model::streams::shared_stream_manager::SharedStreamManager; -use crate::api::scheduler::{exec_scheduler}; +use crate::api::scheduler::exec_scheduler; +use crate::api::serve::serve; use crate::model::{AppConfig, Config, ProcessTargets, RateLimitConfig}; -use crate::model::{Healthcheck}; +use crate::model::Healthcheck; use crate::processing::processor::playlist; +use crate::VERSION; +use arc_swap::{ArcSwap, ArcSwapOption}; +use axum::Router; use log::{error, info}; use std::io::ErrorKind; use std::path::PathBuf; use std::sync::Arc; -use arc_swap::{ArcSwap, ArcSwapOption}; -use axum::Router; use tokio_util::sync::CancellationToken; use tower_governor::key_extractor::SmartIpKeyExtractor; -use crate::api::api_utils::{get_build_time, get_server_time}; -use crate::api::config_watch::exec_config_watch; -use crate::api::endpoints::websocket_api::ws_api_register; -use crate::api::serve::serve; -use crate::VERSION; fn get_web_dir_path(web_ui_enabled: bool, web_root: &str) -> Result { let web_dir = web_root.to_string(); @@ -232,7 +232,7 @@ pub async fn start_server(app_config: Arc, targets: Arc = router.with_state(shared_data.clone()); @@ -246,15 +246,15 @@ pub async fn start_server(app_config: Arc, targets: Arc>, rate_limit_cfg: &RateLimitConfig) -> Router> { if rate_limit_cfg.enabled { - let governor_conf = Arc::new(tower_governor::governor::GovernorConfigBuilder::default() + let governor_conf = tower_governor::governor::GovernorConfigBuilder::default() .key_extractor(SmartIpKeyExtractor) .per_millisecond(rate_limit_cfg.period_millis) .burst_size(rate_limit_cfg.burst_size) - .finish() - .unwrap()); - router.layer(tower_governor::GovernorLayer { - config: governor_conf, - }) + .finish(); + if let Some(config) = governor_conf { router.layer(tower_governor::GovernorLayer { config: Arc::new(config) }) } else { + error!("Failed to initialize rate limiter"); + router + } } else { router } diff --git a/backend/src/api/model/active_provider_manager.rs b/backend/src/api/model/active_provider_manager.rs index c1ddb8123..e7516fe0b 100644 --- a/backend/src/api/model/active_provider_manager.rs +++ b/backend/src/api/model/active_provider_manager.rs @@ -335,11 +335,12 @@ impl MultiProviderLineup { let mut idx = index.load(Ordering::Relaxed) % provider_count; let start = idx; for _ in start..provider_count { - let p = pg.get(idx).unwrap(); - let result = p.get_next(grace, grace_period_timeout_secs).await; - if result.is_some() { - index.store((idx + 1) % provider_count, Ordering::Relaxed); - return result; + if let Some(p) = pg.get(idx) { + let result = p.get_next(grace, grace_period_timeout_secs).await; + if result.is_some() { + index.store((idx + 1) % provider_count, Ordering::Relaxed); + return result; + } } idx = (idx + 1) % provider_count; } diff --git a/backend/src/api/model/download.rs b/backend/src/api/model/download.rs index 2b9489346..3aeb26d27 100644 --- a/backend/src/api/model/download.rs +++ b/backend/src/api/model/download.rs @@ -52,10 +52,10 @@ fn get_download_directory(download_cfg: &VideoDownloadConfig, filestem: &str) -> } } let dir_name = CONSTANTS.re_remove_filename_ending.replace(stem, ""); - let file_dir: PathBuf = [download_cfg.directory.as_ref().unwrap(), dir_name.as_ref()].iter().collect(); + let file_dir: PathBuf = [download_cfg.directory.as_str(), dir_name.as_ref()].iter().collect(); file_dir } else { - PathBuf::from(download_cfg.directory.as_ref().unwrap()) + PathBuf::from(download_cfg.directory.as_str()) } } diff --git a/backend/src/api/model/model_utils.rs b/backend/src/api/model/model_utils.rs index df5060064..31b9e1674 100644 --- a/backend/src/api/model/model_utils.rs +++ b/backend/src/api/model/model_utils.rs @@ -7,7 +7,10 @@ use shared::utils::{filter_response_header}; pub fn get_response_headers(headers: &HeaderMap) -> Vec<(String, String)> { let mut response_headers: Vec<(String, String)> = headers.iter() .filter(|(key, _)| filter_response_header(key.as_str())) - .map(|(key, value)| (key.to_string(), value.to_str().unwrap().to_string())).collect(); + .filter_map(|(key, value)| { + value.to_str().ok().map(|v| (key.to_string(), v.to_string())) + }) + .collect(); response_headers.push((axum::http::header::CONNECTION.as_str().to_string(), "keep-alive".to_string())); response_headers } diff --git a/backend/src/api/model/provider_config.rs b/backend/src/api/model/provider_config.rs index f905c9174..4871b6277 100644 --- a/backend/src/api/model/provider_config.rs +++ b/backend/src/api/model/provider_config.rs @@ -111,7 +111,7 @@ impl ProviderConfig { } fn notify_connection_change(&self, new_connections: usize) { - self.connection_change_tx.send((self.name.clone(), new_connections)).unwrap(); + let _ = self.connection_change_tx.send((self.name.clone(), new_connections)); } #[inline] diff --git a/backend/src/api/model/streams/active_client_stream.rs b/backend/src/api/model/streams/active_client_stream.rs index 79fe46fd5..f6673ee65 100644 --- a/backend/src/api/model/streams/active_client_stream.rs +++ b/backend/src/api/model/streams/active_client_stream.rs @@ -4,18 +4,18 @@ use crate::api::model::active_user_manager::{ActiveUserManager, UserConnectionGu use crate::api::model::app_state::AppState; use crate::api::model::stream::BoxedProviderStream; use crate::api::model::stream_error::StreamError; +use crate::api::model::streams::timed_client_stream::TimedClientStream; use crate::api::model::streams::transport_stream_buffer::TransportStreamBuffer; -use crate::model::{ProxyUserCredentials}; +use crate::model::ProxyUserCredentials; use bytes::Bytes; use futures::Stream; +use futures::StreamExt; use log::{error, info}; +use shared::model::UserConnectionPermission; use std::pin::Pin; use std::sync::atomic::AtomicU8; use std::sync::{Arc, Mutex}; use std::task::{Poll, Waker}; -use crate::api::model::streams::timed_client_stream::TimedClientStream; -use futures::{StreamExt}; -use shared::model::UserConnectionPermission; const INNER_STREAM: u8 = 0_u8; const GRACE_BLOCK_STREAM: u8 = 1_u8; @@ -35,10 +35,10 @@ pub(in crate::api) struct ActiveClientStream { impl ActiveClientStream { pub(crate) fn new(mut stream_details: StreamDetails, - app_state: &AppState, - user: &ProxyUserCredentials, - connection_permission: UserConnectionPermission, - addr: &str) -> Self { + app_state: &AppState, + user: &ProxyUserCredentials, + connection_permission: UserConnectionPermission, + addr: &str) -> Self { let active_user = app_state.active_users.clone(); let active_provider = app_state.active_provider.clone(); if connection_permission == UserConnectionPermission::Exhausted { @@ -59,15 +59,24 @@ impl ActiveClientStream { c.provider_connections_exhausted.clone() )); - let stream = stream_details.stream.take().unwrap(); - let stream = match app_state.app_config.config.load().sleep_timer_mins { - None => stream, - Some(mins) => { - let secs = u32::try_from((u64::from(mins) * 60).min(u64::from(u32::MAX))).unwrap_or(0); - if secs > 0 { - TimedClientStream::new(stream, secs).boxed() - } else { - stream + let stream = match stream_details.stream.take() { + None => { + if let Some(guard) = stream_details.provider_connection_guard.as_ref() { + guard.release(); + } + futures::stream::empty::>().boxed() + } + Some(stream) => { + match app_state.app_config.config.load().sleep_timer_mins { + None => stream, + Some(mins) => { + let secs = u32::try_from((u64::from(mins) * 60).min(u64::from(u32::MAX))).unwrap_or(0); + if secs > 0 { + TimedClientStream::new(stream, secs).boxed() + } else { + stream + } + } } } }; @@ -193,7 +202,7 @@ impl Stream for ActiveClientStream { }; if let Some(buffer) = buffer_opt { - return Poll::Ready(Some(Ok(buffer.next_chunk()))); + return Poll::Ready(Some(Ok(buffer.next_chunk()))); } Poll::Ready(None) diff --git a/backend/src/api/model/streams/provider_stream.rs b/backend/src/api/model/streams/provider_stream.rs index 0895b15a5..b164c609f 100644 --- a/backend/src/api/model/streams/provider_stream.rs +++ b/backend/src/api/model/streams/provider_stream.rs @@ -7,6 +7,7 @@ use reqwest::StatusCode; use axum::response::IntoResponse; use crate::api::model::stream::ProviderStreamResponse; use crate::api::model::streams::transport_stream_buffer::TransportStreamBuffer; +use crate::api::api_utils::try_unwrap_body; #[derive(Debug, Copy, Clone)] pub enum CustomVideoStreamType { @@ -65,7 +66,7 @@ pub fn create_custom_video_stream_response(config: &AppConfig, video_response: C for (key, value) in headers { builder = builder.header(key, value); } - return builder.body(axum::body::Body::from_stream(stream)).unwrap().into_response(); + return try_unwrap_body!(builder.body(axum::body::Body::from_stream(stream))); } axum::http::StatusCode::FORBIDDEN.into_response() } diff --git a/backend/src/model/config/api_proxy.rs b/backend/src/model/config/api_proxy.rs index c923671f9..a8bcc77ab 100644 --- a/backend/src/model/config/api_proxy.rs +++ b/backend/src/model/config/api_proxy.rs @@ -108,9 +108,9 @@ impl ApiProxyConfig { errors.push(err.to_string()); } else { let config = > as Access>::load(&cfg.config); - let backup_dir = config.backup_dir.as_ref().unwrap().as_str(); + let backup_dir = config.get_backup_dir(); self.user = vec![]; - if let Err(err) = utils::save_api_proxy(api_proxy_file, backup_dir, &ApiProxyConfigDto::from(&*self)) { + if let Err(err) = utils::save_api_proxy(api_proxy_file, backup_dir.as_ref(), &ApiProxyConfigDto::from(&*self)) { errors.push(format!("Error saving api proxy file: {err}")); } } @@ -144,8 +144,8 @@ impl ApiProxyConfig { } let config = > as Access>::load(&cfg.config); - let backup_dir = config.backup_dir.as_ref().unwrap().as_str(); - if let Err(err) = save_api_proxy(api_proxy_file, backup_dir, &ApiProxyConfigDto::from(&*self)) { + let backup_dir = config.get_backup_dir(); + if let Err(err) = save_api_proxy(api_proxy_file, backup_dir.as_ref(), &ApiProxyConfigDto::from(&*self)) { errors.push(format!("Error saving api proxy file: {err}")); } else { backup_api_user_db_file(cfg, &user_db_path); diff --git a/backend/src/model/config/app.rs b/backend/src/model/config/app.rs index a53df90ad..2a3de600b 100644 --- a/backend/src/model/config/app.rs +++ b/backend/src/model/config/app.rs @@ -369,7 +369,8 @@ impl AppConfig { TuliproxError::new(TuliproxErrorKind::Info, "API proxy config not loaded".to_string()) }) { let server_info_list = api_proxy.server.clone(); - server_info_list.iter().find(|c| c.name.eq(server_info_name)).map_or_else(|| server_info_list.first().unwrap().clone(), Clone::clone) + server_info_list.iter().find(|c| c.name.eq(server_info_name)) + .map_or_else(|| server_info_list.first().unwrap().clone(), Clone::clone) } else { panic!("ApiProxyServer info not found"); } diff --git a/backend/src/model/config/base.rs b/backend/src/model/config/base.rs index 8f9d506f2..7a14e5573 100644 --- a/backend/src/model/config/base.rs +++ b/backend/src/model/config/base.rs @@ -1,3 +1,4 @@ +use std::borrow::Cow; use std::path::{Path, PathBuf}; use log::{error, info}; use path_clean::PathClean; @@ -8,6 +9,7 @@ use crate::model::{macros, ConfigApi, ReverseProxyConfig, ScheduleConfig}; use crate::model::{HdHomeRunConfig, IpCheckConfig, LogConfig, MessagingConfig, ProxyConfig, VideoConfig, WebUiConfig}; use crate::{utils}; +const DEFAULT_BACKUP_DIR: &str = "backup"; fn create_directories(cfg: &Config, temp_path: &Path) { // Collect the paths into a vector. @@ -15,7 +17,7 @@ fn create_directories(cfg: &Config, temp_path: &Path) { Some(cfg.working_dir.clone()), cfg.backup_dir.clone(), cfg.user_config_dir.clone(), - cfg.video.as_ref().and_then(|v| v.download.as_ref()).and_then(|d| d.directory.clone()), + cfg.video.as_ref().and_then(|v| v.download.as_ref()).map(|d| d.directory.to_string()), cfg.reverse_proxy.as_ref().and_then(|r| r.cache.as_ref().and_then(|c| if c.enabled { Some(c.dir.to_string()) } else { None })) ]; @@ -85,10 +87,14 @@ impl Config { }); } - set_directory(&mut self.backup_dir, "backup", &self.working_dir); + set_directory(&mut self.backup_dir, DEFAULT_BACKUP_DIR, &self.working_dir); set_directory(&mut self.user_config_dir, "user_config", &self.working_dir); } + pub fn get_backup_dir(&self) -> Cow { + self.backup_dir.as_ref().map_or_else(|| Cow::Borrowed(DEFAULT_BACKUP_DIR), |v| Cow::Borrowed(v)) + } + fn prepare_api_web_root(&mut self) { if !self.api.web_root.is_empty() { self.api.web_root = utils::make_absolute_path(&self.api.web_root, &self.working_dir); diff --git a/backend/src/model/config/video_download.rs b/backend/src/model/config/video_download.rs index 4bed8e955..0ca0493c8 100644 --- a/backend/src/model/config/video_download.rs +++ b/backend/src/model/config/video_download.rs @@ -6,20 +6,19 @@ use crate::model::macros; #[derive(Debug, Clone)] pub struct VideoDownloadConfig { pub headers: HashMap, - pub directory: Option, + pub directory: String, pub organize_into_directories: bool, pub episode_pattern: Option, } - macros::from_impl!(VideoDownloadConfig); impl From<&VideoDownloadConfigDto> for VideoDownloadConfig { fn from(dto: &VideoDownloadConfigDto) -> Self { Self { headers: dto.headers.clone(), - directory: dto.directory.clone(), + directory: dto.directory.as_ref().map_or_else(|| "downloads".to_string(), ToString::to_string), organize_into_directories: dto.organize_into_directories, - episode_pattern: dto.episode_pattern.as_ref().map(|s| Regex::new(s).unwrap()), + episode_pattern: dto.episode_pattern.as_ref().and_then(|s| Regex::new(s).ok()), } } } @@ -28,7 +27,7 @@ impl From<&VideoDownloadConfig> for VideoDownloadConfigDto { fn from(instance: &VideoDownloadConfig) -> Self { Self { headers: instance.headers.clone(), - directory: instance.directory.clone(), + directory: Some(instance.directory.clone()), organize_into_directories: instance.organize_into_directories, episode_pattern: instance.episode_pattern.as_ref().map(std::string::ToString::to_string), } diff --git a/backend/src/model/xtream.rs b/backend/src/model/xtream.rs index 584e06fd6..4fcb03a7e 100644 --- a/backend/src/model/xtream.rs +++ b/backend/src/model/xtream.rs @@ -112,13 +112,13 @@ macro_rules! add_opt_i64_property_if_exists { macro_rules! add_opt_f64_property_if_exists { ($vec:expr, $prop:expr, $prop_name:expr) => { - $prop.as_ref().map(|v| $vec.insert(String::from($prop_name), Value::Number(serde_json::value::Number::from_f64(f64::from(*v)).unwrap()))); + $prop.as_ref().map(|v| $vec.insert(String::from($prop_name), Value::Number(serde_json::value::Number::from_f64(f64::from(*v)).unwrap_or_else(|| serde_json::Number::from(0))))); } } macro_rules! add_f64_property_if_exists { ($vec:expr, $prop:expr, $prop_name:expr) => { - $vec.insert(String::from($prop_name), Value::Number(serde_json::value::Number::from_f64(f64::from($prop)).unwrap())); + $vec.insert(String::from($prop_name), Value::Number(serde_json::value::Number::from_f64(f64::from($prop)).unwrap_or_else(|| serde_json::Number::from(0)))); } } @@ -427,7 +427,7 @@ fn append_prepared_series_properties(add_props: Option<&Map>, doc match props.get("rating") { Some(value) => { document.insert("rating".to_string(), match value { - Value::Number(val) => Value::String(format!("{:.0}", val.as_f64().unwrap())), + Value::Number(val) => Value::String(format!("{:.0}", val.as_f64().unwrap_or(0f64))), Value::String(val) => Value::String(val.to_string()), _ => Value::String("0".to_string()), }); diff --git a/backend/src/processing/parser/m3u.rs b/backend/src/processing/parser/m3u.rs index 60e234c7f..e05c05349 100644 --- a/backend/src/processing/parser/m3u.rs +++ b/backend/src/processing/parser/m3u.rs @@ -1,5 +1,5 @@ use crate::model::{Config, ConfigInput}; -use shared::model::{PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, XtreamCluster}; +use shared::model::{PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, XtreamCluster, DEFAULT_VIDEO_EXTENSIONS}; use shared::utils::extract_id_from_url; use std::borrow::BorrowMut; @@ -116,26 +116,25 @@ fn process_header(input_name: &str, video_suffixes: &[&str], content: &str, url: let mut provider_id = None::; let mut c = skip_digit(&mut it); loop { - if c.is_none() { - break; - } - let chr = c.unwrap(); - if chr.is_whitespace() { - // skip - } else if chr == ',' { - plih.title = get_value(&mut stack, &mut it); - } else { - stack.push(chr); - let token = token_till(&mut stack, &mut it, '=', true); - if let Some(t) = token { - let value = token_value(&mut stack, &mut it); - let token = t.to_lowercase(); - if token.as_str() == "xui-id" { - if !value.is_empty() { - provider_id = Some(value); - } + match c { + None=> break, + Some(chr) => { + if chr.is_whitespace() { + // skip + } else if chr == ',' { + plih.title = get_value(&mut stack, &mut it); } else { - process_header_fields!(plih, token.as_str(), + stack.push(chr); + let token = token_till(&mut stack, &mut it, '=', true); + if let Some(t) = token { + let value = token_value(&mut stack, &mut it); + let token = t.to_lowercase(); + if token.as_str() == "xui-id" { + if !value.is_empty() { + provider_id = Some(value); + } + } else { + process_header_fields!(plih, token.as_str(), (id, "tvg-id"), (group, "group-title"), (name, "tvg-name"), @@ -146,6 +145,8 @@ fn process_header(input_name: &str, video_suffixes: &[&str], content: &str, url: (logo_small, "tvg-logo-small"), (time_shift, "timeshift"), (rec, "tvg-rec"); value); + } + } } } } @@ -188,7 +189,6 @@ fn process_header(input_name: &str, video_suffixes: &[&str], content: &str, url: plih } - pub fn consume_m3u<'a, I, F: FnMut(PlaylistItem)>(cfg: &Config, input: &ConfigInput, lines: I, mut visit: F) where I: Iterator, @@ -197,7 +197,12 @@ where let mut group: Option = None; let input_name = input.name.as_str(); - let video_suffixes = cfg.video.as_ref().unwrap().extensions.iter().map(String::as_str).collect::>(); + let video_suffixes = match cfg.video.as_ref() { + Some(config) => { + config.extensions.iter().map(String::as_str).collect::>() + }, + None => DEFAULT_VIDEO_EXTENSIONS.to_vec() + }; for line in lines { if line.starts_with("#EXTINF") { header = Some(String::from(line)); @@ -248,18 +253,23 @@ where sort_order_idx += 1; } std::collections::hash_map::Entry::Occupied(o) => { - sort_order.get_mut(*o.get()).unwrap().push(item); + if let Some(order) = sort_order.get_mut(*o.get()) { + order.push(item); + } } } }); let mut grp_id = 0; - let result: Vec = sort_order.into_iter().map(|channels| { + let result: Vec = sort_order.into_iter().filter_map(|channels| { // create a group based on the first playlist item let channel = channels.first(); - let (cluster, group_title) = channel.map(|pli| - (pli.header.xtream_cluster, &pli.header.group)).unwrap(); - grp_id += 1; - PlaylistGroup { id: grp_id, xtream_cluster: cluster, title: group_title.to_string(), channels } + if let Some((cluster, group_title)) = channel.map(|pli| + (pli.header.xtream_cluster, &pli.header.group)) { + grp_id += 1; + Some(PlaylistGroup { id: grp_id, xtream_cluster: cluster, title: group_title.to_string(), channels }) + } else { + None + } }).collect(); result } diff --git a/backend/src/processing/parser/xmltv.rs b/backend/src/processing/parser/xmltv.rs index 0ef275686..7bf8d073f 100644 --- a/backend/src/processing/parser/xmltv.rs +++ b/backend/src/processing/parser/xmltv.rs @@ -2,6 +2,7 @@ use crate::model::{Epg, TVGuide, XmlTag, XmlTagIcon, EPG_ATTRIB_CHANNEL, EPG_ATT use crate::model::{EpgSmartMatchConfig, PersistedEpgSource}; use crate::processing::processor::epg::EpgIdCache; use crate::utils::compressed_file_reader::CompressedFileReader; +use dashmap::DashMap; use deunicode::deunicode; use quick_xml::events::{BytesStart, BytesText, Event}; use quick_xml::Reader; @@ -14,7 +15,6 @@ use std::collections::HashMap; use std::mem; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; -use dashmap::DashMap; /// Splits a string at the first delimiter if the prefix matches a known country code. /// @@ -143,13 +143,14 @@ impl TVGuide { if !matched && fuzzy_matching { let (fuzzy_matched, matched_normalized_name) = Self::find_best_fuzzy_match(id_cache, tag); if fuzzy_matched { - let key = matched_normalized_name.unwrap(); - let id = epg_id.to_string(); - id_cache.normalized.entry(key).and_modify(|entry| { - entry.replace(id.clone()); - id_cache.channel_epg_id.insert(Cow::Owned(id)); - matched = true; - }); + if let Some(key) = matched_normalized_name { + let id = epg_id.to_string(); + id_cache.normalized.entry(key).and_modify(|entry| { + entry.replace(id.clone()); + id_cache.channel_epg_id.insert(Cow::Owned(id)); + matched = true; + }); + } } } matched @@ -190,9 +191,10 @@ impl TVGuide { #[allow(clippy::cast_sign_loss)] let mjw = min(100, (match_jw * 100.0).round() as u16); if mjw >= match_threshold { - let mut lock = data.lock().unwrap(); - if lock.0 < mjw { - *lock = (mjw, Some(Cow::Borrowed(norm_key))); + if let Ok(mut lock) = data.lock() { + if lock.0 < mjw { + *lock = (mjw, Some(Cow::Borrowed(norm_key))); + } } if mjw > best_match_threshold { return true; // (true, matched_normalized_epg_id.map(|s| s.to_string())); @@ -206,8 +208,9 @@ impl TVGuide { // is there an early exit strategy ??? if early_exit_flag.load(Ordering::SeqCst) { - let result = data.lock().unwrap().1.take(); - return (true, result.as_ref().map(std::string::ToString::to_string)); + if let Ok(mut result) = data.lock() { + return (true, result.1.take().as_ref().map(ToString::to_string)); + } } (false, None) } @@ -354,13 +357,14 @@ where } } - fn handle_text_tag(stack: &mut [XmlTag], e: &BytesText) { if !stack.is_empty() { if let Ok(text) = e.unescape() { let t = text.trim(); if !t.is_empty() { - stack.last_mut().unwrap().value = Some(t.to_string()); + if let Some(tag) = stack.last_mut() { + tag.value = Some(t.to_string()); + } } } } @@ -403,14 +407,16 @@ fn collect_tag_attributes(e: &BytesStart, is_channel: bool, is_program: bool) -> let attributes = e.attributes().filter_map(Result::ok) .filter_map(|a| { let key = String::from_utf8_lossy(a.key.as_ref()).to_string(); - let mut value = String::from(a.unescape_value().unwrap().as_ref()); - if (is_channel && key == EPG_ATTRIB_ID) || (is_program && key == EPG_ATTRIB_CHANNEL) { - value = value.to_lowercase().to_string(); - } - if value.is_empty() { - None + if let Ok(value) = a.unescape_value().as_ref() { + if value.is_empty() { + None + } else if (is_channel && key == EPG_ATTRIB_ID) || (is_program && key == EPG_ATTRIB_CHANNEL) { + Some((key, value.to_lowercase().to_string())) + } else { + Some((key, value.to_string())) + } } else { - Some((key, value)) + None } }).collect::>(); attributes @@ -466,13 +472,20 @@ pub fn flatten_tvguide(tv_guides: &[Epg]) -> Option { } }); - epg_children.lock().unwrap().extend(children); + if let Ok(mut guard) = epg_children.lock() { + guard.extend(children); + } }); + let children = if let Ok(mut children) = epg_children.lock() { + mem::take(&mut *children) + } else { + vec![] + }; let epg = Epg { logo_override: false, priority: 0, attributes: epg_attributes, - children: mem::take(&mut *epg_children.lock().unwrap()), + children, }; Some(epg) } @@ -492,7 +505,7 @@ mod tests { /// parse_normalize().unwrap(); /// ``` fn parse_normalize() { - let epg_normalize_dto = EpgSmartMatchConfigDto {..Default::default()}; + let epg_normalize_dto = EpgSmartMatchConfigDto { ..Default::default() }; let epg_normalize = EpgSmartMatchConfig::from(epg_normalize_dto); let normalized = normalize_channel_name("Love Nature", &epg_normalize); assert_eq!(normalized, "lovenature".to_string()); diff --git a/backend/src/processing/processor/epg.rs b/backend/src/processing/processor/epg.rs index 3a5ff0226..85736d8cb 100644 --- a/backend/src/processing/processor/epg.rs +++ b/backend/src/processing/processor/epg.rs @@ -146,8 +146,9 @@ fn assign_channel_epg(new_epg: &mut Vec, fp: &mut FetchedPlaylist, id_cache for epg_source in epg_sources { // icon tags let icon_tags: HashMap<&String, &XmlTag> = epg_source.children.iter() - .filter(|tag| tag.icon != XmlTagIcon::Undefined && tag.get_attribute_value(EPG_ATTRIB_ID).is_some()) - .map(|t| (t.get_attribute_value(EPG_ATTRIB_ID).unwrap(), t)).collect(); + .filter(|tag| tag.icon != XmlTagIcon::Undefined) + .filter_map(|tag| tag.get_attribute_value(EPG_ATTRIB_ID).map(|id| (id, tag))) + .collect(); let assign_values = |chan: &mut PlaylistItem| { if id_cache.smart_match_enabled && chan.header.epg_channel_id.is_none() { @@ -169,7 +170,7 @@ fn assign_channel_epg(new_epg: &mut Vec, fp: &mut FetchedPlaylist, id_cache if let Some(epg_channel_id) = chan.header.epg_channel_id.as_ref() { if !icon_assigned.contains(epg_channel_id) && (epg_source.logo_override || chan.header.logo.is_empty() || chan.header.logo_small.is_empty()) { - if let Some(icon_tag) = icon_tags.get(chan.header.epg_channel_id.as_ref().unwrap()) { + if let Some(icon_tag) = icon_tags.get(epg_channel_id) { if let XmlTagIcon::Src(icon) = &icon_tag.icon { icon_assigned.insert(epg_channel_id.to_string()); if epg_source.logo_override || chan.header.logo.is_empty() { diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 7ac57b052..3c30760a7 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -234,70 +234,71 @@ fn is_target_enabled(target: &ConfigTarget, user_targets: &ProcessTargets) -> bo async fn process_source(client: Arc, cfg: Arc, source_idx: usize, user_targets: Arc) -> (Vec, Vec, Vec) { let sources = cfg.sources.load(); - let source = sources.get_source_at(source_idx).unwrap(); let mut errors = vec![]; let mut input_stats = HashMap::::new(); let mut target_stats = Vec::::new(); - let mut source_playlists = Vec::with_capacity(128); - // Download the sources - let mut source_downloaded = false; - for input in &source.inputs { - if is_input_enabled(input, &user_targets) { - let config = cfg.config.load(); - let working_dir = &config.working_dir; + if let Some(source) = sources.get_source_at(source_idx) { + let mut source_playlists = Vec::with_capacity(128); + // Download the sources + let mut source_downloaded = false; + for input in &source.inputs { + if is_input_enabled(input, &user_targets) { + let config = cfg.config.load(); + let working_dir = &config.working_dir; - source_downloaded = true; - let start_time = Instant::now(); - let (mut playlistgroups, mut error_list) = match input.input_type { - InputType::M3u => m3u::get_m3u_playlist(Arc::clone(&client), &config, input, working_dir).await, - InputType::Xtream => xtream::get_xtream_playlist(&config, Arc::clone(&client), input, working_dir).await, - InputType::M3uBatch | InputType::XtreamBatch => (vec![], vec![]) - }; - let (tvguide, mut tvguide_errors) = if error_list.is_empty() { - epg::get_xmltv(Arc::clone(&client), input, working_dir).await - } else { - (None, vec![]) - }; - errors.append(&mut error_list); - errors.append(&mut tvguide_errors); - let group_count = playlistgroups.len(); - let channel_count = playlistgroups.iter() - .map(|group| group.channels.len()) - .sum(); - let input_name = &input.name; - if playlistgroups.is_empty() { - info!("Source is empty {input_name}"); - errors.push(notify_err!(format!("Source is empty {input_name}"))); - } else { - playlistgroups.iter_mut().for_each(PlaylistGroup::on_load); - source_playlists.push( - FetchedPlaylist { - input, - playlistgroups, - epg: tvguide, - } - ); - } - let elapsed = start_time.elapsed().as_secs(); - input_stats.insert(input_name.to_string(), create_input_stat(group_count, channel_count, error_list.len(), - input.input_type, input_name, elapsed)); - } - } - if source_downloaded { - if source_playlists.is_empty() { - debug!("Source at index {source_idx} is empty"); - errors.push(notify_err!(format!("Source at {source_idx} is empty"))); - } else { - debug_if_enabled!("Source has {} groups", source_playlists.iter().map(|fpl| fpl.playlistgroups.len()).sum::()); - for target in &source.targets { - if is_target_enabled(target, &user_targets) { - match process_playlist_for_target(&cfg, Arc::clone(&client), &mut source_playlists, target, &mut input_stats, &mut errors).await { - Ok(()) => { - target_stats.push(TargetStats::success(&target.name)); + source_downloaded = true; + let start_time = Instant::now(); + let (mut playlistgroups, mut error_list) = match input.input_type { + InputType::M3u => m3u::get_m3u_playlist(Arc::clone(&client), &config, input, working_dir).await, + InputType::Xtream => xtream::get_xtream_playlist(&config, Arc::clone(&client), input, working_dir).await, + InputType::M3uBatch | InputType::XtreamBatch => (vec![], vec![]) + }; + let (tvguide, mut tvguide_errors) = if error_list.is_empty() { + epg::get_xmltv(Arc::clone(&client), input, working_dir).await + } else { + (None, vec![]) + }; + errors.append(&mut error_list); + errors.append(&mut tvguide_errors); + let group_count = playlistgroups.len(); + let channel_count = playlistgroups.iter() + .map(|group| group.channels.len()) + .sum(); + let input_name = &input.name; + if playlistgroups.is_empty() { + info!("Source is empty {input_name}"); + errors.push(notify_err!(format!("Source is empty {input_name}"))); + } else { + playlistgroups.iter_mut().for_each(PlaylistGroup::on_load); + source_playlists.push( + FetchedPlaylist { + input, + playlistgroups, + epg: tvguide, } - Err(mut err) => { - target_stats.push(TargetStats::failure(&target.name)); - errors.append(&mut err); + ); + } + let elapsed = start_time.elapsed().as_secs(); + input_stats.insert(input_name.to_string(), create_input_stat(group_count, channel_count, error_list.len(), + input.input_type, input_name, elapsed)); + } + } + if source_downloaded { + if source_playlists.is_empty() { + debug!("Source at index {source_idx} is empty"); + errors.push(notify_err!(format!("Source at {source_idx} is empty"))); + } else { + debug_if_enabled!("Source has {} groups", source_playlists.iter().map(|fpl| fpl.playlistgroups.len()).sum::()); + for target in &source.targets { + if is_target_enabled(target, &user_targets) { + match process_playlist_for_target(&cfg, Arc::clone(&client), &mut source_playlists, target, &mut input_stats, &mut errors).await { + Ok(()) => { + target_stats.push(TargetStats::success(&target.name)); + } + Err(mut err) => { + target_stats.push(TargetStats::failure(&target.name)); + errors.append(&mut err); + } } } } @@ -351,13 +352,17 @@ async fn process_sources(client: Arc, config: &Arc, let handles = &mut handle_list; let process = move || { // TODO better way ? - let rt = tokio::runtime::Runtime::new().unwrap(); - rt.block_on(async { - let (input_stats, target_stats, mut res_errors) = process_source(Arc::clone(&http_client), cfg, index, usr_trgts).await; - shared_errors.lock().await.append(&mut res_errors); - let process_stats = SourceStats::new(input_stats, target_stats); - shared_stats.lock().await.push(process_stats); - }); + match tokio::runtime::Runtime::new() { + Ok(rt) => { + rt.block_on(async { + let (input_stats, target_stats, mut res_errors) = process_source(Arc::clone(&http_client), cfg, index, usr_trgts).await; + shared_errors.lock().await.append(&mut res_errors); + let process_stats = SourceStats::new(input_stats, target_stats); + shared_stats.lock().await.push(process_stats); + }); + }, + Err(err) => error!("Could not create runtime !!! {err}"), + } }; handles.push(thread::spawn(process)); if handles.len() >= thread_num as usize { @@ -374,7 +379,11 @@ async fn process_sources(client: Arc, config: &Arc, for handle in handle_list { let _ = handle.join(); } - (Arc::try_unwrap(stats).unwrap().into_inner(), Arc::try_unwrap(errors).unwrap().into_inner()) + if let (Ok(s), Ok(e)) = (Arc::try_unwrap(stats), Arc::try_unwrap(errors)) { + (s.into_inner(), e.into_inner()) + } else { + (vec![], vec![]) + } } pub type ProcessingPipe = Vec Option>>; @@ -430,7 +439,9 @@ fn flatten_groups(playlistgroups: Vec) -> Vec { sort_order.push(group); } std::collections::hash_map::Entry::Occupied(o) => { - sort_order.get_mut(*o.get()).unwrap().channels.extend(group.channels); + if let Some(pl_group) = sort_order.get_mut(*o.get()) { + pl_group.channels.extend(group.channels); + } } } } diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index daf3a5fff..030db216a 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -16,7 +16,7 @@ use std::fs::File; use std::io::{BufWriter, Write}; use std::sync::Arc; use std::time::Instant; -use log::{info, log_enabled, Level}; +use log::{error, info, log_enabled, Level}; use shared::utils::bincode_serialize; use crate::model::{XtreamSeriesEpisode, XtreamSeriesInfoEpisode}; use crate::utils; @@ -36,9 +36,12 @@ fn write_series_episode_record_to_wal_file( let series_episode = XtreamSeriesEpisode::from(episode); if let Ok(content_bytes) = bincode_serialize(&series_episode) { writer.write_all(&provider_id.to_le_bytes())?; - let len = u32::try_from(content_bytes.len()).unwrap(); - writer.write_all(&len.to_le_bytes())?; - writer.write_all(&content_bytes)?; + if let Ok(len) = u32::try_from(content_bytes.len()) { + writer.write_all(&len.to_le_bytes())?; + writer.write_all(&content_bytes)?; + } else { + error!("Cant write to WAL file, content length exceeds u32"); + } } Ok(()) } diff --git a/backend/src/repository/bplustree.rs b/backend/src/repository/bplustree.rs index d5e532224..c57defeeb 100644 --- a/backend/src/repository/bplustree.rs +++ b/backend/src/repository/bplustree.rs @@ -1,16 +1,16 @@ +use crate::utils; +use log::error; +use ruzstd::decoding::StreamingDecoder; +use ruzstd::encoding::{compress_to_vec, CompressionLevel}; +use serde::{Deserialize, Serialize}; +use shared::error::{str_to_io_error, to_io_error}; +use shared::utils::{bincode_deserialize, bincode_serialize}; use std::fs::File; use std::io::{self, BufReader, Read, Seek, SeekFrom, Write}; use std::marker::PhantomData; use std::mem::size_of; use std::path::Path; -use shared::error::{str_to_io_error, to_io_error}; -use log::error; -use ruzstd::decoding::StreamingDecoder; -use ruzstd::encoding::{compress_to_vec, CompressionLevel}; -use serde::{Deserialize, Serialize}; use tempfile::NamedTempFile; -use shared::utils::{bincode_deserialize, bincode_serialize}; -use crate::utils; const BLOCK_SIZE: usize = 4096; const BINCODE_OVERHEAD: usize = 8; @@ -102,9 +102,10 @@ where fn find_leaf_entry(node: &Self) -> Option<&K> { if node.is_leaf { node.keys.first() - } else { - let child = node.children.first().unwrap(); + } else if let Some(child) = node.children.first() { Self::find_leaf_entry(child) + } else { + None } } @@ -185,7 +186,9 @@ where let mut node = Self::new(false); node.keys = self.keys.split_off(median + 1); node.children = self.children.split_off(median + 1); - self.children.push(node.children.first().unwrap().clone()); + if let Some(child) = node.children.first() { + self.children.push(child.clone()); + } node } } @@ -301,7 +304,7 @@ where // Deserialize values if leaf node let values = if is_leaf { - let use_compression = u8::from_le_bytes(buffer[read_pos..=read_pos].try_into().unwrap()) == 1; + let use_compression = u8::from_le_bytes(buffer[read_pos..=read_pos].try_into().unwrap_or([0u8])) == 1; read_pos += FLAG_SIZE; let values_length = u32_from_bytes(&buffer[read_pos..read_pos + LEN_SIZE])? as usize; read_pos += LEN_SIZE; @@ -528,7 +531,11 @@ where }; } let child_idx = get_entry_index_upper_bound::(&node.keys, key); - offset = *pointers.unwrap().get(child_idx).unwrap(); + if let Some(child_offset) = pointers.unwrap().get(child_idx) { + offset = *child_offset; + } else { + return None; + } } Err(err) => { error!("Failed to read id tree from file {err}"); @@ -769,9 +776,9 @@ mod tests { use std::io; use std::path::PathBuf; + use crate::repository::bplustree::{BPlusTree, BPlusTreeQuery, BPlusTreeUpdate}; use serde::{Deserialize, Serialize}; use shared::utils::generate_random_string; - use crate::repository::bplustree::{BPlusTree, BPlusTreeQuery, BPlusTreeUpdate}; // Example usage with a simple struct #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] @@ -878,7 +885,6 @@ mod tests { assert!(format!("{content} {}", k + 1).eq(&v.data), "Wrong entry"); }); }); - } #[test] diff --git a/backend/src/repository/indexed_document.rs b/backend/src/repository/indexed_document.rs index f393e26a7..8c4965db8 100644 --- a/backend/src/repository/indexed_document.rs +++ b/backend/src/repository/indexed_document.rs @@ -208,7 +208,7 @@ where } } Err(err) => { - return Err(str_to_io_error(&format!("failed to write document: {} - {}", self.main_path.to_str().unwrap(), err))); + return Err(str_to_io_error(&format!("failed to write document: {} - {}", self.main_path.display(), err))); } } Ok(()) @@ -320,7 +320,7 @@ where Err(e) => Err(e) } } else { - Err(Error::new(ErrorKind::NotFound, format!("File not found {}", main_path.to_str().unwrap()))) + Err(Error::new(ErrorKind::NotFound, format!("File not found {}", main_path.display()))) } } @@ -404,7 +404,7 @@ impl IndexedDocumentDirectAccess { return Ok(item); } } - Err(str_to_io_error(&format!("Failed to read item for id {:?} - {}", doc_id, main_path.to_str().unwrap()))) + Err(str_to_io_error(&format!("Failed to read item for id {:?} - {}", doc_id, main_path.display()))) } } diff --git a/backend/src/repository/m3u_repository.rs b/backend/src/repository/m3u_repository.rs index faf67e225..e603083eb 100644 --- a/backend/src/repository/m3u_repository.rs +++ b/backend/src/repository/m3u_repository.rs @@ -16,7 +16,7 @@ use crate::utils; macro_rules! cant_write_result { ($path:expr, $err:expr) => { - create_tuliprox_error!(TuliproxErrorKind::Notify, "failed to write m3u playlist: {} - {}", $path.to_str().unwrap() ,$err) + create_tuliprox_error!(TuliproxErrorKind::Notify, "failed to write m3u playlist: {} - {}", $path.display() ,$err) } } @@ -44,7 +44,7 @@ fn persist_m3u_playlist_as_text(cfg: &Config, target: &ConfigTarget, target_outp } } Err(_) => { - error!("Can't write m3u plain playlist {}", &m3u_filename.to_str().unwrap()); + error!("Can't write m3u plain playlist {}", &m3u_filename.display()); } } } diff --git a/backend/src/repository/strm_repository.rs b/backend/src/repository/strm_repository.rs index 5a7ae7056..c1e2484ee 100644 --- a/backend/src/repository/strm_repository.rs +++ b/backend/src/repository/strm_repository.rs @@ -128,10 +128,12 @@ fn style_rename_year<'a>( if let Ok(year) = year_match.as_str().parse::() { if (1900..=cur_year).contains(&year) { years.push(year); - let match_start = caps.get(0).unwrap().start(); - let match_end = caps.get(0).unwrap().end(); - new_name.push_str(&name[last_index..match_start]); - last_index = match_end; + if let Some(matched) = caps.get(0) { + let match_start = matched.start(); + let match_end = matched.end(); + new_name.push_str(&name[last_index..match_start]); + last_index = match_end; + } } } } diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index 342450fe6..ec14207fe 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/backend/src/repository/xtream_repository.rs @@ -32,7 +32,7 @@ macro_rules! cant_write_result { create_tuliprox_error!( TuliproxErrorKind::Notify, "failed to write xtream playlist: {} - {}", - $path.to_str().unwrap(), + $path.display(), $err ) }; @@ -64,7 +64,7 @@ fn ensure_xtream_storage_path(cfg: &Config, target_name: &str) -> Result { - let map_key = first.into_inner().next().unwrap(); - match map_key.as_rule() { - Rule::field => { - MapKey::FieldAccess(map_key.as_str().trim().to_string()) - } - Rule::var_access => { - let text = map_key.as_str(); - if text.contains('.') { - let splitted: Vec<&str> = text.splitn(2, '.').collect(); - MapKey::VarAccess(splitted[0].trim().to_string(), splitted[1].trim().to_string()) - } else { - MapKey::Identifier(text.trim().to_string()) + if let Some(map_key) = first.into_inner().next() { + match map_key.as_rule() { + Rule::field => { + MapKey::FieldAccess(map_key.as_str().trim().to_string()) } + Rule::var_access => { + let text = map_key.as_str(); + if text.contains('.') { + let splitted: Vec<&str> = text.splitn(2, '.').collect(); + MapKey::VarAccess(splitted[0].trim().to_string(), splitted[1].trim().to_string()) + } else { + MapKey::Identifier(text.trim().to_string()) + } + } + _ => return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Unexpected map case key: {:?}", map_key.as_rule()), } - _ => return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Unexpected map case key: {:?}", map_key.as_rule()), + } else { + return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Missing map case key"); } } _ => return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Unexpected map case key: {:?}", first.as_rule()), diff --git a/shared/src/model/config/api_user.rs b/shared/src/model/config/api_user.rs index d6c78b468..63d177cdd 100644 --- a/shared/src/model/config/api_user.rs +++ b/shared/src/model/config/api_user.rs @@ -77,8 +77,7 @@ impl FromStr for ProxyType { return Ok(Self::Reverse(None)); } - if s.starts_with(Self::REVERSE) { - let suffix = s.strip_prefix(Self::REVERSE).unwrap(); + if let Some(suffix) = s.strip_prefix(Self::REVERSE) { if let Ok(force_redirect) = ClusterFlags::try_from(suffix) { if force_redirect.has_full_flags() { return Ok(ProxyType::Reverse(None)); diff --git a/shared/src/model/config/base.rs b/shared/src/model/config/base.rs index f59e54b36..345098d6c 100644 --- a/shared/src/model/config/base.rs +++ b/shared/src/model/config/base.rs @@ -1,7 +1,5 @@ use crate::error::{TuliproxError, TuliproxErrorKind}; -use crate::model::{ConfigApiDto, HdHomeRunConfigDto, IpCheckConfigDto, LogConfigDto, - MessagingConfigDto, ProxyConfigDto, ReverseProxyConfigDto, ScheduleConfigDto, - VideoConfigDto, WebUiConfigDto}; +use crate::model::{ConfigApiDto, HdHomeRunConfigDto, IpCheckConfigDto, LogConfigDto, MessagingConfigDto, ProxyConfigDto, ReverseProxyConfigDto, ScheduleConfigDto, VideoConfigDto, WebUiConfigDto, DEFAULT_VIDEO_EXTENSIONS}; use crate::utils::default_connect_timeout_secs; pub const DEFAULT_USER_AGENT: &str = "VLC/3.0.16 LibVLC/3.0.16"; @@ -103,7 +101,7 @@ impl ConfigDto { match &mut self.video { None => { self.video = Some(VideoConfigDto { - extensions: vec!["mkv".to_string(), "avi".to_string(), "mp4".to_string()], + extensions: DEFAULT_VIDEO_EXTENSIONS.iter().map(ToString::to_string).collect(), download: None, web_search: None, }); diff --git a/shared/src/model/config/cache.rs b/shared/src/model/config/cache.rs index c4d0592f4..5ca65db1d 100644 --- a/shared/src/model/config/cache.rs +++ b/shared/src/model/config/cache.rs @@ -20,14 +20,14 @@ impl CacheConfigDto { pub(crate) fn prepare(&mut self, working_dir: &str) -> Result<(), TuliproxError> { if self.enabled { let work_path = PathBuf::from(working_dir); - if self.dir.is_none() { - self.dir = Some(work_path.join("cache").to_string_lossy().to_string()); - } else { - let mut cache_dir = self.dir.as_ref().unwrap().to_string(); - if PathBuf::from(&cache_dir).is_relative() { - cache_dir = work_path.join(&cache_dir).clean().to_string_lossy().to_string(); - } - self.dir = Some(cache_dir.to_string()); + match self.dir.as_ref() { + None => self.dir = Some(work_path.join("cache").to_string_lossy().to_string()), + Some(work_dir) => { + let mut cache_dir = work_dir.to_string(); + if PathBuf::from(&cache_dir).is_relative() { + cache_dir = work_path.join(&cache_dir).clean().to_string_lossy().to_string(); + } + self.dir = Some(cache_dir.to_string()); } } if let Some(val) = self.size.as_ref() { diff --git a/shared/src/model/config/video_download.rs b/shared/src/model/config/video_download.rs index 5498529ff..2353211f3 100644 --- a/shared/src/model/config/video_download.rs +++ b/shared/src/model/config/video_download.rs @@ -4,6 +4,8 @@ use crate::create_tuliprox_error_result; use crate::error::{TuliproxError, TuliproxErrorKind}; use crate::model::DEFAULT_USER_AGENT; +pub const DEFAULT_VIDEO_EXTENSIONS: [&str; 6] = ["mkv", "avi", "mp4", "mpeg", "divx", "mov"]; + #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default, PartialEq)] #[serde(deny_unknown_fields)] pub struct VideoDownloadConfigDto { @@ -35,7 +37,7 @@ impl VideoConfigDto { /// Will panic if default `RegEx` gets invalid pub fn prepare(&mut self) -> Result<(), TuliproxError> { if self.extensions.is_empty() { - self.extensions = ["mkv", "avi", "mp4", "mpeg", "divx", "mov"] + self.extensions = DEFAULT_VIDEO_EXTENSIONS .iter() .map(|&arg| arg.to_string()) .collect(); diff --git a/shared/src/utils/directed_graph.rs b/shared/src/utils/directed_graph.rs index 04b2369c3..be28fc742 100644 --- a/shared/src/utils/directed_graph.rs +++ b/shared/src/utils/directed_graph.rs @@ -71,9 +71,10 @@ where self.dfs_find_cycles(neighbor, visited, recursion_stack, cycles); } else if recursion_stack.contains(neighbor) { // Cycle detected; collect the cycle path - let cycle_start_index = recursion_stack.iter().position(|n| n == neighbor).unwrap(); - let cycle = recursion_stack[cycle_start_index..].to_vec(); - cycles.push(cycle); + if let Some(cycle_start_index) = recursion_stack.iter().position(|n| n == neighbor) { + let cycle = recursion_stack[cycle_start_index..].to_vec(); + cycles.push(cycle); + } } } } diff --git a/webui/src/app/components/dashboard/stats_view.rs b/webui/src/app/components/dashboard/stats_view.rs index f1bb436ec..4f1464b45 100644 --- a/webui/src/app/components/dashboard/stats_view.rs +++ b/webui/src/app/components/dashboard/stats_view.rs @@ -19,7 +19,7 @@ pub fn StatsView() -> Html { match &status_ctx.status { Some(stats) => { if let Some(map) = &stats.active_provider_connections { - if map.len() > 0 { + if !map.is_empty() { let cards = map.iter().map(|(provider, connections)| { html! { diff --git a/webui/src/app/components/home.rs b/webui/src/app/components/home.rs index 40f708c09..5ea1488d8 100644 --- a/webui/src/app/components/home.rs +++ b/webui/src/app/components/home.rs @@ -93,12 +93,10 @@ pub fn Home() -> Html { } else { treemap.insert(provider, connections); } - } else { - if connections > 0 { - let mut treemap = BTreeMap::new(); - treemap.insert(provider, connections); - server_status.active_provider_connections = Some(treemap); - } + } else if connections > 0 { + let mut treemap = BTreeMap::new(); + treemap.insert(provider, connections); + server_status.active_provider_connections = Some(treemap); } let new_status = Rc::new(server_status); *status_holder_signal.borrow_mut() = Some(Rc::clone(&new_status)); diff --git a/webui/src/services/requests.rs b/webui/src/services/requests.rs index a677e0b91..751229026 100644 --- a/webui/src/services/requests.rs +++ b/webui/src/services/requests.rs @@ -48,7 +48,8 @@ where match response.status() { 200 => { if std::any::TypeId::of::() == std::any::TypeId::of::<()>() { - Ok(serde_json::from_str("null").unwrap()) // `T = ()` valid + // `T = ()` valid + serde_json::from_str("null").map_err(|_| Error::DeserializeError) } else { let data: Result = response.json::().await; if let Ok(data) = data { diff --git a/webui/src/services/websocket_service.rs b/webui/src/services/websocket_service.rs index 677074fe8..571aa6205 100644 --- a/webui/src/services/websocket_service.rs +++ b/webui/src/services/websocket_service.rs @@ -19,12 +19,14 @@ pub enum WsMessage { const WS_PATH: &str = "/ws"; +type Subscriber = RefCell>>; + pub struct WebSocketService { connected: Rc, ws: Rc>>, status_service: Rc, subscriber_id: Rc, - subscribers: Rc>>>, + subscribers: Rc, } impl WebSocketService { @@ -73,7 +75,7 @@ impl WebSocketService { // onmessage let onmessage_callback = Closure::::wrap(Box::new(move |event: MessageEvent| { - trace!("WebSocket received message: {:?}", event); + trace!("WebSocket received message: {event:?}"); if let Ok(buf) = event.data().dyn_into::() { let array = Uint8Array::new(&buf); let bytes = bytes::Bytes::from(array.to_vec());