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

hls reverse proxy handling refactored

Browse Source
This commit is contained in:
euzu
2025-04-16 14:23:16 +02:00
parent c5eeda2a01
commit b9198c9963
18 changed files with 118 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/endpoints/hls_api.rs
+34 -30
View File
@@ -3,7 +3,7 @@ use crate::api::api_utils::try_option_bad_request;
use crate::api::model::app_state::AppState;
use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, CustomVideoStreamType};
use crate::model::api_proxy::ProxyUserCredentials;
use crate::model::config::{ConfigInput, TargetType};
use crate::model::config::{ConfigInput};
use crate::model::playlist::{PlaylistItemType, XtreamCluster};
use crate::processing::parser::hls::{rewrite_hls, RewriteHlsProps};
use crate::utils::network::request;
@@ -12,44 +12,45 @@ use axum::response::IntoResponse;
use log::{debug, error};
use serde::Deserialize;
use std::sync::Arc;
use crate::utils::crypto_utils;
#[derive(Debug, Deserialize)]
struct HlsApiPathParams {
token: u32,
username: String,
password: String,
input_id: u16,
stream_id: u32,
chunk: u32,
token: String,
}
fn hls_response(hls_content: String) -> impl IntoResponse + Send {
axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, "application/x-mpegURL")
.body(hls_content)
.unwrap()
.into_response()
}
pub(in crate::api) async fn handle_hls_stream_request(app_state: &Arc<AppState>,
user: &ProxyUserCredentials,
hls_url: &str,
virtual_id: u32,
input: &ConfigInput,
target_type: TargetType) -> impl axum::response::IntoResponse + Send {
input: &ConfigInput) -> impl IntoResponse + Send {
let url = replace_url_extension(hls_url, HLS_EXT);
let server_info = app_state.config.get_user_server_info(user).await;
match request::download_text_content(Arc::clone(&app_state.http_client), input, &url, None).await {
Ok((content, response_url)) => {
let hls_token = app_state.hls_cache.new_token();
let rewrite_hls_props = RewriteHlsProps {
base_url: &server_info.get_base_url(),
content: &content,
hls_url: response_url,
virtual_id,
token: hls_token,
target_type,
input_id: input.id,
};
let (hls_entry, hls_content) = rewrite_hls(user, &rewrite_hls_props);
app_state.hls_cache.add_entry(hls_entry).await;
axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, "application/x-mpegurl")
.body(hls_content)
.unwrap()
.into_response()
let hls_content = rewrite_hls(user, &rewrite_hls_props);
hls_response(hls_content).into_response()
}
Err(err) => {
error!("Failed to download m3u8 {}", sanitize_sensitive_info(err.to_string().as_str()));
@@ -73,29 +74,32 @@ async fn hls_api_stream(
return create_custom_video_stream_response(&app_state.config, &CustomVideoStreamType::UserConnectionsExhausted).into_response();
}
let Some(hls_entry) = app_state.hls_cache.get_entry(params.token).await else { return axum::http::StatusCode::BAD_REQUEST.into_response(); };
let Some(hls_url) = hls_entry.get_chunk_url(params.chunk) else { return axum::http::StatusCode::BAD_REQUEST.into_response(); };
let Ok(hls_url) = crypto_utils::decrypt_text(&params.token) else { return axum::http::StatusCode::BAD_REQUEST.into_response(); };
let target_name = &target.name;
let virtual_id = params.stream_id;
let input = try_option_bad_request!(app_state.config.get_input_by_id(hls_entry.input_id), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live));
let input = try_option_bad_request!(app_state.config.get_input_by_id(params.input_id), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live));
if is_hls_url(hls_url) {
return handle_hls_stream_request(&app_state, &user, hls_url, virtual_id, input, hls_entry.target_type).await.into_response();
if is_hls_url(&hls_url) {
return handle_hls_stream_request(&app_state, &user, &hls_url, virtual_id, input).await.into_response();
}
// let (pli_url, input_name) = if hls_entry.target_type == TargetType::Xtream {
// let (pli, _) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id(virtual_id, &app_state.config, target, None), true, format!("Failed to read xtream item for stream id {}", virtual_id));
// (pli.url, pli.input_name)
// } else {
// let pli = try_result_bad_request!(m3u_repository::m3u_get_item_for_stream_id(virtual_id, &app_state.config, target).await, true, format!("Failed to read xtream item for stream id {}", virtual_id));
// (pli.url, pli.input_name)
// };
stream_response(&app_state, hls_url, &req_headers, Some(input), PlaylistItemType::LiveHls, target, &user).await.into_response()
stream_response(&app_state, &hls_url, &req_headers, Some(input), PlaylistItemType::LiveHls, target, &user).await.into_response()
}
pub fn hls_api_register() -> axum::Router<Arc<AppState>> {
axum::Router::new()
.route("/hls/{token}/{username}/{password}/{stream_id}/{chunk}", axum::routing::get(hls_api_stream))
.route("/hls/{username}/{password}/{input_id}/{stream_id}/{token}", axum::routing::get(hls_api_stream))
//cfg.service(web::resource("/hls/{token}/{stream}").route(web::get().to(xtream_player_api_hls_stream)));
//cfg.service(web::resource("/play/{token}/{type}").route(web::get().to(xtream_player_api_play_stream)));
}
#[cfg(test)]
mod tests {
#[test]
fn test_hls_api_register() {
}
}
src/api/endpoints/m3u_api.rs
+1 -1
View File
@@ -105,7 +105,7 @@ async fn m3u_api_stream(
if is_hls_request {
let target_name = &target.name;
let hls_input = try_option_bad_request!(input, true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live));
return handle_hls_stream_request(&app_state, &user, &m3u_item.url, m3u_item.virtual_id, hls_input, TargetType::M3u).await.into_response();
return handle_hls_stream_request(&app_state, &user, &m3u_item.url, m3u_item.virtual_id, hls_input).await.into_response();
}
stream_response(&app_state, m3u_item.url.as_str(), &req_headers, input, m3u_item.item_type, target, &user).await.into_response()
src/api/main_api.rs
-2
View File
@@ -9,7 +9,6 @@ use crate::api::model::active_provider_manager::ActiveProviderManager;
use crate::api::model::active_user_manager::ActiveUserManager;
use crate::api::model::app_state::{AppState, HdHomerunAppState};
use crate::api::model::download::DownloadQueue;
use crate::api::model::hls_cache::HlsCache;
use crate::api::model::streams::shared_stream_manager::SharedStreamManager;
use crate::api::scheduler::start_scheduler;
use crate::model::config::{validate_targets, Config, ProcessTargets, RateLimitConfig, ScheduleConfig};
@@ -138,7 +137,6 @@ fn create_shared_data(cfg: &Arc<Config>) -> AppState {
http_client: Arc::new(client),
downloads: Arc::new(DownloadQueue::new()),
cache,
hls_cache: HlsCache::garbage_collected(),
shared_stream_manager: Arc::new(SharedStreamManager::new()),
active_users,
active_provider,
src/api/model/app_state.rs
-2
View File
@@ -3,7 +3,6 @@ use std::sync::Arc;
use crate::api::model::active_provider_manager::ActiveProviderManager;
use crate::api::model::active_user_manager::ActiveUserManager;
use crate::api::model::download::DownloadQueue;
use crate::api::model::hls_cache::HlsCache;
use crate::api::model::streams::shared_stream_manager::SharedStreamManager;
use crate::model::api_proxy::UserConnectionPermission;
use crate::model::config::{Config};
@@ -17,7 +16,6 @@ pub struct AppState {
pub http_client: Arc<reqwest::Client>,
pub downloads: Arc<DownloadQueue>,
pub cache: Arc<Option<Mutex<LRUResourceCache>>>,
pub hls_cache: Arc<HlsCache>,
pub shared_stream_manager: Arc<SharedStreamManager>,
pub active_users: Arc<ActiveUserManager>,
pub active_provider: Arc<ActiveProviderManager>,
src/api/model/hls_cache.rs
-75
View File
@@ -1,75 +0,0 @@
use log::error;
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::AtomicU32;
use std::time::Duration;
use chrono::Local;
use cron::Schedule;
use tokio::sync::RwLock;
use tokio::time::Instant;
use crate::utils::sys_utils::exit;
use crate::model::hls::HlsEntry;
const EXPIRE_DURATION: u64 = 600; // 10 minutes
const TOKEN_MAX: u32 = u32::MAX - u16::MAX as u32;
fn start_garbage_collector(cache: &Arc<HlsCache>) {
let cache_clone = cache.clone();
tokio::spawn(async move {
match Schedule::from_str("0 */15 * * * * *") {
Ok(schedule) => {
let offset = *Local::now().offset();
loop {
let mut upcoming = schedule.upcoming(offset).take(1);
if let Some(datetime) = upcoming.next() {
tokio::time::sleep_until(tokio::time::Instant::from(crate::api::scheduler::datetime_to_instant(datetime))).await;
cache_clone.gc().await;
}
}
}
Err(err) => exit!("Failed to start scheduler: {}", err)
}
});
}
#[repr(align(64))]
pub struct HlsCache {
pub entries: RwLock<HashMap<u32, HlsEntry>>,
counter: AtomicU32,
}
impl HlsCache {
pub fn garbage_collected() -> Arc<Self> {
let cache = Arc::new(Self {
entries: RwLock::new(HashMap::new()),
counter: AtomicU32::new(1),
});
start_garbage_collector(&cache);
cache
}
pub async fn add_entry(&self, entry: HlsEntry) {
self.entries.write().await.insert(entry.token, entry);
}
pub async fn get_entry(&self, token: u32) -> Option<HlsEntry>{
self.entries.read().await.get(&token).cloned()
}
pub async fn gc(&self) {
let threshold = Instant::now() - Duration::from_secs(EXPIRE_DURATION);
// Remove all expired elements
self.entries.write().await.retain(|_, entry| entry.ts > threshold);
}
pub fn new_token(&self) -> u32 {
let token = self.counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
if token > TOKEN_MAX {
self.counter.store(1, std::sync::atomic::Ordering::SeqCst);
return 1;
}
token
}
}
src/api/model/mod.rs
-1
View File
@@ -8,5 +8,4 @@ pub(in crate::api) mod stream_error;
pub(crate) mod streams;
pub(in crate::api) mod active_user_manager;
pub(in crate::api) mod active_provider_manager;
pub(in crate::api) mod hls_cache;
pub(in crate::api) mod stream;