diff --git a/CHANGELOG.md b/CHANGELOG.md index d1b8dbeb1..20887bc9b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ # 2.2.5 (2025-03-xx) - fixed web ui playlist regexp search - added `web_ui_path` to `config.yml` +- added bandwidth throttle `throttle_kibps` attribute for `reverse_proxy.stream` in `config.yml` + +| Resolution |Framerate| Bitrate (KiB/s) | Quality | +|-----------------|---------|-----------------|-------------| +|480p (854x480) | 30 fps | 100–300 KiB/s | Low-Quality | +|720p (1280x720) | 30 fps | 300–700 KiB/s | HD-Streams | +|1080p (1920x1080)| 30 fps | 700–1500 KiB/s | Full-HD | +|4K (3840x2160) | 30 fps | 2500–6000 KiB/s | Ultra-HD | + # 2.2.4 (2025-03-24) - fixed `connect_timeout_secs:0` prevents connection initiation issue. diff --git a/README.md b/README.md index 2615a4f07..fbb43a59f 100644 --- a/README.md +++ b/README.md @@ -197,9 +197,21 @@ You could be banned from your server. Twice a day should be enough. This configuration is only used for reverse proxy mode. The Reverse Proxy mode can be activated for each user individually. #### 1.6.1 `stream` -Contains settings for the streaming. -- The `retry`option is for transparent reconnections to the provider on provider disconnects or stream errors. -- `buffer`: When buffer is `enabled`, the stream is buffered with the configured `size`. +Attributes: +- `retry` +- `buffer` +- `throttle_kibps` + +##### 1.6.1.1 `retry` +If set to `true` on connection loss to provider, the stream will be reconnected. + +##### 1.6.1.2 `buffer` +Has 2 attributes +- `enabled` +- `size` + +If `enabled` = true The stream is buffered. This is only possible if the provider stream is faster than the consumer. +The stream is buffered with the configured `size`. `size` is the amount of `8192 byte` chunks. In this case the value `1024` means approx `8MB` for `2Mbit/s` stream. - *a.* if `retry` is `false` and `buffer.enabled` is `false` the provider stream is piped as is to the client. @@ -207,6 +219,17 @@ Contains settings for the streaming. - The key difference: the `b.` approach is based on complex stream handling and more memory footprint. +##### 1.6.1.3 `throttle_kibps` in KiB/s +Bandwidth throttle (speed limit). + +| Resolution |Framerate| Bitrate (KiB/s) | Quality | +|-----------------|---------|-----------------|-------------| +|480p (854x480) | 30 fps | 100–300 KiB/s | Low-Quality | +|720p (1280x720) | 30 fps | 300–700 KiB/s | HD-Streams | +|1080p (1920x1080)| 30 fps | 700–1500 KiB/s | Full-HD | +|4K (3840x2160) | 30 fps | 2500–6000 KiB/s | Ultra-HD | + + #### 1.6.2 `cache` LRU-Cache is for resources. If it is `enabled`, the resources/images are persisted in the given `dir`. If the cache size exceeds `size`, In an LRU cache, the least recently used items are evicted to make room for new items if the cache `size`is exceeded. @@ -220,6 +243,7 @@ If you set it `true` `cache` is disabled! Because the cache cant work without re reverse_proxy: resource_rewrite_disabled: false stream: + throttle_kibps: 2000 retry: true buffer: enabled: true diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs index 232c94b65..fbddb01a2 100644 --- a/src/api/api_utils.rs +++ b/src/api/api_utils.rs @@ -72,6 +72,7 @@ macro_rules! try_result_bad_request { pub use try_option_bad_request; pub use try_result_bad_request; use crate::api::model::stream::{BoxedProviderStream, ProviderStreamInfo, ProviderStreamResponse}; +use crate::api::model::streams::throttled_stream::ThrottledStream; use crate::tools::atomic_once_flag::AtomicOnceFlag; pub async fn serve_file(file_path: &Path, mime_type: mime::Mime) -> impl axum::response::IntoResponse + Send { @@ -341,7 +342,20 @@ pub async fn stream_response(app_state: &AppState, for (key, value) in &header_map { response = response.header(key, value); } - response.body(axum::body::Body::from_stream(stream)).unwrap().into_response() + + let throttle_kibps = get_stream_throttle(app_state); + + let body_stream = if throttle_kibps > 0 { + match item_type { + PlaylistItemType::Video | PlaylistItemType::Series | PlaylistItemType::SeriesInfo => { + axum::body::Body::from_stream(ThrottledStream::new(stream.boxed(), throttle_kibps as usize)) + } + _ => axum::body::Body::from_stream(stream) + } + } else { + axum::body::Body::from_stream(stream) + }; + response.body(body_stream).unwrap().into_response() // if content_length > 0 { response_builder.body(SizedStream::new(content_length, stream)) } else { response_builder.streaming(stream) } }; @@ -352,6 +366,14 @@ pub async fn stream_response(app_state: &AppState, axum::http::StatusCode::BAD_REQUEST.into_response() } +fn get_stream_throttle(app_state: &AppState) -> u32 { + app_state.config + .reverse_proxy + .as_ref() + .and_then(|reverse_proxy| reverse_proxy.stream.as_ref()) + .map(|stream| stream.throttle_kibps).map_or(0, |t| t.unwrap_or_default()) +} + async fn shared_stream_response(app_state: &AppState, stream_url: &str, user: &ProxyUserCredentials) -> Option { if let Some(stream) = SharedStreamManager::subscribe_shared_stream(app_state, stream_url).await { debug_if_enabled!("Using shared channel {}", sanitize_sensitive_info(stream_url)); diff --git a/src/api/main_api.rs b/src/api/main_api.rs index bf2d75dcd..a13c4a348 100644 --- a/src/api/main_api.rs +++ b/src/api/main_api.rs @@ -29,6 +29,7 @@ use std::io::ErrorKind; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; +use std::net::SocketAddr; use tokio::sync::Mutex; fn get_web_dir_path(web_ui_enabled: bool, web_root: &str) -> Result { @@ -245,7 +246,7 @@ fn start_hdhomerun(cfg: &Arc, app_state: &Arc, infos: &mut Vec match tokio::net::TcpListener::bind(format!("{}:{}", app_host.clone(), port)).await { Ok(listener) => { - if let Err(err) = axum::serve(listener, router).into_future().await { + if let Err(err) = axum::serve(listener, router.into_make_service_with_connect_info::()).into_future().await { error!("{err}"); } } diff --git a/src/api/model/stream_error.rs b/src/api/model/stream_error.rs index 737fb09cd..ee833c51f 100644 --- a/src/api/model/stream_error.rs +++ b/src/api/model/stream_error.rs @@ -6,7 +6,8 @@ pub enum StreamError { // StdIo(std::io::Error), // ReceiverClosed, ReceiverError(BroadcastStreamRecvError), - LockError(String) + LockError(String), + TokenBucketEmpty } impl StreamError { @@ -28,6 +29,7 @@ impl std::fmt::Display for StreamError { // StreamError::ReceiverClosed => write!(f, "Receiver closed"), StreamError::ReceiverError(e) => write!(f, "Receiver error {e}"), StreamError::LockError(e) => write!(f, "{e}"), + StreamError::TokenBucketEmpty => write!(f, "Throttled Stream bucket empty"), } } } \ No newline at end of file diff --git a/src/api/model/streams/mod.rs b/src/api/model/streams/mod.rs index fbff538c3..2b8506851 100644 --- a/src/api/model/streams/mod.rs +++ b/src/api/model/streams/mod.rs @@ -3,9 +3,10 @@ pub(in crate::api) mod persist_pipe_stream; pub(in crate::api) mod provider_stream_factory; pub(in crate::api) mod shared_stream_manager; pub(in crate::api) mod active_client_stream; +pub(in crate::api) mod throttled_stream; mod timed_client_stream; mod buffered_stream; mod client_stream; mod custom_video_stream; mod readonly_ring_buffer; -mod chunked_buffer; \ No newline at end of file +mod chunked_buffer; diff --git a/src/api/model/streams/throttled_stream.rs b/src/api/model/streams/throttled_stream.rs new file mode 100644 index 000000000..890dee139 --- /dev/null +++ b/src/api/model/streams/throttled_stream.rs @@ -0,0 +1,72 @@ +use bytes::Bytes; +use futures::{Stream}; +use std::{ + pin::Pin, + task::{Context, Poll}, + time::Duration, +}; +use std::future::Future; +use tokio::time::{sleep, Sleep}; +use crate::api::model::stream_error::StreamError; + +pub struct ThrottledStream { + inner: S, + rate_bytes_per_sec: f64, + next_delay: Option>>, +} + +impl ThrottledStream { + pub fn new(inner: S, throttle_kibps: usize) -> Self { + assert!(throttle_kibps > 0, "Rate must be greater than 0"); + let rate_bytes_per_sec = (throttle_kibps as f64) * 1024.0; + Self { + inner, + rate_bytes_per_sec, + next_delay: None, + } + } +} + +impl Stream for ThrottledStream +where + S: Stream> + Unpin, +{ + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + let this = &mut *self; + + // Check if there's an active delay + if let Some(mut delay) = this.next_delay.take() { + match delay.as_mut().poll(cx) { + Poll::Ready(()) => { + // Delay completed, proceed to poll inner stream + } + Poll::Pending => { + // Re-insert the delay and return Pending + this.next_delay = Some(delay); + return Poll::Pending; + } + } + } + + // Poll the inner stream + match Pin::new(&mut this.inner).poll_next(cx) { + Poll::Ready(Some(Ok(bytes))) => { + let len = bytes.len() as f64; + let delay_duration = Duration::from_secs_f64(len / this.rate_bytes_per_sec); + + // Schedule the next delay + this.next_delay = Some(Box::pin(sleep(delay_duration))); + + Poll::Ready(Some(Ok(bytes))) + } + Poll::Ready(Some(Err(e))) => { + // Emit error without delaying + Poll::Ready(Some(Err(e))) + } + Poll::Ready(None) => Poll::Ready(None), + Poll::Pending => Poll::Pending, + } + } +} \ No newline at end of file diff --git a/src/model/config.rs b/src/model/config.rs index 15668b8f7..b0306b551 100644 --- a/src/model/config.rs +++ b/src/model/config.rs @@ -1175,6 +1175,7 @@ pub struct StreamConfig { pub retry: bool, #[serde(default, skip_serializing_if = "Option::is_none")] pub buffer: Option, + pub throttle_kibps: Option, #[serde(default)] pub forced_retry_interval_secs: u32, }