euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-15 12:22:18 +02:00
Code Issues Packages Projects Releases 5 Wiki Activity

refactored stream table (#674)

Browse Source 
* refactored stream table
This commit is contained in:
euzu
2026-03-27 23:25:26 +01:00
committed by GitHub
parent 68e1dc80d9
commit 7ebf810836
12 changed files with 1100 additions and 915 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/api/endpoints/v1_api_playlist.rs
+29 -8
View File
@@ -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<Arc<AppState>>,
axum::extract::Json(targets): axum::extract::Json<Vec<String>>,
@@ -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(
frontend/public/assets/i18n/en.json
+5 -2
View File
@@ -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",
frontend/scss/app/components/dashboard/_streams_view.scss
+195 -31
View File
@@ -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;
}
}
}
frontend/src/app/components/authentication.rs
+1 -1
View File
@@ -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);
}
}
|| ()
frontend/src/app/components/dashboard/mod.rs
+2 -2
View File
@@ -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::*,
};
frontend/src/app/components/dashboard/stream_display/helpers.rs
+279
View File
@@ -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<Vec<std::rc::Rc<StreamInfo>>>,
adaptive_last_seen: &HashMap<u32, u64>,
now_secs: u64,
adaptive_ttl_secs: u64,
) -> Option<Vec<std::rc::Rc<StreamInfo>>> {
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<HashMap<u32, u64>>,
streams: &Option<Vec<std::rc::Rc<StreamInfo>>>,
) {
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<u32, u64>,
streams: &Option<Vec<std::rc::Rc<StreamInfo>>>,
now: u64,
) -> HashMap<u32, u64> {
if let Some(streams) = streams {
let adaptive_uids: HashSet<u32> =
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::<Element>() 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::<u64>() {
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<StreamInfo> {
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));
}
}
frontend/src/app/components/dashboard/stream_display/item.rs
+124
View File
@@ -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<StreamInfo>,
pub metrics_enabled: bool,
pub on_popup_click: Callback<(Rc<StreamInfo>, 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! {
<article class="tp__stream-display__item">
<button class="tp__stream-display__menu" onclick={handle_popup_click}>
<AppIcon name="Popup" />
</button>
<div class="tp__stream-display__item-content">
<div class="tp__stream-display__item-head">
<div class="tp__stream-display__identity">
<div class="tp__stream-display__title-row">
<div class="tp__stream-display__title-block">
<div class="tp__stream-display__title">{stream.channel.title.to_string()}</div>
<div class="tp__stream-display__subtitle">
<span class="tp__stream-display__username"> {stream.username.clone()}</span>
{" • "}
<span class="tp__stream-display__channel_type">{render_cluster(&stream.channel)}</span>
{" • "}
<span class="tp__stream-display__provider">{stream.provider.clone()}</span>
if !country.is_empty() {
<>
{" • "}
{country}
</>
}
</div>
</div>
</div>
</div>
<div class="tp__stream-display__stats">
<div class="tp__stream-display__stat">
<span class="tp__stream-display__stat-label">{translate.t("LABEL.CLIENT_IP")}</span>
<span class="tp__stream-display__stat-value">{client_ip.clone()}</span>
</div>
<div class="tp__stream-display__stat">
<span class="tp__stream-display__stat-label">{translate.t("LABEL.SHARED")}</span>
<span class="tp__stream-display__stat-value"><ToggleSwitch value={stream.channel.shared} readonly={true} /></span>
</div>
<div class="tp__stream-display__stat">
<span class="tp__stream-display__stat-label">{translate.t("LABEL.DURATION")}</span>
<span class="tp__stream-display__stat-value tp__stream-display__duration" data-ts={stream.ts.to_string()}>
{format_duration(current_time_secs().saturating_sub(stream.ts))}
</span>
</div>
if props.metrics_enabled {
<div class="tp__stream-display__stat">
<span class="tp__stream-display__stat-label">{translate.t("LABEL.BANDWIDTH")}</span>
<span class="tp__stream-display__stat-value">
<StreamMeterBadge uid={stream.uid} meter_uid={stream.meter_uid} kind={MeterDisplayKind::Bandwidth} />
</span>
</div>
<div class="tp__stream-display__stat">
<span class="tp__stream-display__stat-label">{translate.t("LABEL.TRANSFERRED")}</span>
<span class="tp__stream-display__stat-value">
<StreamMeterBadge uid={stream.uid} meter_uid={stream.meter_uid} kind={MeterDisplayKind::Transferred} />
</span>
</div>
}
</div>
</div>
<div class="tp__stream-display__row">
if !chips.is_empty() {
<div class="tp__stream-display__chips">
{ for chips.into_iter().map(|(label, chip_class)| html! {
<Chip label={label} class={Some(format!("tp__stream-display__chip {chip_class}"))} />
})}
</div>
}
<div class="tp__stream-display__details">
<div class="tp__stream-display__detail">
<span class="tp__stream-display__detail-label">{translate.t("LABEL.GROUP")}</span>
<span class="tp__stream-display__detail-value">{stream.channel.group.to_string()}</span>
</div>
<div class="tp__stream-display__detail tp__stream-display__detail">
<span class="tp__stream-display__detail-label">{translate.t("LABEL.PLAYER")}</span>
<span class="tp__stream-display__detail-value">
<RevealContent preview={Some(html! { &stream.user_agent })}>{&stream.user_agent}</RevealContent>
</span>
</div>
</div>
</div>
</div>
</article>
}
}
frontend/src/app/components/dashboard/stream_display/meter.rs
+97
View File
@@ -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! { <span>{label}</span> }
}
frontend/src/app/components/dashboard/stream_display/mod.rs
+327
View File
@@ -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<Vec<Rc<StreamInfo>>>,
}
#[component]
pub fn StreamDisplay(props: &StreamDisplayProps) -> Html {
let translate = use_translation();
let service_ctx = use_service_context();
let dialog = use_context::<DialogService>().expect("Dialog service not found");
let clipboard = use_clipboard();
let config_ctx = use_context::<ConfigContext>().expect("Config context not found");
let popup_anchor_ref = use_state(|| None::<web_sys::Element>);
let popup_is_open = use_state(|| false);
let selected_dto = use_state(|| None::<Rc<StreamInfo>>);
let adaptive_last_seen = use_state(HashMap::<u32, u64>::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<StreamInfo>, MouseEvent)| {
if let Some(streams) = event.current_target().and_then(|target| target.dyn_into::<web_sys::Element>().ok())
{
set_selected_dto.set(Some(dto));
set_anchor_ref.set(Some(streams));
set_is_open.set(true);
}
})
};
let copy_to_clipboard: Callback<String> = {
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! {<input value={text} readonly={true} class="tp__copy-input"/>}, 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! {
<div class="tp__stream-display">
<div class="tp__stream-display__header">
<label>{translate.t("LABEL.ACTIVE_STREAMS")}</label>
</div>
<div class="tp__stream-display__body">
{
if let Some(streams) = visible_streams.as_ref() {
if streams.is_empty() {
html! { <NoContent /> }
} else {
html! {
<>
<div class="tp__stream-display__list">
{ for streams.iter().cloned().map(|stream| {
let uid = stream.uid;
html! {
<StreamDisplayItem
key={uid}
stream={stream}
metrics_enabled={metrics_enabled}
on_popup_click={handle_popup_onclick.clone()}
/>
}
})}
</div>
<PopupMenu is_open={*popup_is_open} anchor_ref={(*popup_anchor_ref).clone()} on_close={handle_popup_close}>
<MenuItem icon="Disconnect" name={StreamDisplayAction::Kick.to_string()} label={translate.t("LABEL.KICK")} onclick={&handle_menu_click} class="tp__delete_action"></MenuItem>
<MenuItem icon="Clipboard" name={StreamDisplayAction::CopyLinkTuliproxVirtualId.to_string()} label={translate.t("LABEL.COPY_LINK_TULIPROX_VIRTUAL_ID")} onclick={&handle_menu_click}></MenuItem>
<MenuItem icon="Clipboard" name={StreamDisplayAction::CopyLinkTuliproxWebPlayerUrl.to_string()} label={translate.t("LABEL.COPY_LINK_TULIPROX_WEBPLAYER_URL")} onclick={&handle_menu_click}></MenuItem>
<MenuItem icon="Clipboard" name={StreamDisplayAction::CopyLinkProviderUrl.to_string()} label={translate.t("LABEL.COPY_LINK_PROVIDER_URL")} onclick={&handle_menu_click}></MenuItem>
</PopupMenu>
</>
}
}
} else {
html! { <NoContent /> }
}
}
</div>
</div>
}
}
#[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<Self, TuliproxError> {
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)
}
}
}
}
frontend/src/app/components/dashboard/streams_table.rs
-869
View File
@@ -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<Vec<Rc<StreamInfo>>>,
adaptive_last_seen: &HashMap<u32, u64>,
now_secs: u64,
adaptive_ttl_secs: u64,
) -> Option<Vec<Rc<StreamInfo>>> {
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::<u64>() {
el.set_inner_html(&format_duration(current_time_secs() - ts));
}
}
}
}
}
#[derive(Properties, PartialEq, Clone)]
pub struct StreamsTableProps {
pub streams: Option<Vec<Rc<StreamInfo>>>,
}
#[component]
pub fn StreamsTable(props: &StreamsTableProps) -> Html {
let translate = use_translation();
let service_ctx = use_service_context();
let dialog = use_context::<DialogService>().expect("Dialog service not found");
let clipboard = use_clipboard();
let config_ctx = use_context::<ConfigContext>().expect("Config context not found");
let popup_anchor_ref = use_state(|| None::<web_sys::Element>);
let popup_is_open = use_state(|| false);
let selected_dto = use_state(|| None::<Rc<StreamInfo>>);
let adaptive_last_seen = use_state(HashMap::<u32, u64>::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<StreamInfo>, MouseEvent)| {
if let Some(streams) = event.target_dyn_into::<web_sys::Element>() {
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::<usize, Html>::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<StreamInfo>), Html>::from(
move |(row, col, dto): (usize, usize, Rc<StreamInfo>)| match headers[col] {
"EMPTY" => {
let popup_onclick = popup_onclick.clone();
html! {
<button class="tp__icon-button"
onclick={Callback::from(move |event: MouseEvent| popup_onclick.emit((dto.clone(), event)))}
data-row={row.to_string()}>
<AppIcon name="Popup"></AppIcon>
</button>
}
}
"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! {
<div class="tp__streams-table__tech-chips">
{ for chips.into_iter().map(|(label, chip_class)| html! {
<Chip
label={label}
class={Some(format!("tp__streams-table__tech-chip {chip_class}"))}
/>
})}
</div>
}
}
}
"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! { <ToggleSwitch value={dto.channel.shared} readonly={true} /> },
"USER_AGENT" => {
html! { <RevealContent preview={Some(html! { &dto.user_agent })}>{&dto.user_agent}</RevealContent> }
}
"DURATION" => {
html! { <span class="tp__stream-table__duration" data-ts={dto.ts.to_string()}>{format_duration(current_time_secs() - dto.ts)}</span> }
}
"BANDWIDTH" => {
html! { <StreamMeterCell uid={dto.uid} meter_uid={dto.meter_uid} kind={MeterDisplayKind::Bandwidth} /> }
}
"TRANSFERRED" => {
html! { <StreamMeterCell uid={dto.uid} meter_uid={dto.meter_uid} kind={MeterDisplayKind::Transferred} /> }
}
_ => html! {""},
},
)
};
let is_sortable = Callback::<usize, bool>::from(move |_col| false);
let on_sort = Callback::<Option<(usize, SortOrder)>, ()>::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::<StreamInfo> {
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<String> = {
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! {<input value={text} readonly={true} class="tp__copy-input"/>}, 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! {
<div class="tp__streams-table">
{
if let Some(definition) = table_definition.as_ref() {
html! {
<>
<Table::<StreamInfo> definition={definition.clone()} />
<PopupMenu is_open={*popup_is_open} anchor_ref={(*popup_anchor_ref).clone()} on_close={handle_popup_close}>
<MenuItem icon="Disconnect" name={StreamsTableAction::Kick.to_string()} label={translate.t("LABEL.KICK")} onclick={&handle_menu_click} class="tp__delete_action"></MenuItem>
<MenuItem icon="Clipboard" name={StreamsTableAction::CopyLinkTuliproxVirtualId.to_string()} label={translate.t("LABEL.COPY_LINK_TULIPROX_VIRTUAL_ID")} onclick={&handle_menu_click}></MenuItem>
<MenuItem icon="Clipboard" name={StreamsTableAction::CopyLinkTuliproxWebPlayerUrl.to_string()} label={translate.t("LABEL.COPY_LINK_TULIPROX_WEBPLAYER_URL")} onclick={&handle_menu_click}></MenuItem>
<MenuItem icon="Clipboard" name={StreamsTableAction::CopyLinkProviderUrl.to_string()} label={translate.t("LABEL.COPY_LINK_PROVIDER_URL")} onclick={&handle_menu_click}></MenuItem>
</PopupMenu>
</>
}
} else {
html! {}
}
}
</div>
}
}
#[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<Self, TuliproxError> {
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<StreamInfo> {
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::<SocketAddr>().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<u32, u64>,
streams: &[Rc<StreamInfo>],
now: u64,
) -> HashMap<u32, u64> {
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, &current);
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);
}
}
frontend/src/app/components/dashboard/streams_view.rs
+2 -2
View File
@@ -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 {
/>
</Card>
</div>
<StreamsTable streams={ (*memo_streams).clone() } />
<StreamDisplay streams={ (*memo_streams).clone() } />
</div>
</div>
}
shared/src/model/stream_info.rs
+39
View File
@@ -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);
}
}