mirror of
https://github.com/euzu/tuliprox.git
synced 2026-10-26 08:42:16 +01:00
File diff suppressed because it is too large
Load Diff
@@ -11,6 +11,7 @@ mod epg_group_index;
|
||||
mod epg_repository;
|
||||
mod geoip;
|
||||
mod library_repository;
|
||||
mod lineup_guide_index;
|
||||
mod live_stream_metadata_repository;
|
||||
mod m3u_playlist_iterator;
|
||||
mod m3u_repository;
|
||||
@@ -47,6 +48,7 @@ pub use epg_group_index::*;
|
||||
pub use epg_repository::*;
|
||||
pub use geoip::*;
|
||||
pub use library_repository::*;
|
||||
pub use lineup_guide_index::*;
|
||||
pub use live_stream_metadata_repository::{
|
||||
load_input_live_bitrate_bps, persist_input_live_bitrate_bps, LiveBitratePersistenceOutcome,
|
||||
};
|
||||
|
||||
@@ -0,0 +1,623 @@
|
||||
//! `GuideNumber` index of the `HDHomeRun` lineup of a target.
|
||||
//!
|
||||
//! `lineup.json` publishes one guide number per channel — the EPG channel id, else the channel
|
||||
//! name — and the tuner endpoints receive that string back as `/auto/{channel}`. Answering it by
|
||||
//! walking the playlist iterator reads the complete target playlist per tune (measured: 0.4 to
|
||||
//! 0.8 s for 50 000 channels), while the keyed lookup by virtual id this index feeds takes about
|
||||
//! a millisecond.
|
||||
//!
|
||||
//! The index is a sidecar of one target playlist view, written by the same persistence step that
|
||||
//! writes the playlist it describes, and only for targets that publish an `HDHomeRun` output. A
|
||||
//! missing index is therefore not an error: readers fall back to the scan, which stays the
|
||||
//! reference implementation.
|
||||
//!
|
||||
//! Every feed of a guide number is kept, not only the one the lineup publishes first. The tuner
|
||||
//! serves the first feed that this user's output publishes, and channel failover will pick among
|
||||
//! the same list by quality of service instead.
|
||||
|
||||
use crate::{m3u_get_file_path_for_db, storage_const, xtream_get_file_path, xtream_get_storage_path};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use shared::{
|
||||
concat_string,
|
||||
error::TuliproxError,
|
||||
model::{PlaylistGroup, PlaylistItemType, TargetType, XtreamCluster},
|
||||
};
|
||||
use std::{
|
||||
io,
|
||||
path::{Path, PathBuf},
|
||||
sync::Arc,
|
||||
};
|
||||
use tuliprox_btree::{BPlusTree, BPlusTreeQuery};
|
||||
use tuliprox_core::{
|
||||
model::{Config, ConfigTarget},
|
||||
utils::FileLockManager,
|
||||
};
|
||||
|
||||
/// Target playlist view a lineup is published from.
|
||||
///
|
||||
/// `lineup.json` serves either the M3U target playlist or the Xtream target playlists, selected
|
||||
/// by the `use_output` of the `HDHomeRun` output, so each view needs its own index.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum LineupGuideView {
|
||||
/// The flat M3U target playlist: one key space holding every item type.
|
||||
M3u,
|
||||
/// The Xtream target playlists: live and video, each in its own db.
|
||||
Xtream,
|
||||
}
|
||||
|
||||
/// The item a guide number resolves to.
|
||||
///
|
||||
/// A guide number usually names one channel, but providers publish one EPG id for several feeds of
|
||||
/// the same channel (SD, HD, FHD, backup), which is also what channel failover groups. The index
|
||||
/// therefore keeps every feed of a guide number, ordered the way the lineup scan meets them.
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
pub struct LineupGuideEntry {
|
||||
/// Cluster holding the virtual id, needed to fetch from the right Xtream db.
|
||||
pub cluster: XtreamCluster,
|
||||
pub virtual_id: u32,
|
||||
}
|
||||
|
||||
/// Guide number to every [`LineupGuideEntry`] carrying it.
|
||||
pub type LineupGuideTree = BPlusTree<Arc<str>, Vec<LineupGuideEntry>>;
|
||||
|
||||
/// The candidates one requested key has, in the order the lineup meets them.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct LineupGuideCandidate {
|
||||
/// The lookup key that matched, so the caller can verify the item still carries it.
|
||||
pub key: Arc<str>,
|
||||
pub entries: Vec<LineupGuideEntry>,
|
||||
}
|
||||
|
||||
/// Outcome of a guide number lookup.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub enum LineupGuideLookup {
|
||||
/// No usable index: the target never published an `HDHomeRun` lineup, or a persistence step is
|
||||
/// between removing the old index and publishing the new one.
|
||||
Missing,
|
||||
/// The index exists but knows none of the requested keys.
|
||||
NoEntry,
|
||||
/// Candidates of every requested key the index knows, in the order the keys were asked for, so
|
||||
/// the caller can tell a guide number a channel carries from the stripped form of another.
|
||||
Candidates(Vec<LineupGuideCandidate>),
|
||||
}
|
||||
|
||||
/// View `lineup.json` is published from, selected by the `use_output` of the `HDHomeRun` output.
|
||||
///
|
||||
/// Without a `use_output` the M3U playlist wins over the Xtream playlists. `None` means the target
|
||||
/// has no output a lineup can be published from.
|
||||
///
|
||||
/// This deliberately does not require the target to carry the `HDHomeRun` output itself: the device
|
||||
/// serves the target of its user, and `lineup.json` has always published that target's playlist
|
||||
/// with the default selection when the output is not on it. Whether a guide index exists is the
|
||||
/// separate question [`lineup_guide_index_view`] answers.
|
||||
pub fn hdhomerun_lineup_view(target: &ConfigTarget) -> Option<LineupGuideView> {
|
||||
let use_output = target.get_hdhomerun_output().and_then(|output| output.use_output);
|
||||
let has_m3u = target.has_output(TargetType::M3u);
|
||||
let has_xtream = target.has_output(TargetType::Xtream);
|
||||
match use_output {
|
||||
None | Some(TargetType::M3u) if has_m3u => Some(LineupGuideView::M3u),
|
||||
None | Some(TargetType::Xtream) if has_xtream => Some(LineupGuideView::Xtream),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// View whose guide index a target keeps: only a target publishing an `HDHomeRun` output has one.
|
||||
pub fn lineup_guide_index_view(target: &ConfigTarget) -> Option<LineupGuideView> {
|
||||
if target.has_output(TargetType::HdHomeRun) {
|
||||
hdhomerun_lineup_view(target)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
fn lineup_guide_file_name() -> String { concat_string!("lineup_guide.", storage_const::FILE_SUFFIX_DB) }
|
||||
|
||||
/// Index file of the M3U view, next to the M3U playlist db of the target.
|
||||
pub fn m3u_lineup_guide_index_path(target_path: &Path) -> PathBuf {
|
||||
m3u_get_file_path_for_db(target_path).with_file_name(lineup_guide_file_name())
|
||||
}
|
||||
|
||||
/// Index file of the Xtream view, next to the live playlist db of the target.
|
||||
pub fn xtream_lineup_guide_index_path(config: &Config, target_name: &str) -> Option<PathBuf> {
|
||||
xtream_get_storage_path(config, target_name)
|
||||
.map(|path| xtream_get_file_path(&path, XtreamCluster::Live).with_file_name(lineup_guide_file_name()))
|
||||
}
|
||||
|
||||
/// Rank of a cluster when the lineup of the Xtream view is published: the live db is searched
|
||||
/// before the video db, so live wins a guide number two channels share.
|
||||
fn cluster_rank(cluster: XtreamCluster) -> u8 {
|
||||
match cluster {
|
||||
XtreamCluster::Live => 0,
|
||||
XtreamCluster::Video => 1,
|
||||
XtreamCluster::Series => 2,
|
||||
}
|
||||
}
|
||||
|
||||
/// Cluster rank, source ordinal and virtual id: the order the lineup meets its items in.
|
||||
type LineupPosition = (u8, u32, u32);
|
||||
|
||||
/// Position of an item in the lineup of `view`.
|
||||
///
|
||||
/// The playlist dbs are read through their sorted index, which orders by `source_ordinal` (see
|
||||
/// `m3u_write_playlist` and the Xtream publication). The M3U db is one key space, the Xtream view is
|
||||
/// live first and then video. The virtual id breaks ties the way the db key does.
|
||||
fn lineup_position(
|
||||
view: LineupGuideView,
|
||||
cluster: XtreamCluster,
|
||||
source_ordinal: u32,
|
||||
virtual_id: u32,
|
||||
) -> LineupPosition {
|
||||
let cluster_rank = match view {
|
||||
LineupGuideView::M3u => 0,
|
||||
LineupGuideView::Xtream => cluster_rank(cluster),
|
||||
};
|
||||
(cluster_rank, source_ordinal, virtual_id)
|
||||
}
|
||||
|
||||
/// One index row: guide number, lineup position and the entry.
|
||||
type LineupGuideRow = (Arc<str>, LineupPosition, LineupGuideEntry);
|
||||
|
||||
/// Entries of clusters a persistence step keeps, read from the index that already describes them.
|
||||
///
|
||||
/// An Xtream update without items for one cluster keeps that cluster's db, so the new index cannot
|
||||
/// learn its channels from the update. They are carried over from the previous index instead: it
|
||||
/// was written together with the kept db, which has not changed since.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct LineupGuideCarryOver {
|
||||
rows: Vec<LineupGuideRow>,
|
||||
}
|
||||
|
||||
/// Rows of the playlist channels the index of `view` holds.
|
||||
fn playlist_rows(playlist: &[PlaylistGroup], view: LineupGuideView) -> Vec<LineupGuideRow> {
|
||||
playlist
|
||||
.iter()
|
||||
.flat_map(|group| &group.channels)
|
||||
.filter_map(|item| {
|
||||
let header = &item.header;
|
||||
let cluster = header.xtream_cluster;
|
||||
let indexed = match view {
|
||||
LineupGuideView::M3u => {
|
||||
!matches!(header.item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo)
|
||||
}
|
||||
LineupGuideView::Xtream => matches!(cluster, XtreamCluster::Live | XtreamCluster::Video),
|
||||
};
|
||||
if !indexed {
|
||||
return None;
|
||||
}
|
||||
let guide = header.epg_channel_id.as_ref().unwrap_or(&header.name);
|
||||
if guide.is_empty() {
|
||||
return None;
|
||||
}
|
||||
let virtual_id = header.virtual_id.get();
|
||||
Some((
|
||||
Arc::clone(guide),
|
||||
lineup_position(view, cluster, header.source_ordinal, virtual_id),
|
||||
LineupGuideEntry { cluster, virtual_id },
|
||||
))
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Groups rows by guide number into the tree, each group in lineup order.
|
||||
///
|
||||
/// The rows are one flat list sorted by guide number and lineup position, so grouping needs no map
|
||||
/// with one vector per guide number next to the tree being built.
|
||||
fn tree_from_rows(mut rows: Vec<LineupGuideRow>) -> LineupGuideTree {
|
||||
rows.sort_unstable_by(|left, right| (&left.0, left.1).cmp(&(&right.0, right.1)));
|
||||
|
||||
let mut tree = LineupGuideTree::new();
|
||||
let mut rows = rows.into_iter().peekable();
|
||||
while let Some((guide, _, entry)) = rows.next() {
|
||||
let mut entries = vec![entry];
|
||||
while let Some((_, _, entry)) = rows.next_if(|(next, _, _)| *next == guide) {
|
||||
// The same item listed in several groups carries different source ordinals, so its
|
||||
// rows need not be adjacent. A guide number has only a handful of feeds, which keeps
|
||||
// this check cheap; the first occurrence keeps its lineup position.
|
||||
if !entries.contains(&entry) {
|
||||
entries.push(entry);
|
||||
}
|
||||
}
|
||||
tree.insert(guide, entries);
|
||||
}
|
||||
tree
|
||||
}
|
||||
|
||||
/// Builds the index of one target playlist view.
|
||||
///
|
||||
/// The Xtream view indexes live and video only, matching the channels `lineup.json` publishes for
|
||||
/// it. The M3U view indexes what its db holds: every item except series info, which
|
||||
/// `m3u_write_playlist` leaves out.
|
||||
///
|
||||
/// The guide number is the one `lineup.json` publishes: the EPG channel id when it is set,
|
||||
/// otherwise the name. An empty EPG channel id publishes the empty guide number, which no request
|
||||
/// can produce, so such items are left out.
|
||||
pub fn build_lineup_guide_index(playlist: &[PlaylistGroup], view: LineupGuideView) -> LineupGuideTree {
|
||||
tree_from_rows(playlist_rows(playlist, view))
|
||||
}
|
||||
|
||||
/// Reads the entries of `clusters` from the index at `path`, to carry them into the next index.
|
||||
///
|
||||
/// Only the rows of the kept clusters are held; the rest of the index is streamed past. Each keeps
|
||||
/// its position within its guide number, which is lineup order: the clusters rank live before
|
||||
/// video, and the previous index already lists the feeds of one cluster in lineup order. `None`
|
||||
/// when the index is missing or unreadable, so the caller cannot describe the kept clusters.
|
||||
pub async fn lineup_guide_index_carry_over(
|
||||
file_locks: &FileLockManager,
|
||||
path: &Path,
|
||||
clusters: &[XtreamCluster],
|
||||
) -> Option<LineupGuideCarryOver> {
|
||||
let file_lock = file_locks.read_lock(path).await;
|
||||
let path = path.to_path_buf();
|
||||
let clusters = clusters.to_vec();
|
||||
let task = tokio::task::spawn_blocking(move || -> io::Result<LineupGuideCarryOver> {
|
||||
let _guard = file_lock;
|
||||
let mut query = BPlusTreeQuery::<Arc<str>, Vec<LineupGuideEntry>>::try_new(&path)?;
|
||||
let mut rows = Vec::new();
|
||||
for entry in query.iter() {
|
||||
let (guide, entries) = entry?;
|
||||
for (ordinal, entry) in (0u32..).zip(entries) {
|
||||
if clusters.contains(&entry.cluster) {
|
||||
let position = (cluster_rank(entry.cluster), ordinal, entry.virtual_id);
|
||||
rows.push((Arc::clone(&guide), position, entry));
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(LineupGuideCarryOver { rows })
|
||||
});
|
||||
match task.await {
|
||||
Ok(Ok(carry_over)) => Some(carry_over),
|
||||
Ok(Err(err)) => {
|
||||
log::debug!("No lineup guide index to carry over: {err}");
|
||||
None
|
||||
}
|
||||
Err(err) => {
|
||||
log::warn!("Lineup guide index carry over task failed: {err}");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Builds the index of `view` and stores it at `path`, together with the entries `carry_over`
|
||||
/// brings from the previous index.
|
||||
///
|
||||
/// Only collecting the rows from the borrowed playlist runs on the calling task. Sorting, building
|
||||
/// the tree, creating the directory and the store run on a blocking thread under the write lock of
|
||||
/// the index file, so a reader never opens a file the store is still copying (the store falls back
|
||||
/// to a copy when it cannot rename).
|
||||
pub async fn lineup_guide_index_write(
|
||||
file_locks: &FileLockManager,
|
||||
playlist: &[PlaylistGroup],
|
||||
view: LineupGuideView,
|
||||
path: &Path,
|
||||
carry_over: Option<LineupGuideCarryOver>,
|
||||
) -> Result<(), TuliproxError> {
|
||||
let mut rows = playlist_rows(playlist, view);
|
||||
if let Some(carry_over) = carry_over {
|
||||
rows.extend(carry_over.rows);
|
||||
}
|
||||
let file_lock = file_locks.write_lock(path).await;
|
||||
let path = path.to_path_buf();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let _guard = file_lock;
|
||||
let mut tree = tree_from_rows(rows);
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent).map_err(|err| {
|
||||
TuliproxError::RepositoryPlaylist(format!(
|
||||
"Failed to create lineup guide index directory {}: {err}",
|
||||
parent.display()
|
||||
))
|
||||
})?;
|
||||
}
|
||||
tree.store(&path).map(|_| ()).map_err(|err| {
|
||||
TuliproxError::RepositoryPlaylist(format!("Failed to write lineup guide index {}: {err}", path.display()))
|
||||
})
|
||||
})
|
||||
.await
|
||||
.map_err(|err| TuliproxError::RepositoryPlaylist(format!("Lineup guide index write task failed: {err}")))?
|
||||
}
|
||||
|
||||
/// Removes the indexes at `paths`, so they cannot describe a playlist they were not built for.
|
||||
///
|
||||
/// A missing file is fine: it is what a reader expects before the first persistence step, and
|
||||
/// what a failed write leaves behind. Every target persistence step asks for this, so the common
|
||||
/// case of no index at all costs one blocking hop for the existence checks and takes no lock.
|
||||
pub async fn lineup_guide_index_remove(file_locks: &FileLockManager, paths: &[PathBuf]) -> Result<(), TuliproxError> {
|
||||
let candidates = paths.to_vec();
|
||||
let existing = tokio::task::spawn_blocking(move || {
|
||||
// An unanswerable existence check is treated as "exists", so the removal is still tried.
|
||||
candidates.into_iter().filter(|path| path.try_exists().unwrap_or(true)).collect::<Vec<_>>()
|
||||
})
|
||||
.await
|
||||
.map_err(|err| TuliproxError::RepositoryPlaylist(format!("Lineup guide index removal task failed: {err}")))?;
|
||||
|
||||
for path in existing {
|
||||
let _guard = file_locks.write_lock(&path).await;
|
||||
match tokio::fs::remove_file(&path).await {
|
||||
Err(err) if err.kind() != io::ErrorKind::NotFound => {
|
||||
return Err(TuliproxError::RepositoryPlaylist(format!(
|
||||
"Failed to remove lineup guide index {}: {err}",
|
||||
path.display()
|
||||
)));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Every candidate the index holds for `keys`, grouped per key in the order the keys were given.
|
||||
///
|
||||
/// Runs the file access on a blocking thread, so all keys are answered from one open index. An
|
||||
/// index that cannot be opened is reported as [`LineupGuideLookup::Missing`] so the caller uses its
|
||||
/// scan instead of failing the request.
|
||||
pub async fn lineup_guide_candidates(
|
||||
file_locks: &FileLockManager,
|
||||
path: PathBuf,
|
||||
keys: Vec<Arc<str>>,
|
||||
) -> LineupGuideLookup {
|
||||
let file_lock = file_locks.read_lock(&path).await;
|
||||
let task = tokio::task::spawn_blocking(move || {
|
||||
let _guard = file_lock;
|
||||
let Ok(mut query) = BPlusTreeQuery::<Arc<str>, Vec<LineupGuideEntry>>::try_new(&path) else {
|
||||
return LineupGuideLookup::Missing;
|
||||
};
|
||||
let mut candidates = Vec::with_capacity(keys.len());
|
||||
for key in keys {
|
||||
match query.query(&key) {
|
||||
Ok(Some(entries)) => candidates.push(LineupGuideCandidate { key, entries }),
|
||||
Ok(None) => {}
|
||||
Err(err) => {
|
||||
log::warn!("Failed to query lineup guide index {}: {err}", path.display());
|
||||
return LineupGuideLookup::Missing;
|
||||
}
|
||||
}
|
||||
}
|
||||
if candidates.is_empty() {
|
||||
LineupGuideLookup::NoEntry
|
||||
} else {
|
||||
LineupGuideLookup::Candidates(candidates)
|
||||
}
|
||||
});
|
||||
|
||||
match task.await {
|
||||
Ok(lookup) => lookup,
|
||||
// A lookup that could not run is "no index", so the caller falls back to its scan.
|
||||
Err(err) => {
|
||||
log::warn!("Lineup guide index lookup task failed: {err}");
|
||||
LineupGuideLookup::Missing
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use shared::{
|
||||
model::{PlaylistItem, PlaylistItemHeader, VirtualId},
|
||||
utils::Internable,
|
||||
};
|
||||
use tempfile::TempDir;
|
||||
use tuliprox_core::utils::FileLockManager;
|
||||
|
||||
fn channel(virtual_id: u32, name: &str, epg: Option<&str>, cluster: XtreamCluster) -> PlaylistItem {
|
||||
PlaylistItem {
|
||||
header: PlaylistItemHeader {
|
||||
virtual_id: VirtualId::new(virtual_id),
|
||||
name: name.intern(),
|
||||
epg_channel_id: epg.map(Internable::intern),
|
||||
xtream_cluster: cluster,
|
||||
..PlaylistItemHeader::default()
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn group(id: u32, cluster: XtreamCluster, channels: Vec<PlaylistItem>) -> PlaylistGroup {
|
||||
PlaylistGroup { id, title: "Group".intern(), channels, xtream_cluster: cluster }
|
||||
}
|
||||
|
||||
/// Index content as `(guide number, candidates)`, ordered for comparison.
|
||||
fn entries(tree: &LineupGuideTree) -> Vec<(String, Vec<(String, u32)>)> {
|
||||
let mut entries: Vec<_> = tree
|
||||
.iter()
|
||||
.map(|(guide, candidates)| {
|
||||
let candidates = candidates.iter().map(|entry| (entry.cluster.to_string(), entry.virtual_id)).collect();
|
||||
(guide.to_string(), candidates)
|
||||
})
|
||||
.collect();
|
||||
entries.sort();
|
||||
entries
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn m3u_view_keeps_every_feed_of_a_guide_number_in_lineup_order() {
|
||||
let playlist = vec![
|
||||
group(1, XtreamCluster::Live, vec![channel(9, "live feed", Some("shared.de"), XtreamCluster::Live)]),
|
||||
group(
|
||||
2,
|
||||
XtreamCluster::Video,
|
||||
vec![
|
||||
channel(4, "vod feed", Some("shared.de"), XtreamCluster::Video),
|
||||
channel(5, "without epg", None, XtreamCluster::Video),
|
||||
],
|
||||
),
|
||||
];
|
||||
|
||||
let tree = build_lineup_guide_index(&playlist, LineupGuideView::M3u);
|
||||
|
||||
// The M3U playlist is one key space; with equal source ordinals the virtual id decides,
|
||||
// and a channel without an EPG id is published under its name.
|
||||
assert_eq!(
|
||||
entries(&tree),
|
||||
vec![
|
||||
("shared.de".to_string(), vec![("video".to_string(), 4), ("live".to_string(), 9)]),
|
||||
("without epg".to_string(), vec![("video".to_string(), 5)]),
|
||||
]
|
||||
);
|
||||
let candidates = tree.query(&"shared.de".intern()).expect("guide number is indexed");
|
||||
assert_eq!(candidates.first().map(|entry| entry.virtual_id), Some(4));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn feeds_follow_the_source_order_of_the_playlist_db_not_the_virtual_id() {
|
||||
let mut hd = channel(3, "HD", Some("shared.de"), XtreamCluster::Live);
|
||||
hd.header.source_ordinal = 1;
|
||||
let mut sd = channel(1, "SD", Some("shared.de"), XtreamCluster::Live);
|
||||
sd.header.source_ordinal = 2;
|
||||
let mut vod = channel(2, "VOD", Some("shared.de"), XtreamCluster::Video);
|
||||
vod.header.source_ordinal = 0;
|
||||
let playlist = vec![group(1, XtreamCluster::Live, vec![sd, hd]), group(2, XtreamCluster::Video, vec![vod])];
|
||||
|
||||
// The dbs are read through their sorted index by source ordinal: the M3U db as one key
|
||||
// space, the Xtream view live before video.
|
||||
let ids = |view| {
|
||||
build_lineup_guide_index(&playlist, view)
|
||||
.query(&"shared.de".intern())
|
||||
.map(|entries| entries.iter().map(|entry| entry.virtual_id).collect::<Vec<_>>())
|
||||
};
|
||||
assert_eq!(ids(LineupGuideView::M3u), Some(vec![2, 3, 1]));
|
||||
assert_eq!(ids(LineupGuideView::Xtream), Some(vec![3, 1, 2]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn m3u_view_leaves_series_info_out() {
|
||||
let mut series = channel(1, "series", Some("series.de"), XtreamCluster::Series);
|
||||
series.header.item_type = PlaylistItemType::SeriesInfo;
|
||||
let playlist = vec![group(1, XtreamCluster::Series, vec![series])];
|
||||
|
||||
// `m3u_write_playlist` does not store series info, so the index must not point at it.
|
||||
assert!(entries(&build_lineup_guide_index(&playlist, LineupGuideView::M3u)).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn xtream_view_lists_live_before_video_and_leaves_series_out() {
|
||||
let playlist = vec![
|
||||
group(1, XtreamCluster::Live, vec![channel(9, "live feed", Some("shared.de"), XtreamCluster::Live)]),
|
||||
group(
|
||||
2,
|
||||
XtreamCluster::Video,
|
||||
vec![
|
||||
channel(4, "vod feed", Some("shared.de"), XtreamCluster::Video),
|
||||
channel(5, "smaller id", Some("only.vod"), XtreamCluster::Video),
|
||||
channel(6, "smaller id again", Some("only.vod"), XtreamCluster::Video),
|
||||
],
|
||||
),
|
||||
group(3, XtreamCluster::Series, vec![channel(1, "series", Some("series.de"), XtreamCluster::Series)]),
|
||||
];
|
||||
|
||||
let tree = build_lineup_guide_index(&playlist, LineupGuideView::Xtream);
|
||||
|
||||
// The lineup searches the live db before the video db, and the whole series db is not part
|
||||
// of the lineup at all. Within one cluster the smaller id comes first.
|
||||
assert_eq!(
|
||||
entries(&tree),
|
||||
vec![
|
||||
("only.vod".to_string(), vec![("video".to_string(), 5), ("video".to_string(), 6)]),
|
||||
("shared.de".to_string(), vec![("live".to_string(), 9), ("video".to_string(), 4)]),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_epg_channel_id_is_not_indexed_under_the_name() {
|
||||
let playlist = vec![group(
|
||||
1,
|
||||
XtreamCluster::Live,
|
||||
vec![channel(1, "named", Some(""), XtreamCluster::Live), channel(2, "", None, XtreamCluster::Live)],
|
||||
)];
|
||||
|
||||
let tree = build_lineup_guide_index(&playlist, LineupGuideView::M3u);
|
||||
|
||||
// `lineup.json` publishes the empty EPG id as the guide number of the first channel, which
|
||||
// no request can ask for, so neither the empty id nor the name may answer for it.
|
||||
assert!(entries(&tree).is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn index_round_trips_and_remove_tolerates_missing_files() {
|
||||
let temp = TempDir::new().expect("temp dir created");
|
||||
let path = temp.path().join("lineup_guide.db");
|
||||
let locks = FileLockManager::default();
|
||||
let playlist =
|
||||
vec![group(1, XtreamCluster::Live, vec![channel(7, "VOX", Some("vox.de"), XtreamCluster::Live)])];
|
||||
|
||||
assert_eq!(
|
||||
lineup_guide_candidates(&locks, path.clone(), vec!["vox.de".intern()]).await,
|
||||
LineupGuideLookup::Missing
|
||||
);
|
||||
lineup_guide_index_write(&locks, &playlist, LineupGuideView::M3u, &path, None).await.expect("index written");
|
||||
assert_eq!(
|
||||
lineup_guide_candidates(&locks, path.clone(), vec!["missing.de".intern(), "vox.de".intern()]).await,
|
||||
LineupGuideLookup::Candidates(vec![LineupGuideCandidate {
|
||||
key: "vox.de".intern(),
|
||||
entries: vec![LineupGuideEntry { cluster: XtreamCluster::Live, virtual_id: 7 }],
|
||||
}])
|
||||
);
|
||||
assert_eq!(
|
||||
lineup_guide_candidates(&locks, path.clone(), vec!["missing.de".intern()]).await,
|
||||
LineupGuideLookup::NoEntry
|
||||
);
|
||||
|
||||
lineup_guide_index_remove(&locks, std::slice::from_ref(&path)).await.expect("index removed");
|
||||
lineup_guide_index_remove(&locks, std::slice::from_ref(&path)).await.expect("removing a missing index is fine");
|
||||
assert_eq!(lineup_guide_candidates(&locks, path, vec!["vox.de".intern()]).await, LineupGuideLookup::Missing);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn carry_over_keeps_the_feed_order_of_the_kept_cluster_after_the_rewritten_one() {
|
||||
let temp = TempDir::new().expect("temp dir created");
|
||||
let path = temp.path().join("lineup_guide.db");
|
||||
let locks = FileLockManager::default();
|
||||
let mut first = channel(5, "VOD HD", Some("shared.de"), XtreamCluster::Video);
|
||||
first.header.source_ordinal = 1;
|
||||
let mut second = channel(4, "VOD SD", Some("shared.de"), XtreamCluster::Video);
|
||||
second.header.source_ordinal = 2;
|
||||
let previous = vec![
|
||||
group(1, XtreamCluster::Live, vec![channel(9, "old live", Some("shared.de"), XtreamCluster::Live)]),
|
||||
group(
|
||||
2,
|
||||
XtreamCluster::Video,
|
||||
vec![first, second, channel(6, "film", Some("film.de"), XtreamCluster::Video)],
|
||||
),
|
||||
];
|
||||
lineup_guide_index_write(&locks, &previous, LineupGuideView::Xtream, &path, None).await.expect("index written");
|
||||
|
||||
// The update rewrites live only: the old live entry goes, the video entries stay in order.
|
||||
let carry_over = lineup_guide_index_carry_over(&locks, &path, &[XtreamCluster::Video])
|
||||
.await
|
||||
.expect("previous index readable");
|
||||
let update =
|
||||
vec![group(1, XtreamCluster::Live, vec![channel(7, "new live", Some("shared.de"), XtreamCluster::Live)])];
|
||||
lineup_guide_index_write(&locks, &update, LineupGuideView::Xtream, &path, Some(carry_over))
|
||||
.await
|
||||
.expect("index rewritten");
|
||||
|
||||
let tree = BPlusTreeQuery::<Arc<str>, Vec<LineupGuideEntry>>::try_new(&path)
|
||||
.expect("index opens")
|
||||
.query(&"shared.de".intern());
|
||||
let ids =
|
||||
tree.expect("query runs").map(|entries| entries.iter().map(|entry| entry.virtual_id).collect::<Vec<_>>());
|
||||
assert_eq!(ids, Some(vec![7, 5, 4]));
|
||||
assert!(lineup_guide_index_carry_over(&locks, &temp.path().join("missing.db"), &[XtreamCluster::Video])
|
||||
.await
|
||||
.is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_item_listed_in_several_groups_is_indexed_once() {
|
||||
let mut first = channel(5, "VOX", Some("vox.de"), XtreamCluster::Live);
|
||||
first.header.source_ordinal = 1;
|
||||
let mut other = channel(7, "VOX HD", Some("vox.de"), XtreamCluster::Live);
|
||||
other.header.source_ordinal = 2;
|
||||
let mut again = channel(5, "VOX", Some("vox.de"), XtreamCluster::Live);
|
||||
again.header.source_ordinal = 3;
|
||||
let playlist =
|
||||
vec![group(1, XtreamCluster::Live, vec![first, other]), group(2, XtreamCluster::Live, vec![again])];
|
||||
|
||||
// The second listing of virtual id 5 is not adjacent to the first after sorting, the other
|
||||
// feed lies between them.
|
||||
let ids = build_lineup_guide_index(&playlist, LineupGuideView::M3u)
|
||||
.query(&"vox.de".intern())
|
||||
.map(|entries| entries.iter().map(|entry| entry.virtual_id).collect::<Vec<_>>());
|
||||
assert_eq!(ids, Some(vec![5, 7]));
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
use crate::{
|
||||
ensure_target_storage_path, get_file_path_for_db_index, m3u_get_file_path_for_db, open_playlist_reader,
|
||||
storage_const, user_get_bouquet_filter, LockedReceiverStream,
|
||||
ensure_target_storage_path, get_file_path_for_db_index, m3u_get_file_path_for_db, m3u_get_item_for_stream_id,
|
||||
open_playlist_reader, storage_const, user_get_bouquet_filter, LockedReceiverStream, PlaylistStorageState,
|
||||
};
|
||||
use futures::{Stream, StreamExt};
|
||||
use log::error;
|
||||
@@ -8,14 +8,14 @@ use shared::{
|
||||
create_bitset,
|
||||
error::TuliproxError,
|
||||
model::{
|
||||
ConfigTargetOptions, M3uPlaylistItem, PlaylistItemType, ProxyType, ResourceOutputPolicy, StreamProperties,
|
||||
TargetType, XtreamCluster,
|
||||
ClusterFlags, ConfigTargetOptions, M3uPlaylistItem, PlaylistItemType, ProxyType, ResourceOutputPolicy,
|
||||
StreamProperties, TargetType, XtreamCluster,
|
||||
},
|
||||
utils::{concat_path, extract_extension_from_url, sanitize_sensitive_info, Internable, PROVIDER_SCHEME_PREFIX},
|
||||
};
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
collections::HashMap,
|
||||
collections::{HashMap, HashSet},
|
||||
pin::Pin,
|
||||
sync::Arc,
|
||||
task::{Context, Poll},
|
||||
@@ -283,25 +283,31 @@ async fn rewrite_player_resources_for_output(item: &mut M3uPlaylistItem, ctx: &U
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_lines)]
|
||||
impl M3uPlaylistIterator {
|
||||
pub async fn new(
|
||||
cfg: &AppConfig,
|
||||
target: &ConfigTarget,
|
||||
user: &ProxyUserCredentials,
|
||||
) -> Result<Self, TuliproxError> {
|
||||
// TODO use playlist memory cache, but be aware of sorting !
|
||||
/// Everything the M3U output pipeline needs for one item of one user.
|
||||
///
|
||||
/// The playlist iterators and the single-item lookup below have to answer with the same item for
|
||||
/// the same user, so the per-item checks and the URL rewrite live here and both use them.
|
||||
struct M3uItemPipeline {
|
||||
filter: Option<HashSet<String>>,
|
||||
user_filter: Option<Arc<shared::foundation::Filter>>,
|
||||
output_clusters: ClusterFlags,
|
||||
base_url: String,
|
||||
username: String,
|
||||
password: String,
|
||||
target_id: u16,
|
||||
proxy_type: ProxyType,
|
||||
target_options: Option<ConfigTargetOptions>,
|
||||
encrypt_secret: [u8; 16],
|
||||
input_by_name: HashMap<Arc<str>, Arc<tuliprox_core::model::ConfigInput>>,
|
||||
flags: M3uPlaylistIteratorFlagsSet,
|
||||
}
|
||||
|
||||
impl M3uItemPipeline {
|
||||
async fn new(cfg: &AppConfig, target: &ConfigTarget, user: &ProxyUserCredentials) -> Result<Self, TuliproxError> {
|
||||
let m3u_output = target.get_m3u_output().ok_or_else(|| {
|
||||
TuliproxError::Config(format!("Unexpected failure, missing m3u target output for target {}", target.name))
|
||||
})?;
|
||||
let config = cfg.config.load();
|
||||
let target_path = ensure_target_storage_path(&config, target.name.as_str()).await?;
|
||||
let m3u_path = m3u_get_file_path_for_db(&target_path);
|
||||
|
||||
let iter_lock = cfg.file_locks.read_lock(&m3u_path).await;
|
||||
let bg_lock = cfg.file_locks.read_lock(&m3u_path).await;
|
||||
|
||||
let filter = user_get_bouquet_filter(&config, &user.username, None, TargetType::M3u, XtreamCluster::Live).await;
|
||||
let mut flags = M3uPlaylistIteratorFlagsSet::new();
|
||||
if m3u_output.include_type_in_url {
|
||||
@@ -313,19 +319,116 @@ impl M3uPlaylistIterator {
|
||||
if cfg.is_reverse_proxy_resource_rewrite_enabled() {
|
||||
flags.set(M3uPlaylistIteratorFlags::RewriteResource);
|
||||
}
|
||||
|
||||
let base_url = cfg.get_user_server_info(user).map(|si| si.get_base_url()).unwrap_or_default();
|
||||
let encrypt_secret = cfg.get_reverse_proxy_rewrite_secret().unwrap_or(cfg.encrypt_secret);
|
||||
let username = user.username.clone();
|
||||
let password = user.password.clone();
|
||||
let target_id = target.id;
|
||||
let proxy_type = user.proxy;
|
||||
let output_clusters = user.output_clusters;
|
||||
let user_filter = user.t_filter.clone();
|
||||
let target_options = target.options.clone();
|
||||
let input_by_name: HashMap<Arc<str>, Arc<tuliprox_core::model::ConfigInput>> =
|
||||
cfg.sources.load().inputs.iter().map(|input| (Arc::clone(&input.name), Arc::clone(input))).collect();
|
||||
|
||||
Ok(Self {
|
||||
filter,
|
||||
user_filter: user.t_filter.clone(),
|
||||
output_clusters: user.output_clusters,
|
||||
base_url,
|
||||
username: user.username.clone(),
|
||||
password: user.password.clone(),
|
||||
target_id: target.id,
|
||||
proxy_type: user.proxy,
|
||||
target_options: target.options.clone(),
|
||||
encrypt_secret,
|
||||
input_by_name,
|
||||
flags,
|
||||
})
|
||||
}
|
||||
|
||||
/// The item the M3U output publishes for this user, or `None` when it is not part of it.
|
||||
#[inline]
|
||||
fn apply(&self, item: M3uPlaylistItem) -> Option<M3uPlaylistItem> {
|
||||
if !self.output_clusters.has_cluster(item.item_type) {
|
||||
return None;
|
||||
}
|
||||
|
||||
if let Some(set) = &self.filter {
|
||||
if !set.contains(item.group.as_ref()) {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
|
||||
// per-user content filter (plan AND user), applied post-cache
|
||||
if let Some(user_filter) = &self.user_filter {
|
||||
let pli = shared::model::PlaylistItem::from(&item);
|
||||
let provider = shared::foundation::ValueProvider { pli: &pli, match_as_ascii: false };
|
||||
if !user_filter.filter(&provider) {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
|
||||
let rewrite_ctx =
|
||||
UrlRewriteContext { base_url: &self.base_url, username: &self.username, password: &self.password };
|
||||
Some(apply_rewrite(
|
||||
item,
|
||||
&rewrite_ctx,
|
||||
self.target_id,
|
||||
&self.encrypt_secret,
|
||||
&self.input_by_name,
|
||||
self.target_options.as_ref(),
|
||||
self.flags,
|
||||
self.proxy_type,
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
/// Single item lookup of one user's M3U output.
|
||||
///
|
||||
/// The `HDHomeRun` tuner endpoints resolve the guide numbers they find in the target guide index
|
||||
/// with this instead of walking the playlist. It answers with the same item the iterators publish,
|
||||
/// and the per-user pipeline (bouquet, content filter, url rewrite) is prepared once for all the
|
||||
/// candidates of one request.
|
||||
pub struct M3uUserItemLookup<'a> {
|
||||
pipeline: M3uItemPipeline,
|
||||
cfg: &'a AppConfig,
|
||||
playlists: &'a PlaylistStorageState,
|
||||
target: &'a ConfigTarget,
|
||||
}
|
||||
|
||||
impl<'a> M3uUserItemLookup<'a> {
|
||||
pub async fn new(
|
||||
cfg: &'a AppConfig,
|
||||
playlists: &'a PlaylistStorageState,
|
||||
target: &'a ConfigTarget,
|
||||
user: &ProxyUserCredentials,
|
||||
) -> Result<Self, TuliproxError> {
|
||||
Ok(Self { pipeline: M3uItemPipeline::new(cfg, target, user).await?, cfg, playlists, target })
|
||||
}
|
||||
|
||||
/// The item the output publishes for `stream_id`, or `None` when the playlist does not hold it
|
||||
/// or this user's filters exclude it.
|
||||
pub async fn get(&self, stream_id: u32) -> Option<M3uPlaylistItem> {
|
||||
match m3u_get_item_for_stream_id(stream_id, self.cfg, self.playlists, self.target).await {
|
||||
Ok(item) => self.pipeline.apply(item),
|
||||
Err(err) => {
|
||||
log::debug!("No m3u playlist item {stream_id} for target {}: {err}", self.target.name);
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl M3uPlaylistIterator {
|
||||
pub async fn new(
|
||||
cfg: &AppConfig,
|
||||
target: &ConfigTarget,
|
||||
user: &ProxyUserCredentials,
|
||||
) -> Result<Self, TuliproxError> {
|
||||
// TODO use playlist memory cache, but be aware of sorting !
|
||||
|
||||
let pipeline = M3uItemPipeline::new(cfg, target, user).await?;
|
||||
let config = cfg.config.load();
|
||||
let target_path = ensure_target_storage_path(&config, target.name.as_str()).await?;
|
||||
let m3u_path = m3u_get_file_path_for_db(&target_path);
|
||||
|
||||
let iter_lock = cfg.file_locks.read_lock(&m3u_path).await;
|
||||
let bg_lock = cfg.file_locks.read_lock(&m3u_path).await;
|
||||
|
||||
let m3u_path = m3u_path.clone();
|
||||
let index_path = get_file_path_for_db_index(&m3u_path);
|
||||
let (tx, rx) = mpsc::channel::<Result<(M3uPlaylistItem, bool), TuliproxError>>(256);
|
||||
@@ -358,37 +461,9 @@ impl M3uPlaylistIterator {
|
||||
}
|
||||
};
|
||||
|
||||
if !output_clusters.has_cluster(item.item_type) {
|
||||
let Some(item) = pipeline.apply(item) else {
|
||||
continue;
|
||||
}
|
||||
|
||||
if let Some(set) = &filter {
|
||||
if !set.contains(item.group.as_ref()) {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
// per-user content filter (plan AND user), applied post-cache
|
||||
if let Some(user_filter) = &user_filter {
|
||||
let pli = shared::model::PlaylistItem::from(&item);
|
||||
let provider = shared::foundation::ValueProvider { pli: &pli, match_as_ascii: false };
|
||||
if !user_filter.filter(&provider) {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
let rewrite_ctx = UrlRewriteContext { base_url: &base_url, username: &username, password: &password };
|
||||
|
||||
let item = apply_rewrite(
|
||||
item,
|
||||
&rewrite_ctx,
|
||||
target_id,
|
||||
&encrypt_secret,
|
||||
&input_by_name,
|
||||
target_options.as_ref(),
|
||||
flags,
|
||||
proxy_type,
|
||||
);
|
||||
};
|
||||
|
||||
if let Some(prev) = pending.replace(item) {
|
||||
if tx.blocking_send(Ok((prev, true))).is_err() {
|
||||
|
||||
@@ -9,12 +9,21 @@ use super::{
|
||||
PlaylistStorageState, TargetPlaylistPersistOptions,
|
||||
};
|
||||
use crate::{
|
||||
ensure_target_storage_path, epg_write_for_target, load_input_local_library_playlist, m3u_write_playlist,
|
||||
persist_input_library_playlist, write_strm_playlist,
|
||||
ensure_target_storage_path, epg_write_for_target, lineup_guide_index_carry_over, lineup_guide_index_remove,
|
||||
lineup_guide_index_view, lineup_guide_index_write, load_input_local_library_playlist, m3u_lineup_guide_index_path,
|
||||
m3u_write_playlist, persist_input_library_playlist, write_strm_playlist, xtream_get_file_path,
|
||||
xtream_get_storage_path, xtream_lineup_guide_index_path, LineupGuideCarryOver, LineupGuideView,
|
||||
};
|
||||
use shared::{error::TuliproxError, model::PlaylistGroup};
|
||||
use std::{path::Path, sync::Arc};
|
||||
use tuliprox_core::model::{AppConfig, ConfigTarget, Epg, TargetOutput};
|
||||
use log::warn;
|
||||
use shared::{
|
||||
error::TuliproxError,
|
||||
model::{ClusterFlags, PlaylistGroup, XtreamCluster},
|
||||
};
|
||||
use std::{
|
||||
path::{Path, PathBuf},
|
||||
sync::Arc,
|
||||
};
|
||||
use tuliprox_core::model::{AppConfig, Config, ConfigTarget, Epg, TargetOutput};
|
||||
|
||||
pub async fn persist_playlist(
|
||||
app_config: &Arc<AppConfig>,
|
||||
@@ -113,6 +122,22 @@ async fn persist_playlist_views_with_mode(
|
||||
return Err(vec![error]);
|
||||
}
|
||||
|
||||
// Only the view `lineup.json` is published from keeps a guide index. Any other index would
|
||||
// describe a playlist the reader no longer asks, and a later re-configuration could mistake it
|
||||
// for a current one.
|
||||
//
|
||||
// The guide index is a sidecar: none of its failures may fail the playlist publication or skip
|
||||
// the memory cache reload below. A missing index only makes the tuners scan the playlist.
|
||||
let lineup_view = lineup_guide_index_view(target);
|
||||
let outdated_indexes: Vec<PathBuf> = [LineupGuideView::M3u, LineupGuideView::Xtream]
|
||||
.into_iter()
|
||||
.filter(|view| lineup_view != Some(*view))
|
||||
.filter_map(|view| lineup_guide_index_path(config, target, view, &target_path))
|
||||
.collect();
|
||||
if let Err(err) = lineup_guide_index_remove(&app_config.file_locks, &outdated_indexes).await {
|
||||
warn!("Outdated HDHomeRun guide index of target {} not removed: {err}", target.name);
|
||||
}
|
||||
|
||||
for (output, prepared_output) in target.output.iter().zip(&mut prepared_outputs) {
|
||||
let output_publication_plan = options.publication_plan.with_output_filter(output.filter().is_some());
|
||||
let source_playlist = match output {
|
||||
@@ -129,16 +154,51 @@ async fn persist_playlist_views_with_mode(
|
||||
let allows_empty_base =
|
||||
library_empty.replaces_empty_target() || output_publication_plan.allows_empty_base_output();
|
||||
|
||||
let replace_empty_clusters =
|
||||
options.accepted_empty_clusters | library_empty.replacement_clusters() | curation_empty_clusters;
|
||||
|
||||
// Guide index of the HDHomeRun lineup. It describes the playlist db of this output, so it
|
||||
// follows what the writer below does with that db:
|
||||
// - rewritten: the old index goes before the db and the new one after it. A reader that
|
||||
// loses the index falls back to its playlist scan, while one that kept an index of another
|
||||
// playlist would resolve guide numbers to channels this target no longer serves.
|
||||
// - kept (an empty update that may not replace it): the old index still describes it.
|
||||
// - partly rewritten (one Xtream cluster kept): the entries of the kept cluster are carried
|
||||
// over from the old index, read before it goes. Without an old index to read them from,
|
||||
// the index goes and readers scan until a full update brings it back.
|
||||
let mut guide_index = match (lineup_view, output) {
|
||||
(Some(view @ LineupGuideView::M3u), TargetOutput::M3u(_)) => {
|
||||
// `m3u_write_playlist` keeps the db of an empty playlist it may not publish.
|
||||
let action = if pl.is_empty() && !allows_empty_base {
|
||||
GuideIndexAction::Keep
|
||||
} else {
|
||||
GuideIndexAction::Rebuild(None)
|
||||
};
|
||||
lineup_guide_index_path(config, target, view, &target_path).map(|path| (view, path, action))
|
||||
}
|
||||
(Some(view @ LineupGuideView::Xtream), TargetOutput::Xtream(_)) => {
|
||||
match lineup_guide_index_path(config, target, view, &target_path) {
|
||||
Some(path) => {
|
||||
let action =
|
||||
xtream_guide_index_action(app_config, target, pl, replace_empty_clusters, &path).await;
|
||||
Some((view, path, action))
|
||||
}
|
||||
None => None,
|
||||
}
|
||||
}
|
||||
_ => None,
|
||||
};
|
||||
if let Some((_, path, GuideIndexAction::Rebuild(_) | GuideIndexAction::Remove)) = &guide_index {
|
||||
if let Err(err) = lineup_guide_index_remove(&app_config.file_locks, std::slice::from_ref(path)).await {
|
||||
// The rebuild below still replaces it. Only if that fails too does the old index
|
||||
// stay, which is why both failures are logged.
|
||||
warn!("HDHomeRun guide index of target {} not removed: {err}", target.name);
|
||||
}
|
||||
}
|
||||
|
||||
let result = match output {
|
||||
TargetOutput::Xtream(_xtream_output) => {
|
||||
persist_xtream_target_playlist(
|
||||
app_config,
|
||||
target,
|
||||
pl,
|
||||
options.accepted_empty_clusters | library_empty.replacement_clusters() | curation_empty_clusters,
|
||||
persistence_mode,
|
||||
)
|
||||
.await
|
||||
persist_xtream_target_playlist(app_config, target, pl, replace_empty_clusters, persistence_mode).await
|
||||
}
|
||||
TargetOutput::M3u(m3u_output) => {
|
||||
m3u_write_playlist(app_config, target, m3u_output, &target_path, pl, allows_empty_base).await
|
||||
@@ -151,6 +211,16 @@ async fn persist_playlist_views_with_mode(
|
||||
|
||||
match result {
|
||||
Ok(()) => {
|
||||
if let Some((view, path, GuideIndexAction::Rebuild(carry_over))) = &mut guide_index {
|
||||
if let Err(err) =
|
||||
lineup_guide_index_write(&app_config.file_locks, pl, *view, path, carry_over.take()).await
|
||||
{
|
||||
warn!(
|
||||
"HDHomeRun guide index of target {} not written, tuners scan the playlist: {err}",
|
||||
target.name
|
||||
);
|
||||
}
|
||||
}
|
||||
let allows_empty_output = match output {
|
||||
TargetOutput::Xtream(_) => {
|
||||
!curation_empty_clusters.is_empty()
|
||||
@@ -217,3 +287,83 @@ pub async fn load_input_media_server_playlist(
|
||||
) -> Result<Vec<PlaylistGroup>, TuliproxError> {
|
||||
load_input_local_library_playlist(app_config, file_path).await
|
||||
}
|
||||
|
||||
/// What a persistence step does with the guide index of the output it writes.
|
||||
#[derive(Debug)]
|
||||
enum GuideIndexAction {
|
||||
/// The playlist db is rewritten from `pl`, so is the index, plus the entries of the clusters the
|
||||
/// db keeps.
|
||||
Rebuild(Option<LineupGuideCarryOver>),
|
||||
/// The playlist db stays as it is, and so does the index describing it.
|
||||
Keep,
|
||||
/// Only part of the db is rewritten and no old index describes the rest.
|
||||
Remove,
|
||||
}
|
||||
|
||||
/// Guide index action of the Xtream view, from what `xtream_write_playlist` does per cluster: a
|
||||
/// cluster without items keeps its db unless it may be replaced by an empty one. A cluster that has
|
||||
/// no db yet keeps nothing, so `pl` still describes it. The entries of a kept cluster are read from
|
||||
/// the index at `index_path` now, before that index is removed for the rewrite.
|
||||
async fn xtream_guide_index_action(
|
||||
app_config: &AppConfig,
|
||||
target: &ConfigTarget,
|
||||
pl: &[PlaylistGroup],
|
||||
replace_empty_clusters: ClusterFlags,
|
||||
index_path: &Path,
|
||||
) -> GuideIndexAction {
|
||||
// The writer sorts items into clusters by their own cluster, not by their group's, so the items
|
||||
// are checked. One pass answers both clusters and stops once both are seen.
|
||||
let (mut has_live, mut has_video) = (false, false);
|
||||
for item in pl.iter().flat_map(|group| &group.channels) {
|
||||
match item.header.xtream_cluster {
|
||||
XtreamCluster::Live => has_live = true,
|
||||
XtreamCluster::Video => has_video = true,
|
||||
XtreamCluster::Series => {}
|
||||
}
|
||||
if has_live && has_video {
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
let storage_path = xtream_get_storage_path(&app_config.config.load(), &target.name);
|
||||
let mut rewritten = false;
|
||||
let mut kept = Vec::new();
|
||||
for (cluster, flag, has_items) in
|
||||
[(XtreamCluster::Live, ClusterFlags::Live, has_live), (XtreamCluster::Video, ClusterFlags::Vod, has_video)]
|
||||
{
|
||||
if has_items || replace_empty_clusters.contains(flag) {
|
||||
rewritten = true;
|
||||
continue;
|
||||
}
|
||||
let has_db = match &storage_path {
|
||||
Some(path) => tokio::fs::try_exists(xtream_get_file_path(path, cluster)).await.unwrap_or(true),
|
||||
None => false,
|
||||
};
|
||||
if has_db {
|
||||
kept.push(cluster);
|
||||
}
|
||||
}
|
||||
if kept.is_empty() {
|
||||
GuideIndexAction::Rebuild(None)
|
||||
} else if !rewritten {
|
||||
GuideIndexAction::Keep
|
||||
} else {
|
||||
match lineup_guide_index_carry_over(&app_config.file_locks, index_path, &kept).await {
|
||||
Some(carry_over) => GuideIndexAction::Rebuild(Some(carry_over)),
|
||||
None => GuideIndexAction::Remove,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Guide index file of `view` for `target`.
|
||||
fn lineup_guide_index_path(
|
||||
config: &Config,
|
||||
target: &ConfigTarget,
|
||||
view: LineupGuideView,
|
||||
target_path: &Path,
|
||||
) -> Option<PathBuf> {
|
||||
match view {
|
||||
LineupGuideView::M3u => Some(m3u_lineup_guide_index_path(target_path)),
|
||||
LineupGuideView::Xtream => xtream_lineup_guide_index_path(config, &target.name),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
use crate::{
|
||||
get_file_path_for_db_index, open_playlist_reader, user_get_bouquet_filter, xtream_get_file_path,
|
||||
xtream_get_storage_path, LockedReceiverStream,
|
||||
xtream_get_item_for_stream_id, xtream_get_storage_path, LockedReceiverStream, PlaylistStorageState,
|
||||
};
|
||||
use futures::{Stream, StreamExt};
|
||||
use log::error;
|
||||
@@ -16,7 +16,7 @@ use std::{
|
||||
};
|
||||
use tokio::{sync::mpsc, task};
|
||||
use tuliprox_core::{
|
||||
model::{xtream_mapping_option_from_target_options, AppConfig, ConfigTarget, ProxyUserCredentials},
|
||||
model::{xtream_mapping_option_from_target_options, AppConfig, Config, ConfigTarget, ProxyUserCredentials},
|
||||
utils::prepare_xtream_resource_hosts,
|
||||
};
|
||||
|
||||
@@ -28,6 +28,69 @@ fn is_cluster_allowed_for_user(user: &ProxyUserCredentials, cluster: XtreamClust
|
||||
user.allows_cluster(cluster)
|
||||
}
|
||||
|
||||
/// Bouquet and content filter of one user for one Xtream cluster.
|
||||
///
|
||||
/// The playlist iterator and the single-item lookup below have to answer with the same item for
|
||||
/// the same user, so the per-item checks live here and both use them.
|
||||
struct XtreamItemFilter {
|
||||
cluster: XtreamCluster,
|
||||
filter_ids: Option<HashSet<u32>>,
|
||||
user_filter: Option<Arc<shared::foundation::Filter>>,
|
||||
}
|
||||
|
||||
impl XtreamItemFilter {
|
||||
async fn new(
|
||||
config: &Config,
|
||||
cluster: XtreamCluster,
|
||||
category_id: Option<u32>,
|
||||
user: &ProxyUserCredentials,
|
||||
) -> Self {
|
||||
let filter = user_get_bouquet_filter(config, &user.username, category_id, TargetType::Xtream, cluster).await;
|
||||
// Parse bouquet filter (strings) once into u32 set to minimize per-item allocations
|
||||
let filter_ids: Option<HashSet<u32>> = filter.as_ref().map(|set| {
|
||||
set.iter()
|
||||
.filter_map(|s| {
|
||||
s.parse::<u32>()
|
||||
.map_err(|e| {
|
||||
error!("Failed to parse bouquet filter id '{s}': {e}");
|
||||
e
|
||||
})
|
||||
.ok()
|
||||
})
|
||||
.collect()
|
||||
});
|
||||
Self { cluster, filter_ids, user_filter: user.t_filter.clone() }
|
||||
}
|
||||
|
||||
#[inline]
|
||||
fn matches(&self, item: &XtreamPlaylistItem) -> bool {
|
||||
// We can't serve episodes within series
|
||||
if self.cluster == XtreamCluster::Series
|
||||
&& !matches!(item.item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
// category_id-Filter
|
||||
if let Some(set) = &self.filter_ids {
|
||||
if !set.contains(&item.category_id) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
// per-user content filter (plan AND user), applied post-cache
|
||||
if let Some(filter) = &self.user_filter {
|
||||
let pli = shared::model::PlaylistItem::from(item);
|
||||
let provider = shared::foundation::ValueProvider { pli: &pli, match_as_ascii: false };
|
||||
if !filter.filter(&provider) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
impl XtreamPlaylistIterator {
|
||||
fn empty() -> Self {
|
||||
// Ghost Channel pattern
|
||||
@@ -59,26 +122,11 @@ impl XtreamPlaylistIterator {
|
||||
let iter_lock = app_config.file_locks.read_lock(&xtream_path).await;
|
||||
let bg_lock = app_config.file_locks.read_lock(&xtream_path).await;
|
||||
|
||||
let filter =
|
||||
user_get_bouquet_filter(&config, &user.username, category_id, TargetType::Xtream, cluster).await;
|
||||
// Parse bouquet filter (strings) once into u32 set to minimize per-item allocations
|
||||
let filter_ids: Option<HashSet<u32>> = filter.as_ref().map(|set| {
|
||||
set.iter()
|
||||
.filter_map(|s| {
|
||||
s.parse::<u32>()
|
||||
.map_err(|e| {
|
||||
error!("Failed to parse bouquet filter id '{s}': {e}");
|
||||
e
|
||||
})
|
||||
.ok()
|
||||
})
|
||||
.collect()
|
||||
});
|
||||
let item_filter = XtreamItemFilter::new(&config, cluster, category_id, user).await;
|
||||
|
||||
let xtream_path = xtream_path.clone();
|
||||
let index_path = get_file_path_for_db_index(&xtream_path);
|
||||
let (tx, rx) = mpsc::channel::<Result<(XtreamPlaylistItem, bool), TuliproxError>>(256);
|
||||
let user_filter = user.t_filter.clone();
|
||||
|
||||
let xtream_path_for_log = xtream_path.clone();
|
||||
let join_error_tx = tx.clone();
|
||||
@@ -110,7 +158,7 @@ impl XtreamPlaylistIterator {
|
||||
}
|
||||
};
|
||||
|
||||
if !Self::matches_filters(cluster, filter_ids.as_ref(), user_filter.as_ref(), &item) {
|
||||
if !item_filter.matches(&item) {
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -145,37 +193,53 @@ impl XtreamPlaylistIterator {
|
||||
Err(TuliproxError::Config(format!("Failed to find xtream storage for target {}", target.name)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn matches_filters(
|
||||
cluster: XtreamCluster,
|
||||
filter_ids: Option<&HashSet<u32>>,
|
||||
user_filter: Option<&Arc<shared::foundation::Filter>>,
|
||||
item: &XtreamPlaylistItem,
|
||||
) -> bool {
|
||||
// We can't serve episodes within series
|
||||
if cluster == XtreamCluster::Series
|
||||
&& !matches!(item.item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo)
|
||||
/// Single item lookup of one user's Xtream output.
|
||||
///
|
||||
/// The `HDHomeRun` tuner endpoints resolve the guide numbers they find in the target guide index
|
||||
/// with this instead of walking the playlists. It answers with the same item the iterators publish,
|
||||
/// and the bouquet and content filter of every cluster are prepared once for all the candidates of
|
||||
/// one request.
|
||||
pub struct XtreamUserItemLookup<'a> {
|
||||
app_config: &'a Arc<AppConfig>,
|
||||
playlists: &'a PlaylistStorageState,
|
||||
target: &'a ConfigTarget,
|
||||
/// Filters of the requested clusters this user may watch.
|
||||
filters: Vec<XtreamItemFilter>,
|
||||
}
|
||||
|
||||
impl<'a> XtreamUserItemLookup<'a> {
|
||||
pub async fn new(
|
||||
app_config: &'a Arc<AppConfig>,
|
||||
playlists: &'a PlaylistStorageState,
|
||||
target: &'a ConfigTarget,
|
||||
user: &ProxyUserCredentials,
|
||||
clusters: &[XtreamCluster],
|
||||
) -> Self {
|
||||
let config = app_config.config.load();
|
||||
let mut filters = Vec::with_capacity(clusters.len());
|
||||
for &cluster in clusters {
|
||||
if is_cluster_allowed_for_user(user, cluster) {
|
||||
filters.push(XtreamItemFilter::new(&config, cluster, None, user).await);
|
||||
}
|
||||
}
|
||||
Self { app_config, playlists, target, filters }
|
||||
}
|
||||
|
||||
/// The item the output publishes for `stream_id` in `cluster`, or `None` when the cluster is
|
||||
/// not served to this user, the playlist does not hold the id, or this user's filters exclude it.
|
||||
pub async fn get(&self, stream_id: u32, cluster: XtreamCluster) -> Option<XtreamPlaylistItem> {
|
||||
let filter = self.filters.iter().find(|filter| filter.cluster == cluster)?;
|
||||
match xtream_get_item_for_stream_id(stream_id, self.app_config, self.playlists, self.target, Some(cluster))
|
||||
.await
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
// category_id-Filter
|
||||
if let Some(set) = filter_ids {
|
||||
if !set.contains(&item.category_id) {
|
||||
return false;
|
||||
Ok(item) => filter.matches(&item).then_some(item),
|
||||
Err(err) => {
|
||||
log::debug!("No xtream playlist item {stream_id} in {cluster} for target {}: {err}", self.target.name);
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
// per-user content filter (plan AND user), applied post-cache
|
||||
if let Some(filter) = user_filter {
|
||||
let pli = shared::model::PlaylistItem::from(item);
|
||||
let provider = shared::foundation::ValueProvider { pli: &pli, match_as_ascii: false };
|
||||
if !filter.filter(&provider) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
@@ -238,7 +302,7 @@ impl Stream for XtreamPlaylistJsonIterator {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{is_cluster_allowed_for_user, XtreamPlaylistIterator};
|
||||
use super::{is_cluster_allowed_for_user, XtreamItemFilter, XtreamPlaylistIterator};
|
||||
use crate::LockedReceiverStream;
|
||||
use futures::StreamExt;
|
||||
use shared::{
|
||||
@@ -286,16 +350,15 @@ mod tests {
|
||||
shared::foundation::get_filter(r#"NOT Group ~ "^VIP.*""#, None).expect("filter parses"),
|
||||
);
|
||||
|
||||
let without_filter = XtreamItemFilter { cluster: XtreamCluster::Live, filter_ids: None, user_filter: None };
|
||||
let with_filter =
|
||||
XtreamItemFilter { cluster: XtreamCluster::Live, filter_ids: None, user_filter: Some(filter) };
|
||||
|
||||
// no filter: everything passes
|
||||
assert!(XtreamPlaylistIterator::matches_filters(XtreamCluster::Live, None, None, &make_item("VIP Sports")));
|
||||
assert!(without_filter.matches(&make_item("VIP Sports")));
|
||||
// filter hides matching groups, keeps the rest
|
||||
assert!(!XtreamPlaylistIterator::matches_filters(
|
||||
XtreamCluster::Live,
|
||||
None,
|
||||
Some(&filter),
|
||||
&make_item("VIP Sports")
|
||||
));
|
||||
assert!(XtreamPlaylistIterator::matches_filters(XtreamCluster::Live, None, Some(&filter), &make_item("News")));
|
||||
assert!(!with_filter.matches(&make_item("VIP Sports")));
|
||||
assert!(with_filter.matches(&make_item("News")));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
Reference in New Issue
Block a user