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

Add shared streaming diagnostics

Browse Source
This commit is contained in:
DarkBreakpoint
2025-11-17 08:57:29 -06:00
parent e5c6ca9e8d
commit e81f839853
7 changed files with 169 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/api/api_utils.rs
+30 -6
View File
@@ -22,7 +22,7 @@ use axum::response::IntoResponse;
use chrono::{DateTime, Utc};
use futures::{StreamExt, TryStreamExt};
use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation};
use log::{debug, error, log_enabled, trace};
use log::{debug, error, info, log_enabled, trace, warn};
use reqwest::header::RETRY_AFTER;
use serde::Serialize;
use shared::model::{Claims, InputFetchMethod, PlaylistEntry, PlaylistItemType, StreamChannel, TargetType, UserConnectionPermission, XtreamCluster};
@@ -1043,6 +1043,7 @@ async fn build_stream_response(
resource_url: &str,
response: reqwest::Response,
) -> axum::response::Response {
let sanitized_resource_url = sanitize_sensitive_info(resource_url);
let status = response.status();
let mut response_builder =
axum::response::Response::builder().status(status);
@@ -1070,18 +1071,41 @@ async fn build_stream_response(
// Cache only complete responses (200 OK without Content-Range)
let can_cache = status == axum::http::StatusCode::OK && !has_content_range;
if can_cache {
info!(
"Caching eligible resource stream {}",
sanitized_resource_url
);
let cache_resource_path = if let Some(cache) = app_state.cache.load().as_ref() {
Some(cache.lock().await.store_path(resource_url))
} else {
None
};
if let Some(resource_path) = cache_resource_path {
if let Ok(file) = create_new_file_for_write(&resource_path).await {
let writer = BufWriter::new(file);
let add_cache_content = get_add_cache_content(resource_url, &app_state.cache);
let stream = PersistPipeStream::new(byte_stream, writer, add_cache_content);
return try_unwrap_body!(response_builder.body(axum::body::Body::from_stream(stream)));
match create_new_file_for_write(&resource_path).await {
Ok(file) => {
info!(
"Persisting resource stream {} to {}",
sanitized_resource_url,
resource_path.display()
);
let writer = BufWriter::new(file);
let add_cache_content = get_add_cache_content(resource_url, &app_state.cache);
let stream = PersistPipeStream::new(byte_stream, writer, add_cache_content);
return try_unwrap_body!(response_builder.body(axum::body::Body::from_stream(stream)));
}
Err(err) => {
warn!(
"Failed to create cache file {} for {}: {err}",
resource_path.display(),
sanitized_resource_url
);
}
}
} else {
debug!(
"Resource cache unavailable; streaming response for {} without persistence",
sanitized_resource_url
);
}
}