From 7ebf810836e75e7dd4cbaac8dbc595c06ea6fe69 Mon Sep 17 00:00:00 2001 From: euzu <33094714+euzu@users.noreply.github.com> Date: Fri, 27 Mar 2026 23:25:26 +0100 Subject: [PATCH] refactored stream table (#674) * refactored stream table --- backend/src/api/endpoints/v1_api_playlist.rs | 37 +- frontend/public/assets/i18n/en.json | 7 +- .../components/dashboard/_streams_view.scss | 226 ++++- frontend/src/app/components/authentication.rs | 2 +- frontend/src/app/components/dashboard/mod.rs | 4 +- .../dashboard/stream_display/helpers.rs | 279 ++++++ .../dashboard/stream_display/item.rs | 124 +++ .../dashboard/stream_display/meter.rs | 97 ++ .../dashboard/stream_display/mod.rs | 327 +++++++ .../app/components/dashboard/streams_table.rs | 869 ------------------ .../app/components/dashboard/streams_view.rs | 4 +- shared/src/model/stream_info.rs | 39 + 12 files changed, 1100 insertions(+), 915 deletions(-) create mode 100644 frontend/src/app/components/dashboard/stream_display/helpers.rs create mode 100644 frontend/src/app/components/dashboard/stream_display/item.rs create mode 100644 frontend/src/app/components/dashboard/stream_display/meter.rs create mode 100644 frontend/src/app/components/dashboard/stream_display/mod.rs delete mode 100644 frontend/src/app/components/dashboard/streams_table.rs diff --git a/backend/src/api/endpoints/v1_api_playlist.rs b/backend/src/api/endpoints/v1_api_playlist.rs index 038f836a9..4f19bf39a 100644 --- a/backend/src/api/endpoints/v1_api_playlist.rs +++ b/backend/src/api/endpoints/v1_api_playlist.rs @@ -100,6 +100,21 @@ fn resolve_provider_url_for_request(app_config: &AppConfig, playlist_request: &P } } +fn build_playlist_webplayer_url( + base_url: &str, + access_token: &str, + target_id: u16, + virtual_id: u32, + cluster: XtreamCluster, +) -> String { + format!( + "{base_url}/token/{access_token}/{}/{}/{}", + target_id, + cluster.as_stream_type(), + virtual_id + ) +} + async fn playlist_update( axum::extract::State(app_state): axum::extract::State>, axum::extract::Json(targets): axum::extract::Json>, @@ -268,13 +283,7 @@ fn playlist_webplayer( .map_or("default", |server_name| server_name.as_str()); let server_info = app_state.app_config.get_server_info(server_name); let base_url = server_info.get_base_url(); - format!( - "{base_url}/token/{access_token}/{}/{}/{}", - target_id, - cluster.as_stream_type(), - virtual_id - ) - .into_response() + build_playlist_webplayer_url(&base_url, &access_token, target_id, virtual_id, cluster).into_response() } async fn playlist_epg( @@ -398,7 +407,7 @@ mod tests { use arc_swap::{ArcSwap, ArcSwapOption}; use shared::foundation::Filter; use shared::{ - model::{ConfigPaths, ConfigProviderDto, PlaylistRequest}, + model::{ConfigPaths, ConfigProviderDto, PlaylistRequest, XtreamCluster}, utils::Internable, }; use std::sync::Arc; @@ -642,6 +651,18 @@ mod tests { assert_eq!(resolved, original); } + + #[test] + fn build_playlist_webplayer_url_uses_cluster_stream_type() { + let live = super::build_playlist_webplayer_url("http://player.example", "token123", 1, 42, XtreamCluster::Live); + let movie = super::build_playlist_webplayer_url("http://player.example", "token123", 1, 42, XtreamCluster::Video); + let series = + super::build_playlist_webplayer_url("http://player.example", "token123", 1, 42, XtreamCluster::Series); + + assert_eq!(live, "http://player.example/token/token123/1/live/42"); + assert_eq!(movie, "http://player.example/token/token123/1/movie/42"); + assert_eq!(series, "http://player.example/token/token123/1/series/42"); + } } async fn playlist_episode_item( diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 0e55430b4..6a7f60e01 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -273,7 +273,7 @@ "CACHE_CONFIG": { "DIR": "Filesystem directory where playlist and metadata caches are stored.", "ENABLED": "Globally enables or disables disk caching for input resources.", - "SIZE": "Maximum size (in bytes) allowed for the cache directory." + "SIZE": "Maximum size (in bytes) allowed for the cache directory. Allowed units are: B, KB, MB, GB, TB" }, "CONFIG_API": { "HOST": "Host where the tuliprox API listens. Default: 0.0.0.0", @@ -1234,7 +1234,9 @@ "XTREAM_CODES": "xtream codes", "X_HEADER": "Remove X-* headers", "YES": "Yes", - "FORCE": "Force" + "FORCE": "Force", + "ACTIVE_STREAMS": "Active Streams", + "PLAYER": "Player" }, "MESSAGES": { "CLIPBOARD_NOT_SUPPORTED": "Clipboard not supported.\nYour browser or current context does not allow clipboard access.\nPlease use HTTPS or localhost.", @@ -1260,6 +1262,7 @@ } }, "FAILED_TO_KICK_USER_STREAM": "Failed to kick user stream", + "FAILED_TO_RETRIEVE_PROVIDER_URL": "Failed to retrieve provider URL", "FAILED_TO_RETRIEVE_WEBPLAYER_URL": "Failed to retrieve webplayer URL", "INVALID_FILETYPE": "Invalid filetype", "INVALID_REGEXP": "Invalid regexp", diff --git a/frontend/scss/app/components/dashboard/_streams_view.scss b/frontend/scss/app/components/dashboard/_streams_view.scss index 6e00aab05..64e494143 100644 --- a/frontend/scss/app/components/dashboard/_streams_view.scss +++ b/frontend/scss/app/components/dashboard/_streams_view.scss @@ -1,5 +1,4 @@ -@use 'sass:list'; -@use '../../../size' as size; +@use "../../../size"; .tp__streams { display: flex; @@ -7,7 +6,6 @@ flex: 1 1 auto; box-sizing: border-box; width: 100%; - max-width: var(--max-view-width); overflow: hidden; &__header { @@ -23,23 +21,186 @@ } } -.tp__stream-table__duration { - letter-spacing: 2px; -} - -.tp__streams-table { +.tp__stream-display { display: flex; flex-flow: column; - overflow: auto; + flex: 1 1 auto; + min-height: 0; + overflow: hidden; + border: 1px solid var(--card-border-color); + border-radius: var(--border-radius); + background: var(--card-background-color); + box-shadow: var(--card-shadow); - &__tech-chips { + + &__header { + font-size: 1.1rem; + color: var(--modest-text-color); + padding: var(--padding-default) var(--padding-default) 0 var(--padding-default); + } + + &__body { + display: flex; + flex: 1 1 auto; + overflow: hidden; + padding: var(--padding-default); + } + + &__list { + display: flex; + flex: 1 1 auto; + flex-flow: column; + gap: var(--gap-small); + min-height: 0; + overflow: auto; + padding-right: var(--padding-small); + } + + &__item { + display: flex; + flex-flow: row nowrap; + gap: var(--gap-small); + padding: var(--padding-default); + border: 1px solid var(--card-border-color); + border-radius: var(--border-radius); + background: var(--card-background-color); + + &-content { + display: flex; + flex-flow: column; + gap: var(--gap-small); + width: 100%; + } + + &-head { + display: flex; + gap: var(--gap-default); + justify-content: space-between; + align-items: flex-start; + } + } + + &__identity { + display: flex; + flex: 1 1 auto; + flex-flow: column; + gap: var(--gap-small); + min-width: 0; + } + + &__title-row { + display: flex; + gap: var(--gap-small); + align-items: flex-start; + } + + &__menu { + display: inline-flex; + flex: 0 0 auto; + align-items: center; + justify-content: center; + width: 2rem; + height: 2rem; + border: 0; + border-radius: 50%; + background: transparent; + color: var(--text-color); + fill: var(--text-color); + cursor: pointer; + &:hover { + background: var(--menu-background-active-color); + color: var(--text-color); + } + } + + &__title-block { + display: flex; + flex: 1 1 auto; + flex-flow: column; + gap: var(--gap-default); + min-width: 0; + margin-right: var(--margin-default); + } + + &__title { + font-weight: 700; + line-height: 1.2; + min-width: 0; + overflow-wrap: anywhere; + } + + &__subtitle { + display: flex; + flex-flow: row nowrap; + gap: var(--gap-default); + } + + &__username { + font-weight: bold; + color: var(--stream-tech-chip-video-codec-color); + } + + &__channel_type { + font-weight: bold; + color: var(--stream-tech-chip-audio-codec-color); + } + + &__provider { + font-weight: bold; + color: var(--stream-tech-chip-container-color); + } + + &__stats { + display: flex; + flex-flow: row wrap; + flex: 0 0 auto; + gap: var(--gap-small); + } + + &__stat { + display: flex; + flex-flow: column; + gap: var(--gap-small); + border-radius: var(--border-radius); + } + + &__stat-label, + &__detail-label { + color: var(--tertiary-title-text-color); + font-size: 0.72rem; + font-weight: 700; + text-transform: uppercase; + } + + &__stat-value, + &__detail-value { + font-weight: 600; + } + + &__stat, &__detail { + min-width: 90px; + width: 90px; + } + + &__duration { + letter-spacing: 0.12em; + } + + &__row { + display: flex; + flex-flow: row wrap; + gap: var(--gap-default); + justify-content: space-between; + } + + &__chips { display: flex; flex-wrap: wrap; gap: var(--gap-mini); } - &__tech-chip { - font-weight: bold; + &__chip { + font-weight: 700; padding: var(--padding-micro) var(--padding-small) !important; &--container { @@ -79,26 +240,29 @@ } } - .tp__table__th { - $column-widths: - 40px, /*index*/ - 100px, /*username*/ - 120px, /*stream_id*/ - 80px, /*type*/ - 200px, /*channel*/ - 120px, /*group*/ - 260px, /*tech*/ - 90px, /*client ip*/ - 80px, /*country*/ - 90px, /*provider*/ - 80px, /*shared*/ - 120px, /*player*/ - 100px; /* duration */ + &__details { + display: flex; + flex-flow: row; + gap: var(--gap-small); + justify-content: flex-end; + } - @for $i from 1 through list.length($column-widths) { - &--#{$i} { - @include size.fixed-width(list.nth($column-widths, $i)); - } + &__detail { + display: flex; + flex-flow: column; + gap: var(--gap-small); + min-width: 0; + + } + + @media (max-width: size.$dashboard-mobile-breakpoint) { + &__item-head { + flex-flow: column; + } + + &__stats { + width: 100%; + min-width: 0; } } } diff --git a/frontend/src/app/components/authentication.rs b/frontend/src/app/components/authentication.rs index 08a54eefa..cc52efdef 100644 --- a/frontend/src/app/components/authentication.rs +++ b/frontend/src/app/components/authentication.rs @@ -66,7 +66,7 @@ pub fn Authentication(props: &AuthenticationProps) -> Html { use_effect_with((*loading, *authenticated), move |(loading, authenticated)| { if !*loading && !*authenticated { if let Some(navigator) = navigator.clone() { - navigator.replace(&AppRoute::Login); + navigator.replace(&AppRoute::Home); } } || () diff --git a/frontend/src/app/components/dashboard/mod.rs b/frontend/src/app/components/dashboard/mod.rs index 0925f8fed..667cf430f 100644 --- a/frontend/src/app/components/dashboard/mod.rs +++ b/frontend/src/app/components/dashboard/mod.rs @@ -10,11 +10,11 @@ mod status_card; mod user_action_card; mod version_action_card; -mod streams_table; +mod stream_display; mod streams_view; pub use self::{ action_card::*, dashboard_view::*, discord_action_card::*, documentation_action_card::*, github_action_card::*, - ipinfo_action_card::*, playlist_progress_status_card::*, stats_view::*, status_card::*, streams_table::*, + ipinfo_action_card::*, playlist_progress_status_card::*, stats_view::*, status_card::*, stream_display::*, streams_view::*, user_action_card::*, version_action_card::*, }; diff --git a/frontend/src/app/components/dashboard/stream_display/helpers.rs b/frontend/src/app/components/dashboard/stream_display/helpers.rs new file mode 100644 index 000000000..c2743dd9d --- /dev/null +++ b/frontend/src/app/components/dashboard/stream_display/helpers.rs @@ -0,0 +1,279 @@ +use crate::app::ConfigContext; +use gloo_utils::window; +use shared::{ + model::{PlaylistItemType, StreamChannel, StreamInfo, StreamTechnicalInfo}, + utils::{current_time_secs, default_hls_session_ttl_secs}, +}; +use std::collections::{HashMap, HashSet}; +use wasm_bindgen::{JsCast, JsValue}; +use web_sys::{console, Element}; +use yew::UseStateHandle; + +const LIVE: &str = "Live"; +const MOVIE: &str = "Movie"; +const SERIES: &str = "Series"; +const CATCHUP: &str = "Archive"; +const HLS: &str = "HLS"; +const DASH: &str = "DASH"; + +pub const ADAPTIVE_STREAM_CLEANUP_BUFFER_SECS: u64 = 5; +pub const ADAPTIVE_STREAM_CLEANUP_INTERVAL_MILLIS: u32 = 5_000; + +pub fn is_stream_metrics_enabled(config_ctx: &ConfigContext) -> bool { + config_ctx + .config + .as_ref() + .and_then(|cfg| cfg.config.reverse_proxy.as_ref()) + .and_then(|reverse_proxy| reverse_proxy.stream.as_ref()) + .is_some_and(|stream| stream.metrics_enabled) +} + +pub fn get_adaptive_session_ttl_secs(config_ctx: &ConfigContext) -> u64 { + config_ctx + .config + .as_ref() + .and_then(|cfg| cfg.config.reverse_proxy.as_ref()) + .and_then(|reverse_proxy| reverse_proxy.stream.as_ref()) + .map_or_else(default_hls_session_ttl_secs, |stream| stream.hls_session_ttl_secs) +} + +pub fn is_adaptive_session_stream(stream: &StreamInfo) -> bool { + stream.session_token.is_some() && stream.channel.item_type.is_live_adaptive() +} + +pub fn filter_visible_streams( + streams: Option>>, + adaptive_last_seen: &HashMap, + now_secs: u64, + adaptive_ttl_secs: u64, +) -> Option>> { + streams.map(|streams| { + streams + .into_iter() + .filter(|stream| { + if !is_adaptive_session_stream(stream) { + return true; + } + + adaptive_last_seen.get(&stream.uid).is_none_or(|last_seen| { + now_secs.saturating_sub(*last_seen) + <= adaptive_ttl_secs.saturating_add(ADAPTIVE_STREAM_CLEANUP_BUFFER_SECS) + }) + }) + .collect() + }) +} + +pub fn refresh_adaptive_last_seen( + adaptive_last_seen: &UseStateHandle>, + streams: &Option>>, +) { + let now = current_time_secs(); + adaptive_last_seen.set(compute_adaptive_last_seen((**adaptive_last_seen).clone(), streams, now)); +} + +fn compute_adaptive_last_seen( + mut next: HashMap, + streams: &Option>>, + now: u64, +) -> HashMap { + if let Some(streams) = streams { + let adaptive_uids: HashSet = + streams.iter().filter(|stream| is_adaptive_session_stream(stream)).map(|stream| stream.uid).collect(); + + next.retain(|uid, _| adaptive_uids.contains(uid)); + + for stream in streams { + if is_adaptive_session_stream(stream) { + if !stream.preserved || !next.contains_key(&stream.uid) { + next.insert(stream.uid, now); + } + } else { + next.remove(&stream.uid); + } + } + } else { + next.clear(); + } + + next +} + +pub fn format_duration(seconds: u64) -> String { + let hours = seconds / 3600; + let minutes = (seconds % 3600) / 60; + let seconds = seconds % 60; + format!("{hours:02}:{minutes:02}:{seconds:02}") +} + +pub fn format_bandwidth(rate_kbps: u32) -> String { + if rate_kbps == 0 { + return "-".to_string(); + } + if rate_kbps >= 1_048_576 { + format!("{:.1} GB/s", f64::from(rate_kbps) / 1_048_576.0) + } else if rate_kbps >= 1024 { + format!("{:.1} MB/s", f64::from(rate_kbps) / 1024.0) + } else { + format!("{rate_kbps} KB/s") + } +} + +pub fn format_transferred(total_kb: u32) -> String { + if total_kb == 0 { + return "-".to_string(); + } + if total_kb >= 1_048_576 { + format!("{:.2} GB", f64::from(total_kb) / 1_048_576.0) + } else if total_kb >= 1024 { + format!("{:.1} MB", f64::from(total_kb) / 1024.0) + } else { + format!("{total_kb} KB") + } +} + +fn adaptive_tech_label(item_type: PlaylistItemType) -> Option<&'static str> { + match item_type { + PlaylistItemType::LiveHls => Some(HLS), + PlaylistItemType::LiveDash => Some(DASH), + _ => None, + } +} + +pub fn build_technical_chips( + item_type: PlaylistItemType, + technical: Option<&StreamTechnicalInfo>, +) -> Vec<(String, &'static str)> { + let mut chips = Vec::new(); + let Some(tech) = technical else { + if let Some(label) = adaptive_tech_label(item_type) { + chips.push((label.to_string(), "tp__stream-display__chip--container")); + } + return chips; + }; + + if let Some(label) = adaptive_tech_label(item_type) { + chips.push((label.to_string(), "tp__stream-display__chip--container")); + } + if !tech.container.is_empty() { + chips.push((tech.container.to_ascii_uppercase(), "tp__stream-display__chip--container")); + } + if !tech.video_codec.is_empty() { + chips.push((tech.video_codec.clone(), "tp__stream-display__chip--video-codec")); + } + if !tech.resolution.is_empty() { + chips.push((tech.resolution.clone(), "tp__stream-display__chip--resolution")); + } + if !tech.fps.is_empty() { + chips.push((format!("{} FPS", tech.fps), "tp__stream-display__chip--fps")); + } + if !tech.audio_codec.is_empty() { + chips.push((tech.audio_codec.clone(), "tp__stream-display__chip--audio-codec")); + } + if !tech.audio_channels.is_empty() { + chips.push((tech.audio_channels.clone(), "tp__stream-display__chip--audio-channels")); + } + + chips +} + +pub fn render_cluster(channel: &StreamChannel) -> &'static str { + match channel.item_type { + PlaylistItemType::LiveUnknown | PlaylistItemType::Live => LIVE, + PlaylistItemType::Video | PlaylistItemType::LocalVideo => MOVIE, + PlaylistItemType::Series + | PlaylistItemType::SeriesInfo + | PlaylistItemType::LocalSeries + | PlaylistItemType::LocalSeriesInfo => SERIES, + PlaylistItemType::Catchup => CATCHUP, + PlaylistItemType::LiveHls => HLS, + PlaylistItemType::LiveDash => DASH, + } +} + +pub fn update_timestamps() { + let window = window(); + let Some(document) = window.document() else { + return; + }; + let spans = match document.query_selector_all("span[data-ts]") { + Ok(spans) => spans, + Err(err) => { + console::error_1(&err); + return; + } + }; + for i in 0..spans.length() { + if let Some(node) = spans.item(i) { + let Ok(el) = node.dyn_into::() else { + console::error_1(&JsValue::from_str("failed to convert timestamp node to Element")); + continue; + }; + if let Some(ts_str) = el.get_attribute("data-ts") { + if let Ok(ts) = ts_str.parse::() { + el.set_inner_html(&format_duration(current_time_secs().saturating_sub(ts))); + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::compute_adaptive_last_seen; + use shared::{ + model::{PlaylistItemType, StreamChannel, StreamInfo, XtreamCluster}, + utils::Internable, + }; + use std::{ + collections::HashMap, + net::{IpAddr, Ipv4Addr, SocketAddr}, + rc::Rc, + }; + + fn test_stream(uid: u32, item_type: PlaylistItemType, preserved: bool, has_session: bool) -> Rc { + Rc::new(StreamInfo { + uid, + meter_uid: 0, + username: "user".to_string(), + channel: StreamChannel { + target_id: 1, + virtual_id: uid, + provider_id: 1, + item_type, + cluster: XtreamCluster::try_from(item_type).unwrap_or(XtreamCluster::Live), + group: "Group".intern(), + title: "Title".intern(), + url: "http://example.com/stream.m3u8".intern(), + shared: false, + technical: None, + }, + provider: "provider".to_string(), + addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 8080), + client_ip: "127.0.0.1:1234".to_string(), + user_agent: String::new(), + ts: 0, + country: None, + session_token: has_session.then(|| "session".to_string()), + preserved, + }) + } + + #[test] + fn compute_adaptive_last_seen_prunes_missing_uids() { + let existing = HashMap::from([(1, 100), (2, 200), (3, 300)]); + let streams = Some(vec![ + test_stream(2, PlaylistItemType::LiveHls, true, true), + test_stream(4, PlaylistItemType::LiveDash, false, true), + test_stream(9, PlaylistItemType::Live, false, false), + ]); + + let refreshed = compute_adaptive_last_seen(existing, &streams, 999); + + assert_eq!(refreshed.get(&2), Some(&200)); + assert_eq!(refreshed.get(&4), Some(&999)); + assert!(!refreshed.contains_key(&1)); + assert!(!refreshed.contains_key(&3)); + assert!(!refreshed.contains_key(&9)); + } +} diff --git a/frontend/src/app/components/dashboard/stream_display/item.rs b/frontend/src/app/components/dashboard/stream_display/item.rs new file mode 100644 index 000000000..12be83fa1 --- /dev/null +++ b/frontend/src/app/components/dashboard/stream_display/item.rs @@ -0,0 +1,124 @@ +use super::{ + helpers::{build_technical_chips, format_duration, render_cluster}, + meter::{MeterDisplayKind, StreamMeterBadge}, +}; +use crate::{ + app::components::{AppIcon, Chip, RevealContent, ToggleSwitch}, + i18n::use_translation, + utils::t_safe, +}; +use shared::{ + model::StreamInfo, + utils::{current_time_secs, strip_port}, +}; +use std::rc::Rc; +use web_sys::MouseEvent; +use yew::prelude::*; + +#[derive(Properties, PartialEq, Clone)] +pub struct StreamDisplayItemProps { + pub stream: Rc, + pub metrics_enabled: bool, + pub on_popup_click: Callback<(Rc, MouseEvent)>, +} + +#[component] +pub fn StreamDisplayItem(props: &StreamDisplayItemProps) -> Html { + let translate = use_translation(); + let stream = props.stream.clone(); + let chips = build_technical_chips(stream.channel.item_type, stream.channel.technical.as_ref()); + let country = stream + .country + .as_ref() + .map_or_else(String::new, |c| t_safe(&translate, &format!("COUNTRY.{c}")).unwrap_or_else(|| c.to_string())); + let client_ip = strip_port(&stream.client_ip).to_string(); + + let handle_popup_click = { + let stream = stream.clone(); + let on_popup_click = props.on_popup_click.clone(); + Callback::from(move |event: MouseEvent| on_popup_click.emit((stream.clone(), event))) + }; + + html! { +
+ +
+
+
+
+
+
{stream.channel.title.to_string()}
+
+ {stream.username.clone()} + {" • "} + {render_cluster(&stream.channel)} + {" • "} + {stream.provider.clone()} + if !country.is_empty() { + <> + {" • "} + {country} + + } +
+
+
+
+
+
+ {translate.t("LABEL.CLIENT_IP")} + {client_ip.clone()} +
+
+ {translate.t("LABEL.SHARED")} + +
+
+ {translate.t("LABEL.DURATION")} + + {format_duration(current_time_secs().saturating_sub(stream.ts))} + +
+ if props.metrics_enabled { +
+ {translate.t("LABEL.BANDWIDTH")} + + + +
+
+ {translate.t("LABEL.TRANSFERRED")} + + + +
+ } +
+
+
+ if !chips.is_empty() { +
+ { for chips.into_iter().map(|(label, chip_class)| html! { + + })} +
+ } +
+
+ {translate.t("LABEL.GROUP")} + {stream.channel.group.to_string()} +
+
+ {translate.t("LABEL.PLAYER")} + + {&stream.user_agent} + +
+
+
+
+
+ } +} diff --git a/frontend/src/app/components/dashboard/stream_display/meter.rs b/frontend/src/app/components/dashboard/stream_display/meter.rs new file mode 100644 index 000000000..7d1dbf853 --- /dev/null +++ b/frontend/src/app/components/dashboard/stream_display/meter.rs @@ -0,0 +1,97 @@ +use super::helpers::{format_bandwidth, format_transferred}; +use crate::{hooks::use_service_context, model::EventMessage}; +use shared::model::StreamMeterEntry; +use yew::prelude::*; + +#[derive(Clone, Copy, PartialEq, Eq)] +pub enum MeterDisplayKind { + Bandwidth, + Transferred, +} + +#[derive(Properties, PartialEq)] +pub struct StreamMeterBadgeProps { + pub uid: u32, + pub meter_uid: u32, + pub kind: MeterDisplayKind, +} + +#[derive(Clone, PartialEq, Eq, Default)] +struct StreamMeterBadgeState { + rate_kbps: u32, + transferred_total_kb: u32, + current_meter_uid: u32, + current_meter_total_kb: u32, +} + +fn apply_stream_meter_entry( + state: &StreamMeterBadgeState, + current_meter_uid: u32, + entry: &StreamMeterEntry, +) -> StreamMeterBadgeState { + let mut next = state.clone(); + if entry.meter_uid == current_meter_uid { + if next.current_meter_uid != current_meter_uid { + next.transferred_total_kb = next.transferred_total_kb.saturating_add(next.current_meter_total_kb); + next.current_meter_uid = current_meter_uid; + next.current_meter_total_kb = 0; + } + next.current_meter_total_kb = next.current_meter_total_kb.max(entry.total_kb); + next.rate_kbps = entry.rate_kbps; + } else { + next.transferred_total_kb = next.transferred_total_kb.saturating_add(entry.total_kb); + } + + next +} + +#[component] +pub fn StreamMeterBadge(props: &StreamMeterBadgeProps) -> Html { + let services = use_service_context(); + let meter_state = use_state(StreamMeterBadgeState::default); + + { + let meter_state = meter_state.clone(); + let reset_key = props.uid; + use_effect_with(reset_key, move |_| { + meter_state.set(StreamMeterBadgeState::default()); + || () + }); + } + + { + let services = services.clone(); + let meter_state = meter_state.clone(); + let listen_key = (props.uid, props.meter_uid); + use_effect_with(listen_key, move |(uid, meter_uid)| { + let uid = *uid; + let meter_uid = *meter_uid; + let subid = services.event.subscribe(move |msg| { + if let EventMessage::StreamMeterBatch(entries) = msg { + let mut next_state = (*meter_state).clone(); + let mut changed = false; + for entry in entries.iter().filter(|entry| entry.uids.contains(&uid)) { + let updated = apply_stream_meter_entry(&next_state, meter_uid, entry); + if updated != next_state { + next_state = updated; + changed = true; + } + } + if changed { + meter_state.set(next_state); + } + } + }); + move || services.event.unsubscribe(subid) + }); + } + + let label = match props.kind { + MeterDisplayKind::Bandwidth => format_bandwidth(meter_state.rate_kbps), + MeterDisplayKind::Transferred => { + let total_kb = meter_state.transferred_total_kb.saturating_add(meter_state.current_meter_total_kb); + format_transferred(total_kb) + } + }; + html! { {label} } +} diff --git a/frontend/src/app/components/dashboard/stream_display/mod.rs b/frontend/src/app/components/dashboard/stream_display/mod.rs new file mode 100644 index 000000000..6a6f7358d --- /dev/null +++ b/frontend/src/app/components/dashboard/stream_display/mod.rs @@ -0,0 +1,327 @@ +mod helpers; +mod item; +mod meter; + +use self::{ + helpers::{ + filter_visible_streams, get_adaptive_session_ttl_secs, is_stream_metrics_enabled, refresh_adaptive_last_seen, + update_timestamps, ADAPTIVE_STREAM_CLEANUP_INTERVAL_MILLIS, + }, + item::StreamDisplayItem, +}; +use crate::{ + app::{ + components::{menu_item::MenuItem, popup_menu::PopupMenu, NoContent}, + ConfigContext, + }, + hooks::use_service_context, + i18n::use_translation, + model::EventMessage, + services::DialogService, +}; +use gloo_timers::callback::Interval; +use shared::{ + error::{info_err_res, TuliproxError}, + model::{PlaylistRequest, PlaylistUrlResolveRequest, ProtocolMessage, StreamInfo, UserCommand}, + utils::default_kick_secs, +}; +use std::{collections::HashMap, fmt::Display, rc::Rc, str::FromStr}; +use wasm_bindgen::JsCast; +use yew::{platform::spawn_local, prelude::*}; +use yew_hooks::use_clipboard; + +const KICK: &str = "kick"; +const COPY_LINK_TULIPROX_VIRTUAL_ID: &str = "copy_link_tuliprox_virtual_id"; +const COPY_LINK_TULIPROX_WEBPLAYER_URL: &str = "copy_link_tuliprox_webplayer_url"; +const COPY_LINK_PROVIDER_URL: &str = "copy_link_provider_url"; + +#[derive(Properties, PartialEq, Clone)] +pub struct StreamDisplayProps { + pub streams: Option>>, +} + +#[component] +pub fn StreamDisplay(props: &StreamDisplayProps) -> Html { + let translate = use_translation(); + let service_ctx = use_service_context(); + let dialog = use_context::().expect("Dialog service not found"); + let clipboard = use_clipboard(); + let config_ctx = use_context::().expect("Config context not found"); + let popup_anchor_ref = use_state(|| None::); + let popup_is_open = use_state(|| false); + let selected_dto = use_state(|| None::>); + let adaptive_last_seen = use_state(HashMap::::new); + let cleanup_now_secs = use_state(shared::utils::current_time_secs); + let adaptive_session_ttl_secs = get_adaptive_session_ttl_secs(&config_ctx); + let metrics_enabled = is_stream_metrics_enabled(&config_ctx); + + use_effect_with((), move |_| { + let interval = Interval::new(1000, update_timestamps); + move || drop(interval) + }); + + { + let adaptive_last_seen = adaptive_last_seen.clone(); + let streams = props.streams.clone(); + use_effect_with(streams, move |streams| { + refresh_adaptive_last_seen(&adaptive_last_seen, streams); + || () + }); + } + + { + let cleanup_now_secs = cleanup_now_secs.clone(); + use_effect_with((), move |_| { + let interval = Interval::new(ADAPTIVE_STREAM_CLEANUP_INTERVAL_MILLIS, move || { + cleanup_now_secs.set(shared::utils::current_time_secs()); + }); + move || drop(interval) + }); + } + + { + let websocket = service_ctx.websocket.clone(); + let event_service = service_ctx.event.clone(); + use_effect_with(metrics_enabled, move |metrics_enabled| { + let subid = if *metrics_enabled { + websocket.send_message(ProtocolMessage::StreamMeterSubscribe); + let websocket_for_events = websocket.clone(); + Some(event_service.subscribe(move |msg| { + if let EventMessage::WebSocketStatus(true) = msg { + websocket_for_events.send_message(ProtocolMessage::StreamMeterSubscribe); + } + })) + } else { + None + }; + + move || { + if let Some(subid) = subid { + event_service.unsubscribe(subid); + websocket.send_message(ProtocolMessage::StreamMeterUnsubscribe); + } + } + }); + } + + let visible_streams = use_memo( + (props.streams.clone(), (*adaptive_last_seen).clone(), *cleanup_now_secs, adaptive_session_ttl_secs), + |(streams, adaptive_last_seen, cleanup_now_secs, adaptive_session_ttl_secs)| { + filter_visible_streams(streams.clone(), adaptive_last_seen, *cleanup_now_secs, *adaptive_session_ttl_secs) + }, + ); + + { + let popup_is_open = popup_is_open.clone(); + let popup_anchor_ref = popup_anchor_ref.clone(); + let selected_dto = selected_dto.clone(); + let visible_streams_dep = (*visible_streams).clone(); + use_effect_with( + (visible_streams_dep, (*selected_dto).as_ref().map(|stream| stream.uid)), + move |(visible_streams, selected_uid)| { + let selected_uid = *selected_uid; + let is_selected_visible = selected_uid.is_none_or(|uid| { + visible_streams.as_ref().is_some_and(|streams| streams.iter().any(|stream| stream.uid == uid)) + }); + + if !is_selected_visible { + popup_is_open.set(false); + popup_anchor_ref.set(None); + selected_dto.set(None); + } + + || () + }, + ); + } + + let handle_popup_close = { + let set_is_open = popup_is_open.clone(); + Callback::from(move |()| set_is_open.set(false)) + }; + + let handle_popup_onclick = { + let set_selected_dto = selected_dto.clone(); + let set_anchor_ref = popup_anchor_ref.clone(); + let set_is_open = popup_is_open.clone(); + Callback::from(move |(dto, event): (Rc, MouseEvent)| { + if let Some(streams) = event.current_target().and_then(|target| target.dyn_into::().ok()) + { + set_selected_dto.set(Some(dto)); + set_anchor_ref.set(Some(streams)); + set_is_open.set(true); + } + }) + }; + + let copy_to_clipboard: Callback = { + let clipboard = clipboard.clone(); + let dialog = dialog.clone(); + Callback::from(move |text: String| { + if *clipboard.is_supported { + clipboard.write_text(text); + } else { + let dlg = dialog.clone(); + spawn_local(async move { + let _ = dlg + .content(html! {}, None, false) + .await; + }); + } + }) + }; + + let handle_menu_click = { + let popup_is_open_state = popup_is_open.clone(); + let translate = translate.clone(); + let services = service_ctx.clone(); + let selected_dto = selected_dto.clone(); + let copy_to_clipboard = copy_to_clipboard.clone(); + let kick_secs = config_ctx + .config + .as_ref() + .and_then(|app_cfg| app_cfg.config.web_ui.as_ref()) + .map(|web_ui| web_ui.kick_secs) + .unwrap_or_else(default_kick_secs); + Callback::from(move |(name, _): (String, _)| { + if let Ok(action) = StreamDisplayAction::from_str(&name) { + match action { + StreamDisplayAction::Kick => { + if let Some(dto) = (*selected_dto).as_ref() { + if !services.websocket.send_message(ProtocolMessage::UserAction(UserCommand::Kick( + dto.addr, + dto.channel.virtual_id, + kick_secs, + ))) { + services.toastr.error(translate.t("MESSAGES.FAILED_TO_KICK_USER_STREAM")); + } + } + } + StreamDisplayAction::CopyLinkTuliproxVirtualId => { + if let Some(dto) = &*selected_dto { + copy_to_clipboard.emit(dto.channel.virtual_id.to_string()); + } + } + StreamDisplayAction::CopyLinkProviderUrl => { + if let Some(dto) = &*selected_dto { + let url = dto.channel.url.to_string(); + let playlist_request = PlaylistRequest::Target(dto.channel.target_id); + let copy_to_clipboard = copy_to_clipboard.clone(); + let services = services.clone(); + let translate = translate.clone(); + spawn_local(async move { + let request = PlaylistUrlResolveRequest::Provider { playlist_request, url }; + if let Some(resolved) = services.playlist.resolve_url(request).await { + copy_to_clipboard.emit(resolved); + } else { + services.toastr.error(translate.t("MESSAGES.FAILED_TO_RETRIEVE_PROVIDER_URL")); + } + }); + } + } + StreamDisplayAction::CopyLinkTuliproxWebPlayerUrl => { + if let Some(dto) = &*selected_dto { + let target_id = dto.channel.target_id; + let virtual_id = dto.channel.virtual_id; + let cluster = dto.channel.cluster; + let services = services.clone(); + let translate = translate.clone(); + let copy_to_clipboard = copy_to_clipboard.clone(); + spawn_local(async move { + let request = PlaylistUrlResolveRequest::Webplayer { target_id, virtual_id, cluster }; + if let Some(url) = services.playlist.resolve_url(request).await { + copy_to_clipboard.emit(url); + } else { + services.toastr.error(translate.t("MESSAGES.FAILED_TO_RETRIEVE_WEBPLAYER_URL")); + } + }); + } + } + } + } + popup_is_open_state.set(false); + }) + }; + + html! { +
+
+ +
+
+ { + if let Some(streams) = visible_streams.as_ref() { + if streams.is_empty() { + html! { } + } else { + html! { + <> +
+ { for streams.iter().cloned().map(|stream| { + let uid = stream.uid; + html! { + + } + })} +
+ + + + + + + + } + } + } else { + html! { } + } + } +
+
+ } +} + +#[derive(Debug, Clone, Eq, PartialEq)] +enum StreamDisplayAction { + Kick, + CopyLinkTuliproxVirtualId, + CopyLinkTuliproxWebPlayerUrl, + CopyLinkProviderUrl, +} + +impl Display for StreamDisplayAction { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "{}", + match self { + Self::Kick => KICK, + Self::CopyLinkTuliproxVirtualId => COPY_LINK_TULIPROX_VIRTUAL_ID, + Self::CopyLinkTuliproxWebPlayerUrl => COPY_LINK_TULIPROX_WEBPLAYER_URL, + Self::CopyLinkProviderUrl => COPY_LINK_PROVIDER_URL, + } + ) + } +} + +impl FromStr for StreamDisplayAction { + type Err = TuliproxError; + + fn from_str(s: &str) -> Result { + match s { + KICK => Ok(Self::Kick), + COPY_LINK_TULIPROX_VIRTUAL_ID => Ok(Self::CopyLinkTuliproxVirtualId), + COPY_LINK_TULIPROX_WEBPLAYER_URL => Ok(Self::CopyLinkTuliproxWebPlayerUrl), + COPY_LINK_PROVIDER_URL => Ok(Self::CopyLinkProviderUrl), + _ => { + info_err_res!("Unknown Stream Action: {}", s) + } + } + } +} diff --git a/frontend/src/app/components/dashboard/streams_table.rs b/frontend/src/app/components/dashboard/streams_table.rs deleted file mode 100644 index 4427e276d..000000000 --- a/frontend/src/app/components/dashboard/streams_table.rs +++ /dev/null @@ -1,869 +0,0 @@ -use crate::{ - app::{ - components::{ - menu_item::MenuItem, popup_menu::PopupMenu, AppIcon, Chip, RevealContent, Table, TableDefinition, - ToggleSwitch, - }, - ConfigContext, - }, - hooks::use_service_context, - i18n::use_translation, - model::EventMessage, - services::DialogService, - utils::t_safe, -}; -use gloo_timers::callback::Interval; -use gloo_utils::window; -use shared::{ - concat_string, - error::{info_err_res, TuliproxError}, - model::{ - PlaylistItemType, PlaylistRequest, PlaylistUrlResolveRequest, ProtocolMessage, SortOrder, StreamChannel, - StreamInfo, StreamMeterEntry, StreamTechnicalInfo, UserCommand, - }, - utils::{current_time_secs, default_hls_session_ttl_secs, default_kick_secs, strip_port}, -}; -use std::{collections::HashMap, fmt::Display, rc::Rc, str::FromStr}; -use wasm_bindgen::JsCast; -use web_sys::Element; -use yew::{platform::spawn_local, prelude::*}; -use yew_hooks::use_clipboard; - -const LIVE: &str = "Live"; -const MOVIE: &str = "Movie"; -const SERIES: &str = "Series"; -const CATCHUP: &str = "Archive"; -const HLS: &str = "HLS"; -const DASH: &str = "DASH"; -const ADAPTIVE_STREAM_CLEANUP_BUFFER_SECS: u64 = 5; -const ADAPTIVE_STREAM_CLEANUP_INTERVAL_MILLIS: u32 = 5_000; - -const KICK: &str = "kick"; -const COPY_LINK_TULIPROX_VIRTUAL_ID: &str = "copy_link_tuliprox_virtual_id"; -const COPY_LINK_TULIPROX_WEBPLAYER_URL: &str = "copy_link_tuliprox_webplayer_url"; -const COPY_LINK_PROVIDER_URL: &str = "copy_link_provider_url"; - -const HEADERS: [&str; 15] = [ - "EMPTY", - "USERNAME", - "STREAM_ID", - "CLUSTER", - "CHANNEL", - "GROUP", - "TECH", - "CLIENT_IP", - "COUNTRY", - "PROVIDER", - "SHARED", - "USER_AGENT", - "DURATION", - "BANDWIDTH", - "TRANSFERRED", -]; - -fn is_stream_metrics_enabled(config_ctx: &ConfigContext) -> bool { - config_ctx - .config - .as_ref() - .and_then(|cfg| cfg.config.reverse_proxy.as_ref()) - .and_then(|reverse_proxy| reverse_proxy.stream.as_ref()) - .is_some_and(|stream| stream.metrics_enabled) -} - -fn get_adaptive_session_ttl_secs(config_ctx: &ConfigContext) -> u64 { - config_ctx - .config - .as_ref() - .and_then(|cfg| cfg.config.reverse_proxy.as_ref()) - .and_then(|reverse_proxy| reverse_proxy.stream.as_ref()) - .map_or_else(default_hls_session_ttl_secs, |stream| stream.hls_session_ttl_secs) -} - -fn is_adaptive_session_stream(stream: &StreamInfo) -> bool { - stream.session_token.is_some() && stream.channel.item_type.is_live_adaptive() -} - -fn filter_visible_streams( - streams: Option>>, - adaptive_last_seen: &HashMap, - now_secs: u64, - adaptive_ttl_secs: u64, -) -> Option>> { - streams.map(|streams| { - streams - .into_iter() - .filter(|stream| { - if !is_adaptive_session_stream(stream) { - return true; - } - - adaptive_last_seen.get(&stream.uid).is_none_or(|last_seen| { - now_secs.saturating_sub(*last_seen) - <= adaptive_ttl_secs.saturating_add(ADAPTIVE_STREAM_CLEANUP_BUFFER_SECS) - }) - }) - .collect() - }) -} - -fn format_duration(seconds: u64) -> String { - let hours = seconds / 3600; - let minutes = (seconds % 3600) / 60; - let seconds = seconds % 60; - format!("{hours:02}:{minutes:02}:{seconds:02}") -} - -fn format_bandwidth(rate_kbps: u32) -> String { - if rate_kbps == 0 { - return "-".to_string(); - } - if rate_kbps >= 1_048_576 { - format!("{:.1} GB/s", f64::from(rate_kbps) / 1_048_576.0) - } else if rate_kbps >= 1024 { - format!("{:.1} MB/s", f64::from(rate_kbps) / 1024.0) - } else { - format!("{rate_kbps} KB/s") - } -} - -fn format_transferred(total_kb: u32) -> String { - if total_kb == 0 { - return "-".to_string(); - } - if total_kb >= 1_048_576 { - format!("{:.2} GB", f64::from(total_kb) / 1_048_576.0) - } else if total_kb >= 1024 { - format!("{:.1} MB", f64::from(total_kb) / 1024.0) - } else { - format!("{total_kb} KB") - } -} - -#[derive(Clone, Copy, PartialEq, Eq)] -enum MeterDisplayKind { - Bandwidth, - Transferred, -} - -#[derive(Properties, PartialEq)] -struct StreamMeterCellProps { - pub uid: u32, - pub meter_uid: u32, - pub kind: MeterDisplayKind, -} - -#[derive(Clone, PartialEq, Eq, Default)] -struct StreamMeterCellState { - rate_kbps: u32, - transferred_total_kb: u32, - current_meter_uid: u32, - current_meter_total_kb: u32, -} - -fn apply_stream_meter_entry( - state: &StreamMeterCellState, - current_meter_uid: u32, - entry: &StreamMeterEntry, -) -> StreamMeterCellState { - let mut next = state.clone(); - if entry.meter_uid == current_meter_uid { - if next.current_meter_uid != current_meter_uid { - next.transferred_total_kb = next.transferred_total_kb.saturating_add(next.current_meter_total_kb); - next.current_meter_uid = current_meter_uid; - next.current_meter_total_kb = 0; - } - next.current_meter_total_kb = next.current_meter_total_kb.max(entry.total_kb); - next.rate_kbps = entry.rate_kbps; - } else { - next.transferred_total_kb = next.transferred_total_kb.saturating_add(entry.total_kb); - } - - next -} - -#[component] -fn StreamMeterCell(props: &StreamMeterCellProps) -> Html { - let services = use_service_context(); - let meter_state = use_state(StreamMeterCellState::default); - - { - let meter_state = meter_state.clone(); - let reset_key = props.uid; - use_effect_with(reset_key, move |_| { - meter_state.set(StreamMeterCellState::default()); - || () - }); - } - - { - let services = services.clone(); - let meter_state = meter_state.clone(); - let listen_key = (props.uid, props.meter_uid); - use_effect_with(listen_key, move |(uid, meter_uid)| { - let uid = *uid; - let meter_uid = *meter_uid; - let subid = services.event.subscribe(move |msg| { - if let EventMessage::StreamMeterBatch(entries) = msg { - let mut next_state = (*meter_state).clone(); - let mut changed = false; - for entry in entries.iter().filter(|entry| entry.uids.contains(&uid)) { - let updated = apply_stream_meter_entry(&next_state, meter_uid, entry); - if updated != next_state { - next_state = updated; - changed = true; - } - } - if changed { - meter_state.set(next_state); - } - } - }); - move || services.event.unsubscribe(subid) - }); - } - - let rate_kbps = meter_state.rate_kbps; - let total_kb = meter_state.transferred_total_kb.saturating_add(meter_state.current_meter_total_kb); - match props.kind { - MeterDisplayKind::Bandwidth => html! { format_bandwidth(rate_kbps) }, - MeterDisplayKind::Transferred => html! { format_transferred(total_kb) }, - } -} - -fn adaptive_tech_label(item_type: PlaylistItemType) -> Option<&'static str> { - match item_type { - PlaylistItemType::LiveHls => Some(HLS), - PlaylistItemType::LiveDash => Some(DASH), - _ => None, - } -} - -fn build_technical_chips( - item_type: PlaylistItemType, - technical: Option<&StreamTechnicalInfo>, -) -> Vec<(String, &'static str)> { - let mut chips = Vec::new(); - let Some(tech) = technical else { - if let Some(label) = adaptive_tech_label(item_type) { - chips.push((label.to_string(), "tp__streams-table__tech-chip--container")); - } - return chips; - }; - - if let Some(label) = adaptive_tech_label(item_type) { - chips.push((label.to_string(), "tp__streams-table__tech-chip--container")); - } - if !tech.container.is_empty() { - chips.push((tech.container.to_ascii_uppercase(), "tp__streams-table__tech-chip--container")); - } - if !tech.video_codec.is_empty() { - chips.push((tech.video_codec.clone(), "tp__streams-table__tech-chip--video-codec")); - } - if !tech.resolution.is_empty() { - chips.push((tech.resolution.clone(), "tp__streams-table__tech-chip--resolution")); - } - if !tech.fps.is_empty() { - chips.push((format!("{} fps", tech.fps), "tp__streams-table__tech-chip--fps")); - } - if !tech.audio_codec.is_empty() { - chips.push((tech.audio_codec.clone(), "tp__streams-table__tech-chip--audio-codec")); - } - if !tech.audio_channels.is_empty() { - chips.push((tech.audio_channels.clone(), "tp__streams-table__tech-chip--audio-channels")); - } - - chips -} - -fn update_timestamps() { - let window = window(); - let document = window.document().unwrap(); - let spans = document.query_selector_all("span[data-ts]").unwrap(); - for i in 0..spans.length() { - if let Some(node) = spans.item(i) { - let el: Element = node.dyn_into().unwrap(); - if let Some(ts_str) = el.get_attribute("data-ts") { - if let Ok(ts) = ts_str.parse::() { - el.set_inner_html(&format_duration(current_time_secs() - ts)); - } - } - } - } -} - -#[derive(Properties, PartialEq, Clone)] -pub struct StreamsTableProps { - pub streams: Option>>, -} - -#[component] -pub fn StreamsTable(props: &StreamsTableProps) -> Html { - let translate = use_translation(); - let service_ctx = use_service_context(); - let dialog = use_context::().expect("Dialog service not found"); - let clipboard = use_clipboard(); - let config_ctx = use_context::().expect("Config context not found"); - let popup_anchor_ref = use_state(|| None::); - let popup_is_open = use_state(|| false); - let selected_dto = use_state(|| None::>); - let adaptive_last_seen = use_state(HashMap::::new); - let cleanup_now_secs = use_state(current_time_secs); - let adaptive_session_ttl_secs = get_adaptive_session_ttl_secs(&config_ctx); - - let headers = use_memo(config_ctx.clone(), |cfg| { - let include_country = if let Some(app_cfg) = &cfg.config { app_cfg.config.is_geoip_enabled() } else { false }; - let metrics_enabled = is_stream_metrics_enabled(cfg); - - let visible_headers: Vec<&str> = if include_country { - HEADERS - .iter() - .filter(|h| metrics_enabled || (**h != "BANDWIDTH" && **h != "TRANSFERRED")) - .copied() - .collect() - } else { - HEADERS - .iter() - .filter(|h| **h != "COUNTRY") - .filter(|h| metrics_enabled || (**h != "BANDWIDTH" && **h != "TRANSFERRED")) - .copied() - .collect() - }; - visible_headers - }); - - use_effect_with((), move |_| { - let interval = Interval::new(1000, update_timestamps); - move || drop(interval) - }); - - { - let adaptive_last_seen = adaptive_last_seen.clone(); - let streams = props.streams.clone(); - use_effect_with(streams, move |streams| { - let now = current_time_secs(); - let mut next = (*adaptive_last_seen).clone(); - - if let Some(streams) = streams { - for stream in streams { - if is_adaptive_session_stream(stream) { - if !stream.preserved || !next.contains_key(&stream.uid) { - next.insert(stream.uid, now); - } - } else { - next.remove(&stream.uid); - } - } - } else { - next.clear(); - } - - adaptive_last_seen.set(next); - || () - }); - } - - { - let cleanup_now_secs = cleanup_now_secs.clone(); - use_effect_with((), move |_| { - let interval = Interval::new(ADAPTIVE_STREAM_CLEANUP_INTERVAL_MILLIS, move || { - cleanup_now_secs.set(current_time_secs()); - }); - move || drop(interval) - }); - } - - { - let metrics_enabled = is_stream_metrics_enabled(&config_ctx); - let websocket = service_ctx.websocket.clone(); - let event_service = service_ctx.event.clone(); - use_effect_with(metrics_enabled, move |metrics_enabled| { - let subid = if *metrics_enabled { - websocket.send_message(ProtocolMessage::StreamMeterSubscribe); - let websocket_for_events = websocket.clone(); - Some(event_service.subscribe(move |msg| { - if let EventMessage::WebSocketStatus(true) = msg { - websocket_for_events.send_message(ProtocolMessage::StreamMeterSubscribe); - } - })) - } else { - None - }; - - move || { - if let Some(subid) = subid { - event_service.unsubscribe(subid); - websocket.send_message(ProtocolMessage::StreamMeterUnsubscribe); - } - } - }); - } - - let handle_popup_close = { - let set_is_open = popup_is_open.clone(); - Callback::from(move |()| { - set_is_open.set(false); - }) - }; - - let handle_popup_onclick = { - let set_selected_dto = selected_dto.clone(); - let set_anchor_ref = popup_anchor_ref.clone(); - let set_is_open = popup_is_open.clone(); - Callback::from(move |(dto, event): (Rc, MouseEvent)| { - if let Some(streams) = event.target_dyn_into::() { - set_selected_dto.set(Some(dto.clone())); - set_anchor_ref.set(Some(streams)); - set_is_open.set(true); - } - }) - }; - - let render_header_cell = { - let translator = translate.clone(); - let headers = headers.clone(); - Callback::::from(move |col| { - html! { - { - if col < headers.len() { - translator.t(&concat_string!("LABEL.", headers[col])) - } else { - String::new() - } - } - } - }) - }; - - let render_cluster = |channel: &StreamChannel| -> &str { - match channel.item_type { - PlaylistItemType::LiveUnknown | PlaylistItemType::Live => LIVE, - PlaylistItemType::Video | PlaylistItemType::LocalVideo => MOVIE, - PlaylistItemType::Series - | PlaylistItemType::SeriesInfo - | PlaylistItemType::LocalSeries - | PlaylistItemType::LocalSeriesInfo => SERIES, - PlaylistItemType::Catchup => CATCHUP, - PlaylistItemType::LiveHls => HLS, - PlaylistItemType::LiveDash => DASH, - } - }; - - let render_data_cell = { - let popup_onclick = handle_popup_onclick.clone(); - let headers = headers.clone(); - let translate = translate.clone(); - Callback::<(usize, usize, Rc), Html>::from( - move |(row, col, dto): (usize, usize, Rc)| match headers[col] { - "EMPTY" => { - let popup_onclick = popup_onclick.clone(); - html! { - - } - } - "USERNAME" => html! {&dto.username}, - "STREAM_ID" => html! { <> - { dto.channel.virtual_id.to_string() } - {" ("} - { dto.channel.provider_id.to_string() } - {")"} - }, - "CLUSTER" => html! { render_cluster(&dto.channel) }, - "CHANNEL" => html! {&dto.channel.title}, - "GROUP" => html! {&*dto.channel.group}, - "TECH" => { - let chips = build_technical_chips(dto.channel.item_type, dto.channel.technical.as_ref()); - if chips.is_empty() { - html! {} - } else { - html! { -
- { for chips.into_iter().map(|(label, chip_class)| html! { - - })} -
- } - } - } - "CLIENT_IP" => html! { strip_port(&dto.client_ip)}, - "COUNTRY" => { - html! { dto.country.as_ref().map_or_else(String::new, |c| t_safe(&translate, &format!("COUNTRY.{c}")).unwrap_or_else(||c.to_string())) } - } - "PROVIDER" => html! {&dto.provider}, - "SHARED" => html! { }, - "USER_AGENT" => { - html! { {&dto.user_agent} } - } - "DURATION" => { - html! { {format_duration(current_time_secs() - dto.ts)} } - } - "BANDWIDTH" => { - html! { } - } - "TRANSFERRED" => { - html! { } - } - _ => html! {""}, - }, - ) - }; - - let is_sortable = Callback::::from(move |_col| false); - - let on_sort = Callback::, ()>::from(move |_args| {}); - - let table_definition = { - // first register for config update - let render_header_cell_cb = render_header_cell.clone(); - let render_data_cell_cb = render_data_cell.clone(); - let is_sortable = is_sortable.clone(); - let on_sort = on_sort.clone(); - let num_cols = headers.len(); - use_memo( - ( - props.streams.clone(), - (*headers).clone(), - (*adaptive_last_seen).clone(), - *cleanup_now_secs, - adaptive_session_ttl_secs, - ), - move |(streams, _, adaptive_last_seen, cleanup_now_secs, adaptive_session_ttl_secs)| { - filter_visible_streams( - streams.clone(), - adaptive_last_seen, - *cleanup_now_secs, - *adaptive_session_ttl_secs, - ) - .as_ref() - .map(|list| { - Rc::new(TableDefinition:: { - items: if list.is_empty() { None } else { Some(Rc::new(list.clone())) }, - num_cols, - is_sortable, - on_sort, - render_header_cell: render_header_cell_cb, - render_data_cell: render_data_cell_cb, - }) - }) - }, - ) - }; - - let copy_to_clipboard: Callback = { - let clipboard = clipboard.clone(); - let dialog = dialog.clone(); - Callback::from(move |text: String| { - if *clipboard.is_supported { - clipboard.write_text(text); - } else { - let dlg = dialog.clone(); - spawn_local(async move { - let _result = dlg - .content(html! {}, None, false) - .await; - }); - } - }) - }; - - let handle_menu_click = { - let popup_is_open_state = popup_is_open.clone(); - let translate = translate.clone(); - let services = service_ctx.clone(); - let selected_dto = selected_dto.clone(); - let copy_to_clipboard = copy_to_clipboard.clone(); - let kick_secs = config_ctx - .config - .as_ref() - .and_then(|app_cfg| app_cfg.config.web_ui.as_ref()) - .map(|web_ui| web_ui.kick_secs) - .unwrap_or_else(default_kick_secs); - Callback::from(move |(name, _): (String, _)| { - if let Ok(action) = StreamsTableAction::from_str(&name) { - match action { - StreamsTableAction::Kick => { - if let Some(dto) = (*selected_dto).as_ref() { - if !services.websocket.send_message(ProtocolMessage::UserAction(UserCommand::Kick( - dto.addr, - dto.channel.virtual_id, - kick_secs, - ))) { - services.toastr.error(translate.t("MESSAGES.FAILED_TO_KICK_USER_STREAM")); - } - } - } - StreamsTableAction::CopyLinkTuliproxVirtualId => { - if let Some(dto) = &*selected_dto { - copy_to_clipboard.emit(dto.channel.virtual_id.to_string()); - } - } - StreamsTableAction::CopyLinkProviderUrl => { - if let Some(dto) = &*selected_dto { - let url = dto.channel.url.to_string(); - let playlist_request = PlaylistRequest::Target(dto.channel.target_id); - let copy_to_clipboard = copy_to_clipboard.clone(); - let services = services.clone(); - spawn_local(async move { - let request = - PlaylistUrlResolveRequest::Provider { playlist_request, url: url.to_string() }; - let resolved = services.playlist.resolve_url(request).await.unwrap_or(url); - copy_to_clipboard.emit(resolved); - }); - } - } - StreamsTableAction::CopyLinkTuliproxWebPlayerUrl => { - if let Some(dto) = &*selected_dto { - let target_id = dto.channel.target_id; - let virtual_id = dto.channel.virtual_id; - let cluster = dto.channel.cluster; - let services = services.clone(); - let translate = translate.clone(); - let copy_to_clipboard = copy_to_clipboard.clone(); - spawn_local(async move { - let request = PlaylistUrlResolveRequest::Webplayer { target_id, virtual_id, cluster }; - if let Some(url) = services.playlist.resolve_url(request).await { - copy_to_clipboard.emit(url); - } else { - services.toastr.error(translate.t("MESSAGES.FAILED_TO_RETRIEVE_WEBPLAYER_URL")); - } - }); - } - } - } - } - popup_is_open_state.set(false); - }) - }; - - html! { -
- { - if let Some(definition) = table_definition.as_ref() { - html! { - <> - definition={definition.clone()} /> - - - - - - - - } - } else { - html! {} - } - } -
- } -} - -#[derive(Debug, Clone, Eq, PartialEq)] -enum StreamsTableAction { - Kick, - CopyLinkTuliproxVirtualId, - CopyLinkTuliproxWebPlayerUrl, - CopyLinkProviderUrl, -} - -impl Display for StreamsTableAction { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!( - f, - "{}", - match self { - Self::Kick => KICK, - Self::CopyLinkTuliproxVirtualId => COPY_LINK_TULIPROX_VIRTUAL_ID, - Self::CopyLinkTuliproxWebPlayerUrl => COPY_LINK_TULIPROX_WEBPLAYER_URL, - Self::CopyLinkProviderUrl => COPY_LINK_PROVIDER_URL, - } - ) - } -} - -impl FromStr for StreamsTableAction { - type Err = TuliproxError; - - fn from_str(s: &str) -> Result { - if s.eq(KICK) { - Ok(Self::Kick) - } else if s.eq(COPY_LINK_TULIPROX_VIRTUAL_ID) { - Ok(Self::CopyLinkTuliproxVirtualId) - } else if s.eq(COPY_LINK_TULIPROX_WEBPLAYER_URL) { - Ok(Self::CopyLinkTuliproxWebPlayerUrl) - } else if s.eq(COPY_LINK_PROVIDER_URL) { - Ok(Self::CopyLinkProviderUrl) - } else { - info_err_res!("Unknown Stream Action: {}", s) - } - } -} - -#[cfg(test)] -mod tests { - use super::{ - apply_stream_meter_entry, build_technical_chips, filter_visible_streams, is_adaptive_session_stream, - StreamMeterCellState, - }; - use shared::{ - model::{PlaylistItemType, StreamChannel, StreamInfo, StreamMeterEntry, StreamTechnicalInfo, XtreamCluster}, - utils::Internable, - }; - use std::{collections::HashMap, net::SocketAddr, rc::Rc}; - - fn test_stream(uid: u32, item_type: PlaylistItemType, session_token: Option<&str>) -> Rc { - Rc::new(StreamInfo { - uid, - meter_uid: uid + 100, - username: "user".to_string(), - channel: StreamChannel { - target_id: 1, - virtual_id: uid, - provider_id: 1, - item_type, - cluster: XtreamCluster::Live, - group: "group".intern(), - title: "title".intern(), - url: "http://localhost/stream".intern(), - shared: false, - technical: None, - }, - provider: "provider".to_string(), - addr: "127.0.0.1:1234".parse::().unwrap_or_else(|_| unreachable!()), - client_ip: "127.0.0.1".to_string(), - user_agent: "ua".to_string(), - ts: 1, - country: None, - session_token: session_token.map(ToOwned::to_owned), - preserved: false, - }) - } - - fn refresh_adaptive_last_seen( - mut current: HashMap, - streams: &[Rc], - now: u64, - ) -> HashMap { - for stream in streams { - if is_adaptive_session_stream(stream) { - if !stream.preserved || !current.contains_key(&stream.uid) { - current.insert(stream.uid, now); - } - } else { - current.remove(&stream.uid); - } - } - current - } - - #[test] - fn test_filter_visible_streams_keeps_adaptive_stream_until_ttl_buffer_expires() { - let stream = test_stream(7, PlaylistItemType::LiveHls, Some("tok")); - let mut adaptive_last_seen = HashMap::new(); - adaptive_last_seen.insert(7, 100); - - let visible = - filter_visible_streams(Some(vec![stream]), &adaptive_last_seen, 119, 15).unwrap_or_else(|| unreachable!()); - assert_eq!(visible.len(), 1); - } - - #[test] - fn test_filter_visible_streams_drops_stale_adaptive_stream_after_ttl_buffer() { - let stream = test_stream(7, PlaylistItemType::LiveDash, Some("tok")); - let mut adaptive_last_seen = HashMap::new(); - adaptive_last_seen.insert(7, 100); - - let visible = - filter_visible_streams(Some(vec![stream]), &adaptive_last_seen, 121, 15).unwrap_or_else(|| unreachable!()); - assert!(visible.is_empty()); - } - - #[test] - fn test_filter_visible_streams_keeps_new_adaptive_stream_without_last_seen_entry() { - let stream = test_stream(8, PlaylistItemType::LiveHls, Some("tok-new")); - - let visible = - filter_visible_streams(Some(vec![stream]), &HashMap::new(), 200, 15).unwrap_or_else(|| unreachable!()); - assert_eq!(visible.len(), 1); - } - - #[test] - fn test_preserved_adaptive_stream_does_not_refresh_last_seen_on_status_refresh() { - let first = test_stream(7, PlaylistItemType::LiveHls, Some("tok-a")); - let mut second = (*test_stream(9, PlaylistItemType::LiveHls, Some("tok-b"))).clone(); - second.preserved = true; - let second = Rc::new(second); - - let mut adaptive_last_seen = HashMap::new(); - adaptive_last_seen.insert(7, 100); - adaptive_last_seen.insert(9, 50); - - let refreshed = refresh_adaptive_last_seen(adaptive_last_seen, &[first, second], 120); - - assert_eq!(refreshed.get(&7), Some(&120)); - assert_eq!(refreshed.get(&9), Some(&50)); - } - - #[test] - fn test_build_technical_chips_prepends_hls_label_before_container() { - let technical = StreamTechnicalInfo { container: "mpegts".to_string(), ..Default::default() }; - - let chips = build_technical_chips(PlaylistItemType::LiveHls, Some(&technical)); - - assert_eq!(chips.first().map(|(label, _)| label.as_str()), Some("HLS")); - assert_eq!(chips.get(1).map(|(label, _)| label.as_str()), Some("MPEGTS")); - } - - #[test] - fn test_build_technical_chips_prepends_dash_label_before_container() { - let technical = StreamTechnicalInfo { container: "mpegts".to_string(), ..Default::default() }; - - let chips = build_technical_chips(PlaylistItemType::LiveDash, Some(&technical)); - - assert_eq!(chips.first().map(|(label, _)| label.as_str()), Some("DASH")); - assert_eq!(chips.get(1).map(|(label, _)| label.as_str()), Some("MPEGTS")); - } - - #[test] - fn test_apply_stream_meter_entry_accumulates_totals_across_meter_switches_for_same_uid() { - let first = StreamMeterEntry { meter_uid: 101, uids: vec![7], rate_kbps: 50, total_kb: 100 }; - let second = StreamMeterEntry { meter_uid: 202, uids: vec![7], rate_kbps: 25, total_kb: 30 }; - - let state = apply_stream_meter_entry(&StreamMeterCellState::default(), 101, &first); - let state = apply_stream_meter_entry(&state, 202, &second); - - assert_eq!(state.rate_kbps, 25); - assert_eq!(state.transferred_total_kb.saturating_add(state.current_meter_total_kb), 130); - } - - #[test] - fn test_apply_stream_meter_entry_ignores_stale_rate_from_old_meter() { - let current = StreamMeterEntry { meter_uid: 202, uids: vec![7], rate_kbps: 40, total_kb: 60 }; - let stale = StreamMeterEntry { meter_uid: 101, uids: vec![7], rate_kbps: 99, total_kb: 120 }; - - let state = apply_stream_meter_entry(&StreamMeterCellState::default(), 202, ¤t); - let state = apply_stream_meter_entry(&state, 202, &stale); - - assert_eq!(state.rate_kbps, 40); - assert_eq!(state.transferred_total_kb.saturating_add(state.current_meter_total_kb), 180); - } - - #[test] - fn test_apply_stream_meter_entry_does_not_grow_state_per_historical_meter() { - let first = StreamMeterEntry { meter_uid: 101, uids: vec![7], rate_kbps: 50, total_kb: 100 }; - let second = StreamMeterEntry { meter_uid: 202, uids: vec![7], rate_kbps: 25, total_kb: 30 }; - - let state = apply_stream_meter_entry(&StreamMeterCellState::default(), 101, &first); - let state = apply_stream_meter_entry(&state, 202, &second); - - assert_eq!(state.current_meter_uid, 202); - assert_eq!(state.transferred_total_kb, 100); - assert_eq!(state.current_meter_total_kb, 30); - } -} diff --git a/frontend/src/app/components/dashboard/streams_view.rs b/frontend/src/app/components/dashboard/streams_view.rs index b02365257..d8019fcc3 100644 --- a/frontend/src/app/components/dashboard/streams_view.rs +++ b/frontend/src/app/components/dashboard/streams_view.rs @@ -1,6 +1,6 @@ use crate::{ app::{ - components::{Card, StatusCard, StreamsTable}, + components::{Card, StatusCard, StreamDisplay}, StatusContext, }, hooks::use_service_context, @@ -74,7 +74,7 @@ pub fn StreamsView(props: &StreamsViewProps) -> Html { /> - + } diff --git a/shared/src/model/stream_info.rs b/shared/src/model/stream_info.rs index be110e5d1..1c688ff1c 100644 --- a/shared/src/model/stream_info.rs +++ b/shared/src/model/stream_info.rs @@ -58,6 +58,7 @@ pub fn create_stream_channel_with_type( ) -> StreamChannel { let mut stream_channel = pli.to_stream_channel(target_id); stream_channel.item_type = item_type; + stream_channel.cluster = XtreamCluster::try_from(item_type).unwrap_or(stream_channel.cluster); stream_channel } @@ -307,3 +308,41 @@ impl StreamInfo { } } } + +#[cfg(test)] +mod tests { + use super::create_stream_channel_with_type; + use crate::{ + model::{PlaylistItemType, XtreamCluster, XtreamPlaylistItem}, + utils::Internable, + }; + + #[test] + fn create_stream_channel_with_type_keeps_cluster_consistent_with_item_type() { + let playlist_item = XtreamPlaylistItem { + virtual_id: 93_995, + provider_id: 1, + name: "Example".intern(), + logo: "".intern(), + logo_small: "".intern(), + group: "Movies".intern(), + title: "Example".intern(), + parent_code: "".intern(), + rec: "".intern(), + url: "http://provider.example/movie/93995.mkv".intern(), + epg_channel_id: None, + xtream_cluster: XtreamCluster::Live, + additional_properties: None, + item_type: PlaylistItemType::Live, + category_id: 0, + input_name: "demo".intern(), + channel_no: 0, + source_ordinal: 0, + }; + + let stream_channel = create_stream_channel_with_type(1, &playlist_item, PlaylistItemType::Video); + + assert_eq!(stream_channel.item_type, PlaylistItemType::Video); + assert_eq!(stream_channel.cluster, XtreamCluster::Video); + } +}