mirror of
https://github.com/euzu/tuliprox.git
synced 2026-10-27 09:12:35 +01:00
Merge pull request #292 from euzu/feature/shared_streams_uhd
Refactoring shared streams for UHD streams
This commit is contained in:
euzu
Generated
+44
-43
@@ -13,9 +13,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "adler2"
|
||||
version = "2.0.0"
|
||||
version = "2.0.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "512761e0bb2578dd7380c6baaa0f4ce03e84f95e960231d1dec8bf4d7d6e2627"
|
||||
checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa"
|
||||
|
||||
[[package]]
|
||||
name = "aho-corasick"
|
||||
@@ -143,9 +143,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "async-compression"
|
||||
version = "0.4.23"
|
||||
version = "0.4.24"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b37fc50485c4f3f736a4fb14199f6d5f5ba008d7f28fe710306c92780f004c07"
|
||||
checksum = "d615619615a650c571269c00dca41db04b9210037fa76ed8239f70404ab56985"
|
||||
dependencies = [
|
||||
"brotli",
|
||||
"flate2",
|
||||
@@ -379,9 +379,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "cfg-if"
|
||||
version = "1.0.0"
|
||||
version = "1.0.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
|
||||
checksum = "9555578bc9e57714c812a1f84e4fc5b4d21fcb063490c624de019f7464c91268"
|
||||
|
||||
[[package]]
|
||||
name = "cfg_aliases"
|
||||
@@ -405,9 +405,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "clap"
|
||||
version = "4.5.39"
|
||||
version = "4.5.40"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fd60e63e9be68e5fb56422e397cf9baddded06dae1d2e523401542383bc72a9f"
|
||||
checksum = "40b6887a1d8685cebccf115538db5c0efe625ccac9696ad45c409d96566e910f"
|
||||
dependencies = [
|
||||
"clap_builder",
|
||||
"clap_derive",
|
||||
@@ -415,9 +415,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "clap_builder"
|
||||
version = "4.5.39"
|
||||
version = "4.5.40"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "89cc6392a1f72bbeb820d71f32108f61fdaf18bc526e1d23954168a67759ef51"
|
||||
checksum = "e0c66c08ce9f0c698cbce5c0279d0bb6ac936d8674174fe48f736533b964f59e"
|
||||
dependencies = [
|
||||
"anstream",
|
||||
"anstyle",
|
||||
@@ -427,9 +427,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "clap_derive"
|
||||
version = "4.5.32"
|
||||
version = "4.5.40"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "09176aae279615badda0765c0c0b3f6ed53f4709118af73cf4655d85d1530cd7"
|
||||
checksum = "d2c7947ae4cc3d851207c1adb5b5e260ff0cca11446b1d6d1423788e442257ce"
|
||||
dependencies = [
|
||||
"heck",
|
||||
"proc-macro2",
|
||||
@@ -439,9 +439,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "clap_lex"
|
||||
version = "0.7.4"
|
||||
version = "0.7.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f46ad14479a25103f283c0f10005961cf086d8dc42205bb44c46ac563475dca6"
|
||||
checksum = "b94f61472cee1439c0b966b47e3aca9ae07e45d070759512cd390ea2bebc6675"
|
||||
|
||||
[[package]]
|
||||
name = "colorchoice"
|
||||
@@ -546,9 +546,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "curl"
|
||||
version = "0.4.47"
|
||||
version = "0.4.48"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d9fb4d13a1be2b58f14d60adba57c9834b78c62fd86c3e76a148f732686e9265"
|
||||
checksum = "9e2d5c8f48d9c0c23250e52b55e82a6ab4fdba6650c931f5a0a57a43abda812b"
|
||||
dependencies = [
|
||||
"curl-sys",
|
||||
"libc",
|
||||
@@ -556,14 +556,14 @@ dependencies = [
|
||||
"openssl-sys",
|
||||
"schannel",
|
||||
"socket2",
|
||||
"windows-sys 0.52.0",
|
||||
"windows-sys 0.59.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "curl-sys"
|
||||
version = "0.4.80+curl-8.12.1"
|
||||
version = "0.4.82+curl-8.14.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "55f7df2eac63200c3ab25bde3b2268ef2ee56af3d238e76d61f01c3c49bff734"
|
||||
checksum = "c4d63638b5ec65f1a4ae945287b3fd035be4554bbaf211901159c9a2a74fb5be"
|
||||
dependencies = [
|
||||
"cc",
|
||||
"libc",
|
||||
@@ -572,7 +572,7 @@ dependencies = [
|
||||
"openssl-sys",
|
||||
"pkg-config",
|
||||
"vcpkg",
|
||||
"windows-sys 0.52.0",
|
||||
"windows-sys 0.59.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -809,9 +809,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "flate2"
|
||||
version = "1.1.1"
|
||||
version = "1.1.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7ced92e76e966ca2fd84c8f7aa01a4aea65b0eb6648d72f7c8f3e2764a67fece"
|
||||
checksum = "4a3d7db9596fecd151c5f638c0ee5d5bd487b6e0ea232e5dc96d5250f6f94b1d"
|
||||
dependencies = [
|
||||
"crc32fast",
|
||||
"miniz_oxide",
|
||||
@@ -1069,9 +1069,9 @@ checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
|
||||
|
||||
[[package]]
|
||||
name = "hashbrown"
|
||||
version = "0.15.3"
|
||||
version = "0.15.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "84b26c544d002229e640969970a2e74021aadf6e2f96372b9c58eff97de08eb3"
|
||||
checksum = "5971ac85611da7067dbfcabef3c70ebb5606018acd9e2a3903a0da507521e0d5"
|
||||
|
||||
[[package]]
|
||||
name = "heck"
|
||||
@@ -1165,9 +1165,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "hyper-rustls"
|
||||
version = "0.27.6"
|
||||
version = "0.27.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "03a01595e11bdcec50946522c32dde3fc6914743000a68b93000965f2f02406d"
|
||||
checksum = "e3c93eb611681b207e1fe55d5a71ecf91572ec8a6705cdb6857f7d8d5242cf58"
|
||||
dependencies = [
|
||||
"http 1.3.1",
|
||||
"hyper",
|
||||
@@ -1366,7 +1366,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cea70ddb795996207ad57735b50c5982d8844f38ba9ee5f1aedcfb708a2aa11e"
|
||||
dependencies = [
|
||||
"equivalent",
|
||||
"hashbrown 0.15.3",
|
||||
"hashbrown 0.15.4",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1649,9 +1649,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "miniz_oxide"
|
||||
version = "0.8.8"
|
||||
version = "0.8.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3be647b768db090acb35d5ec5db2b0e1f1de11133ca123b9eacf5137868f892a"
|
||||
checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316"
|
||||
dependencies = [
|
||||
"adler2",
|
||||
]
|
||||
@@ -2013,9 +2013,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "portable-atomic"
|
||||
version = "1.11.0"
|
||||
version = "1.11.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "350e9b48cbc6b0e028b0473b114454c6316e57336ee184ceab6e53f72c178b3e"
|
||||
checksum = "f84267b20a16ea918e43c6a88433c2d54fa145c92a811b5b047ccbe153674483"
|
||||
|
||||
[[package]]
|
||||
name = "portable-atomic-util"
|
||||
@@ -2362,9 +2362,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "rustc-demangle"
|
||||
version = "0.1.24"
|
||||
version = "0.1.25"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "719b953e2095829ee67db738b3bfa9fa368c94900df327b3f07fe6e794d2fe1f"
|
||||
checksum = "989e6739f80c4ad5b13e0fd7fe89531180375b18520cc8c82080e4dc4035b84f"
|
||||
|
||||
[[package]]
|
||||
name = "rustc-hash"
|
||||
@@ -2619,9 +2619,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "smallvec"
|
||||
version = "1.15.0"
|
||||
version = "1.15.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8917285742e9f3e1683f0a9c4e6b57960b7314d0b08d30d1ecd426713ee2eee9"
|
||||
checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
|
||||
|
||||
[[package]]
|
||||
name = "socket2"
|
||||
@@ -2662,9 +2662,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
|
||||
|
||||
[[package]]
|
||||
name = "syn"
|
||||
version = "2.0.101"
|
||||
version = "2.0.102"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8ce2b7fc941b3a24138a0a7cf8e858bfc6a992e7978a068a5c760deb0ed43caf"
|
||||
checksum = "f6397daf94fa90f058bd0fd88429dd9e5738999cca8d701813c80723add80462"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -2984,9 +2984,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tracing-attributes"
|
||||
version = "0.1.28"
|
||||
version = "0.1.29"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "395ae124c09f9e6918a2310af6038fba074bcf474ac352496d5910dd59a2226d"
|
||||
checksum = "1b1ffbcf9c6f6b99d386e7444eb608ba646ae452a36b39737deb9663b610f662"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -2995,9 +2995,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tracing-core"
|
||||
version = "0.1.33"
|
||||
version = "0.1.34"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e672c95779cf947c5311f83787af4fa8fffd12fb27e4993211a84bdfd9610f9c"
|
||||
checksum = "b9d12581f227e93f094d3af2ae690a574abb8a2b9b7a96e7cfe9647b2b617678"
|
||||
dependencies = [
|
||||
"once_cell",
|
||||
]
|
||||
@@ -3032,6 +3032,7 @@ dependencies = [
|
||||
"chrono",
|
||||
"clap",
|
||||
"cron",
|
||||
"dashmap",
|
||||
"deunicode",
|
||||
"enum-iterator",
|
||||
"env_logger",
|
||||
@@ -3078,9 +3079,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "twox-hash"
|
||||
version = "2.1.0"
|
||||
version = "2.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e7b17f197b3050ba473acf9181f7b1d3b66d1cf7356c6cc57886662276e65908"
|
||||
checksum = "8b907da542cbced5261bd3256de1b3a1bf340a3d37f93425a07362a1d687de56"
|
||||
|
||||
[[package]]
|
||||
name = "typenum"
|
||||
|
||||
@@ -63,6 +63,7 @@ bitflags = "2"
|
||||
indexmap = "2"
|
||||
notify = "8"
|
||||
arc-swap = "1"
|
||||
dashmap = "6"
|
||||
#[cfg(target_os = "macos")]
|
||||
libc = "0"
|
||||
#[cfg(target_os = "windows")]
|
||||
|
||||
@@ -4,7 +4,7 @@ use crate::api::model::stream_error::StreamError;
|
||||
use crate::utils::debug_if_enabled;
|
||||
use crate::utils::request::sanitize_sensitive_info;
|
||||
use bytes::Bytes;
|
||||
use futures::stream::BoxStream;
|
||||
use futures::stream::{BoxStream, FuturesUnordered};
|
||||
use futures::{Stream, StreamExt};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc};
|
||||
@@ -12,8 +12,9 @@ use tokio::sync::RwLock;
|
||||
use tokio::sync::mpsc::{Sender};
|
||||
|
||||
use std::pin::Pin;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::task::{Context, Poll};
|
||||
use std::time::{Duration, Instant};
|
||||
use dashmap::DashMap;
|
||||
use log::{trace};
|
||||
use tokio::sync::{mpsc};
|
||||
use tokio::sync::mpsc::error::TrySendError;
|
||||
@@ -42,18 +43,11 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// impl<S> Drop for ReceiverStreamWrapper<S>
|
||||
// {
|
||||
// fn drop(&mut self) {
|
||||
// println!("receiver_dropped");
|
||||
// }
|
||||
// }
|
||||
|
||||
fn convert_stream(stream: BoxStream<Bytes>) -> BoxStream<Result<Bytes, StreamError>> {
|
||||
Box::pin(ReceiverStreamWrapper { stream }.boxed())
|
||||
}
|
||||
|
||||
type SubscriberId = usize;
|
||||
|
||||
/// Represents the state of a shared provider URL.
|
||||
///
|
||||
@@ -61,7 +55,8 @@ fn convert_stream(stream: BoxStream<Bytes>) -> BoxStream<Result<Bytes, StreamErr
|
||||
struct SharedStreamState {
|
||||
headers: Vec<(String, String)>,
|
||||
buf_size: usize,
|
||||
subscribers: Arc<RwLock<Vec<Sender<Bytes>>>>,
|
||||
subscribers: Arc<DashMap<SubscriberId, Sender<Bytes>>>, //Arc<RwLock<Vec<Sender<Bytes>>>>,
|
||||
next_subscriber_id: AtomicUsize,
|
||||
}
|
||||
|
||||
impl SharedStreamState {
|
||||
@@ -70,64 +65,104 @@ impl SharedStreamState {
|
||||
Self {
|
||||
headers,
|
||||
buf_size,
|
||||
subscribers: Arc::new(RwLock::new(Vec::new())),
|
||||
subscribers: Arc::new(DashMap::new()), //Arc::new(RwLock::new(Vec::new())),
|
||||
next_subscriber_id: AtomicUsize::new(1),
|
||||
}
|
||||
}
|
||||
|
||||
async fn subscribe(&self) -> BoxedProviderStream {
|
||||
fn subscribe(&self) -> BoxedProviderStream {
|
||||
let (tx, rx) = mpsc::channel(self.buf_size);
|
||||
self.subscribers.write().await.push(tx);
|
||||
let id = self.next_subscriber_id.fetch_add(1, Ordering::AcqRel);
|
||||
self.subscribers.insert(id, tx);
|
||||
convert_stream(ReceiverStream::new(rx).boxed())
|
||||
}
|
||||
|
||||
fn broadcast<S, E>(&self, stream_url: &str, bytes_stream: S, shared_streams: Arc<SharedStreamManager>)
|
||||
fn broadcast<S, E>(
|
||||
&self,
|
||||
stream_url: &str,
|
||||
bytes_stream: S,
|
||||
shared_streams: Arc<SharedStreamManager>,
|
||||
)
|
||||
where
|
||||
S: Stream<Item=Result<Bytes, E>> + Unpin + 'static + std::marker::Send,
|
||||
E: std::fmt::Debug + std::marker::Send
|
||||
S: Stream<Item = Result<Bytes, E>> + Unpin + 'static + Send,
|
||||
E: std::fmt::Debug + Send,
|
||||
{
|
||||
let mut source_stream = Box::pin(bytes_stream);
|
||||
let subscriber = Arc::clone(&self.subscribers);
|
||||
let subscribers = self.subscribers.clone();
|
||||
let streaming_url = stream_url.to_string();
|
||||
|
||||
let mut tick = tokio::time::interval(Duration::from_millis(5));
|
||||
|
||||
//Spawn a task to forward items from the source stream to the broadcast channel
|
||||
tokio::spawn(async move {
|
||||
while let Some(item) = source_stream.next().await {
|
||||
if let Ok(data) = item {
|
||||
if subscriber.read().await.is_empty() {
|
||||
debug_if_enabled!("No active subscribers. Closing shared provider stream {}", sanitize_sensitive_info(&streaming_url));
|
||||
// Cleanup for removing unused shared streams
|
||||
shared_streams.unregister(&streaming_url).await;
|
||||
break;
|
||||
}
|
||||
let Ok(data) = item else {
|
||||
trace!("Shared stream received Err, skipping...");
|
||||
continue;
|
||||
};
|
||||
|
||||
let start_time = Instant::now();
|
||||
loop {
|
||||
if subscriber.read().await.iter().any(|sender| sender.capacity() > 0) {
|
||||
break;
|
||||
}
|
||||
if start_time.elapsed().as_secs() > 5 {
|
||||
break;
|
||||
}
|
||||
tick.tick().await;
|
||||
}
|
||||
if subscribers.is_empty() {
|
||||
debug_if_enabled!("No active subscribers. Closing shared provider stream {}", sanitize_sensitive_info(&streaming_url));
|
||||
shared_streams.unregister(&streaming_url).await;
|
||||
break;
|
||||
}
|
||||
|
||||
let mut subs = subscriber.write().await;
|
||||
// TODO use drain_filter when stable
|
||||
(*subs).retain(|sender| {
|
||||
match sender.try_send(data.clone()) {
|
||||
Ok(()) => true,
|
||||
Err(TrySendError::Closed(_)) => false,
|
||||
Err(err) => {
|
||||
trace!("broadcast send error {err}");
|
||||
true
|
||||
// Fast-path: at least one has capacity
|
||||
if subscribers.iter().any(|sender| sender.value().capacity() > 0) {
|
||||
// Try sending immediately
|
||||
subscribers.retain(|_id, sender| match sender.try_send(data.clone()) {
|
||||
Ok(()) => true,
|
||||
Err(TrySendError::Closed(_)) => false,
|
||||
Err(err) => {
|
||||
trace!("broadcast try_send error: {:?}", err);
|
||||
true
|
||||
}
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
// All are full → wait until one becomes available
|
||||
let mut futures = FuturesUnordered::new();
|
||||
for sender in subscribers.iter() {
|
||||
let tx = sender.value().clone();
|
||||
futures.push(async move {
|
||||
match tx.reserve().await {
|
||||
Ok(permit) => {
|
||||
drop(permit);
|
||||
Ok(())
|
||||
}
|
||||
Err(_) => Err(()),
|
||||
}
|
||||
});
|
||||
}
|
||||
tick.tick().await;
|
||||
|
||||
let mut has_fillable_subscriber = false;
|
||||
while let Some(result) = futures.next().await {
|
||||
match result {
|
||||
Ok(_) => {
|
||||
has_fillable_subscriber = true;
|
||||
break;
|
||||
}
|
||||
Err(_) => {
|
||||
// ignore; continue waiting
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !has_fillable_subscriber {
|
||||
debug_if_enabled!("All subscribers closed. Shutting down shared provider stream {}",sanitize_sensitive_info(&streaming_url));
|
||||
shared_streams.unregister(&streaming_url).await;
|
||||
break;
|
||||
}
|
||||
|
||||
// Re-validate subscribers and send again
|
||||
subscribers.retain(|_id, sender| match sender.try_send(data.clone()) {
|
||||
Ok(()) => true,
|
||||
Err(TrySendError::Closed(_)) => false,
|
||||
Err(err) => {
|
||||
trace!("broadcast try_send error after reserve: {:?}", err);
|
||||
true
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
debug_if_enabled!("Shared stream exhausted. Closing shared provider stream {}", sanitize_sensitive_info(&streaming_url));
|
||||
shared_streams.unregister(&streaming_url).await;
|
||||
});
|
||||
@@ -156,7 +191,7 @@ impl SharedStreamManager {
|
||||
}
|
||||
|
||||
async fn subscribe_stream(&self, stream_url: &str) -> Option<BoxedProviderStream> {
|
||||
let stream_data = self.shared_streams.read().await.get(stream_url)?.subscribe().await;
|
||||
let stream_data = self.shared_streams.read().await.get(stream_url)?.subscribe();
|
||||
Some(stream_data)
|
||||
}
|
||||
|
||||
@@ -169,7 +204,7 @@ impl SharedStreamManager {
|
||||
stream_url: &str,
|
||||
bytes_stream: S,
|
||||
headers: Vec<(String, String)>,
|
||||
buffer_size: usize,)
|
||||
buffer_size: usize,) -> Option<BoxedProviderStream>
|
||||
where
|
||||
S: Stream<Item=Result<Bytes, E>> + Unpin + 'static + std::marker::Send,
|
||||
E: std::fmt::Debug + std::marker::Send
|
||||
@@ -179,6 +214,7 @@ impl SharedStreamManager {
|
||||
shared_state.broadcast(stream_url, bytes_stream, Arc::clone(&app_state.shared_stream_manager));
|
||||
app_state.shared_stream_manager.register(stream_url, shared_state).await;
|
||||
debug_if_enabled!("Created shared provider stream {}", sanitize_sensitive_info(stream_url));
|
||||
Self::subscribe_shared_stream(app_state, stream_url).await
|
||||
}
|
||||
|
||||
/// Creates a broadcast notify stream for the given URL if a shared stream exists.
|
||||
|
||||
Reference in New Issue
Block a user