diff --git a/CHANGELOG.md b/CHANGELOG.md index 84d945ee2..0f041a331 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,12 @@ # Changelog # 2.2.3 (2023-04-xx) - hls reverse proxy implemented +- !BREAKING CHANGE! `channel_unavailable_file` is now under `custom_stream_response`, and new file `user_connections_exhausted` added. +```yaml +custom_stream_response: + channel_unavailable: /home/m3u-filter/channel_unavailable.ts + user_connections_exhausted: /home/m3u-filter/user_connections_exhausted.ts +``` - input alias definition for same provider with same content but different credentials ```yaml - sources: diff --git a/bin/build_github_docker.sh b/bin/build_github_docker.sh index 98519de4c..aad1f2b28 100755 --- a/bin/build_github_docker.sh +++ b/bin/build_github_docker.sh @@ -37,7 +37,7 @@ BIN_FILE=${WORKING_DIR}/target/${TARGET}/release/m3u-filter cp "${WORKING_DIR}/target/${TARGET}/release/m3u-filter" "${DOCKER_DIR}/" rm -rf "${DOCKER_DIR}/web" cp -r "${FRONTEND_DIR}/build" "${DOCKER_DIR}/web" -cp "${RESOURCES_DIR}/freeze_frame.ts" "${DOCKER_DIR}/" +cp -r "${RESOURCES_DIR}"/*.ts "${DOCKER_DIR}/" # Get the version from the binary VERSION=$("$BIN_FILE" -V | sed 's/m3u-filter *//') @@ -74,6 +74,6 @@ docker push ghcr.io/euzu/${ALPINE_IMAGE_NAME}:latest echo "Cleaning up build artifacts..." rm -rf "${DOCKER_DIR}/web" rm -f "${DOCKER_DIR}/m3u-filter" -rm -f "${DOCKER_DIR}/freeze_frame.ts" +rm -f "${DOCKER_DIR}"/*.ts echo "Docker images for version ${VERSION} have been successfully built, tagged, and pushed." diff --git a/bin/build_github_docker_aarch64.sh b/bin/build_github_docker_aarch64.sh index 2c43b79b2..d63e2e5ca 100755 --- a/bin/build_github_docker_aarch64.sh +++ b/bin/build_github_docker_aarch64.sh @@ -70,7 +70,7 @@ fi cp "${WORKING_DIR}/target/${TARGET}/release/m3u-filter" "${DOCKER_DIR}/" rm -rf "${DOCKER_DIR}/web" cp -r "${WORKING_DIR}/frontend/build" "${DOCKER_DIR}/web" -cp -r "${RESOURCES_DIR}/freeze_frame.ts" "${DOCKER_DIR}/" +cp -r "${RESOURCES_DIR}"/*.ts "${DOCKER_DIR}/" cd "${DOCKER_DIR}" echo "Building Docker images for version ${VERSION}" @@ -95,6 +95,6 @@ docker push ghcr.io/euzu/${SCRATCH_IMAGE_NAME}:latest echo "Cleaning up build artifacts..." rm -rf "${DOCKER_DIR}/web" rm -f "${DOCKER_DIR}/m3u-filter" -rm -f "${DOCKER_DIR}/freeze_frame.ts" +rm -f "${DOCKER_DIR}"/*.ts -echo "Docker images ghcr.io/euzu/${SCRATCH_IMAGE_NAME}${VERSION} have been successfully built, tagged, and pushed." +echo "Docker images ghcr.io/euzu/${SCRATCH_IMAGE_NAME}:${VERSION} have been successfully built, tagged, and pushed." diff --git a/bin/build_github_docker_beta.sh b/bin/build_github_docker_beta.sh index c23252107..22452e3bd 100755 --- a/bin/build_github_docker_beta.sh +++ b/bin/build_github_docker_beta.sh @@ -47,7 +47,7 @@ fi cp "${WORKING_DIR}/target/${TARGET}/release/m3u-filter" "${DOCKER_DIR}/" rm -rf "${DOCKER_DIR}/web" cp -r "${WORKING_DIR}/frontend/build" "${DOCKER_DIR}/web" -cp -r "${RESOURCES_DIR}/freeze_frame.ts" "${DOCKER_DIR}/" +cp -r "${RESOURCES_DIR}"/*.ts "${DOCKER_DIR}/" cd "${DOCKER_DIR}" echo "Building Docker images for version ${VERSION}" @@ -68,6 +68,6 @@ docker push ghcr.io/euzu/${SCRATCH_IMAGE_NAME}:latest echo "Cleaning up build artifacts..." rm -rf "${DOCKER_DIR}/web" rm -f "${DOCKER_DIR}/m3u-filter" -rm -f "${DOCKER_DIR}/freeze_frame.ts" +rm -f "${DOCKER_DIR}"/*.ts echo "Docker images ghcr.io/euzu/${SCRATCH_IMAGE_NAME}${VERSION} have been successfully built, tagged, and pushed." diff --git a/bin/build_resources.sh b/bin/build_resources.sh index 8011a2391..889100321 100755 --- a/bin/build_resources.sh +++ b/bin/build_resources.sh @@ -1,6 +1,5 @@ #!/usr/bin/env bash - # Function to print usage instructions print_usage() { echo "Usage: $(basename "$0") [-f] [-h]" @@ -22,17 +21,20 @@ while getopts "fh" opt; do esac done -if [ "$flag_force" = false ]; then - if [ -e ./resources/freeze_frame.ts ]; then - echo "Resource exists, skipping creation" - exit; +declare -a resources=("channel_unavailable" "user_connections_exhausted") + +for resource in "${resources[@]}"; do + if [ "$flag_force" = false ]; then + if [ -e "./resources/${resource}.ts" ]; then + echo "Resource ${resource} exists, skipping creation" + continue + fi fi -fi - -if which ffmpeg > /dev/null 2>&1; then - ffmpeg -loop 1 -i ./resources/freeze_frame.jpg -t 10 -r 1 -an -vf "scale=1920:1080" -c:v libx264 -preset veryfast -crf 23 -pix_fmt yuv420p ./resources/freeze_frame.ts -else - echo "ffmpeg not found"; - exit; -fi + if which ffmpeg > /dev/null 2>&1; then + ffmpeg -loop 1 -i "./resources/${resource}.jpg" -t 10 -r 1 -an -vf "scale=1920:1080" -c:v libx264 -preset veryfast -crf 23 -pix_fmt yuv420p "./resources/${resource}.ts" + else + echo "ffmpeg not found" + exit + fi +done diff --git a/bin/release.sh b/bin/release.sh index aeb423003..7fc908637 100755 --- a/bin/release.sh +++ b/bin/release.sh @@ -91,7 +91,7 @@ for PLATFORM in "${!TARGETS[@]}"; do cp "$BIN" "$DIR" cp ../config/*.yml "$DIR" cp -rf "${FRONTEND_DIR}/build" "$DIR"/web - cp "${RESOURCES_DIR}/freeze_frame.ts" "$DIR" + cp -rf "${RESOURCES_DIR}"/*.ts "$DIR" # Create archive for the platform if [[ $PLATFORM == "WINDOWS" ]]; then diff --git a/config/config.yml b/config/config.yml index 1a9a2dff0..bb4c59638 100644 --- a/config/config.yml +++ b/config/config.yml @@ -3,7 +3,10 @@ threads: 0 working_dir: /home/m3u-filter/data backup_dir: /home/m3u-filter/.backup update_on_boot: false -channel_unavailable_file: /home/m3u-filter/freeze_frame.ts +custom_stream_response: + channel_unavailable: /home/m3u-filter/channel_unavailable.ts + user_connections_exhausted: /home/m3u-filter/user_connections_exhausted.ts + # sec min hour day of month month day of week year schedules: - schedule: "0 0 8,12,16,20,22,1 * * * *" diff --git a/docker/Dockerfile b/docker/Dockerfile index e03291493..0af649a0c 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -29,10 +29,15 @@ RUN yarn install --prefer-offline --frozen-lockfile --non-interactive --producti FROM jrottenberg/ffmpeg:latest as resource-build WORKDIR /src COPY resources ./resources -RUN ffmpeg -loop 1 -i ./resources/freeze_frame.jpg -t 10 -r 1 -an \ +RUN ffmpeg -loop 1 -i ./resources/channel_unavailable.jpg -t 10 -r 1 -an \ -vf "scale=1920:1080" \ -c:v libx264 -preset veryfast -crf 23 -pix_fmt yuv420p \ - ./resources/freeze_frame.ts + ./resources/channel_unavailable.ts + +RUN ffmpeg -loop 1 -i ./resources/user_connections_exhausted.jpg -t 10 -r 1 -an \ + -vf "scale=1920:1080" \ + -c:v libx264 -preset veryfast -crf 23 -pix_fmt yuv420p \ + ./resources/user_connections_exhausted.ts \ # Final container (Scratch) FROM scratch as scratch-final @@ -41,7 +46,8 @@ COPY --from=rust-build /usr/share/zoneinfo /usr/share/zoneinfo COPY --from=rust-build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ COPY --from=rust-build /src/target/x86_64-unknown-linux-musl/release/m3u-filter /m3u-filter COPY --from=node-build /app/build /web -COPY --from=resource-build /src/resources/freeze_frame.ts /freeze_frame.ts +COPY --from=resource-build /src/resources/channel_unavailable.ts /channel_unavailable.ts +COPY --from=resource-build /src/resources/user_connections_exhausted.ts /user_connections_exhausted.ts ENTRYPOINT ["/m3u-filter"] CMD ["-s", "-p", "/config"] @@ -51,6 +57,7 @@ RUN apk add --no-cache bash curl ca-certificates tini WORKDIR /app COPY --from=rust-build /src/target/x86_64-unknown-linux-musl/release/m3u-filter m3u-filter COPY --from=node-build /app/build web -COPY --from=resource-build /src/resources/freeze_frame.ts /freeze_frame.ts +COPY --from=resource-build /src/resources/channel_unavailable.ts /channel_unavailable.ts +COPY --from=resource-build /src/resources/user_connections_exhausted.ts /user_connections_exhausted.ts ENTRYPOINT ["/sbin/tini", "--", "/app/m3u-filter"] CMD ["-s", "-p", "/app/config"] diff --git a/docker/Dockerfile-manual b/docker/Dockerfile-manual index a4730964e..44c4e7d38 100644 --- a/docker/Dockerfile-manual +++ b/docker/Dockerfile-manual @@ -9,7 +9,8 @@ COPY --from=build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ COPY ./m3u-filter / COPY ./web /web -COPY ./freeze_frame.ts ./freeze_frame.ts +COPY ./channel_unavailable.ts ./channel_unavailable.ts +COPY ./user_connections_exhausted.ts ./user_connections_exhausted.ts CMD ["./m3u-filter", "-s", "-p", "/config"] diff --git a/resources/freeze_frame.jpg b/resources/channel_unavailable.jpg similarity index 100% rename from resources/freeze_frame.jpg rename to resources/channel_unavailable.jpg diff --git a/resources/user_connections_exhausted.jpg b/resources/user_connections_exhausted.jpg new file mode 100644 index 000000000..3cd076373 Binary files /dev/null and b/resources/user_connections_exhausted.jpg differ diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs index ad5cfa00c..69ea68b12 100644 --- a/src/api/api_utils.rs +++ b/src/api/api_utils.rs @@ -120,7 +120,16 @@ pub async fn get_user_target<'a>(api_req: &'a UserApiRequest, app_state: &'a App get_user_target_by_credentials(username, password, api_req, app_state).await } -fn get_stream_options(app_state: &AppState) -> (bool, u32, u32, bool, usize, bool) { +pub struct StreamOptions { + pub stream_retry: bool, + pub stream_force_retry_secs: u32, + pub stream_connect_timeout_secs: u32, + pub buffer_enabled: bool, + pub buffer_size: usize, + pub pipe_provider_stream: bool +} + +fn get_stream_options(app_state: &AppState) -> StreamOptions { let (stream_retry, stream_force_retry_secs, stream_connect_timeout_secs, buffer_enabled, buffer_size) = app_state .config .reverse_proxy @@ -134,7 +143,7 @@ fn get_stream_options(app_state: &AppState) -> (bool, u32, u32, bool, usize, boo (stream.retry, stream.forced_retry_interval_secs, stream.connect_timeout_secs, buffer_enabled, buffer_size) }); let pipe_provider_stream = !stream_retry && !buffer_enabled; - (stream_retry, stream_force_retry_secs, stream_connect_timeout_secs, buffer_enabled, buffer_size, pipe_provider_stream) + StreamOptions { stream_retry, stream_force_retry_secs, stream_connect_timeout_secs, buffer_enabled, buffer_size, pipe_provider_stream} } // fn get_stream_content_length(provider_response: Option<&(Vec<(String, String)>, StatusCode)>) -> u64 { @@ -160,15 +169,14 @@ pub async fn stream_response(app_state: &AppState, stream_url: &str, } } - let (stream_retry, stream_force_retry_secs, stream_connect_timeout, buffer_enabled, buffer_size, direct_pipe_provider_stream) = - get_stream_options(app_state); + let stream_options = get_stream_options(app_state); get_stream_options(app_state); if let Ok(url) = Url::parse(stream_url) { let event_manager = Arc::clone(&app_state.event_manager); - let (stream_opt, provider_response) = if direct_pipe_provider_stream { - provider_stream::get_provider_pipe_stream(&app_state.config, &app_state.http_client, &url, req_headers, input, item_type).await + let (stream_opt, provider_response) = if stream_options.pipe_provider_stream { + provider_stream::get_provider_pipe_stream(&app_state.config, &app_state.http_client, &url, req_headers, input, item_type, &stream_options).await } else { - let buffer_stream_options = BufferStreamOptions::new(item_type, stream_retry, stream_force_retry_secs, stream_connect_timeout, buffer_enabled, buffer_size, share_stream); + let buffer_stream_options = BufferStreamOptions::new(item_type,share_stream, &stream_options); provider_stream::get_provider_reconnect_buffered_stream(&app_state.config, &app_state.http_client, &url, req_headers, input, buffer_stream_options).await }; if let Some(stream) = stream_opt { @@ -176,7 +184,7 @@ pub async fn stream_response(app_state: &AppState, stream_url: &str, let stream = ActiveClientStream::new(stream, event_manager, &user.username, input.map(|c| c.name.clone())).await; let stream_resp = if share_stream { let shared_headers = provider_response.as_ref().map_or_else(Vec::new, |(h, _)| h.clone()); - SharedStreamManager::subscribe(app_state, stream_url, stream, shared_headers, buffer_size).await; + SharedStreamManager::subscribe(app_state, stream_url, stream, shared_headers, stream_options.buffer_size).await; if let Some(broadcast_stream) = SharedStreamManager::subscribe_shared_stream(app_state, stream_url).await { let (status_code, header_map) = get_stream_response_with_headers(provider_response, stream_url); let mut response = axum::response::Response::builder() @@ -352,4 +360,4 @@ pub fn redirect(url: &str) -> impl IntoResponse { .header("Location", url) .body(axum::body::Body::empty()) .unwrap() -} \ No newline at end of file +} diff --git a/src/api/endpoints/hls_api.rs b/src/api/endpoints/hls_api.rs index 88d999587..1387d4611 100644 --- a/src/api/endpoints/hls_api.rs +++ b/src/api/endpoints/hls_api.rs @@ -1,17 +1,15 @@ use crate::api::api_utils::stream_response; use crate::api::api_utils::try_option_bad_request; use crate::api::model::app_state::AppState; -use crate::api::model::streams::provider_stream::create_freeze_frame_stream; +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::playlist::{PlaylistItemType, XtreamCluster}; use crate::processing::parser::hls::rewrite_hls; -use crate::repository::playlist_repository::HLS_EXT; use crate::utils::network::request; -use crate::utils::network::request::{replace_extension, sanitize_sensitive_info}; +use crate::utils::network::request::{is_hls_url, replace_url_extension, sanitize_sensitive_info, HLS_EXT}; use axum::response::IntoResponse; use log::{debug, error}; -use reqwest::StatusCode; use serde::Deserialize; use std::sync::Arc; @@ -31,7 +29,7 @@ pub(in crate::api) async fn handle_hls_stream_request(app_state: &Arc, virtual_id: u32, input: &ConfigInput, target_type: TargetType) -> impl axum::response::IntoResponse + Send { - let url = replace_extension(hls_url, HLS_EXT); + 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) => { @@ -46,17 +44,7 @@ pub(in crate::api) async fn handle_hls_stream_request(app_state: &Arc, } Err(err) => { error!("Failed to download m3u8 {}", sanitize_sensitive_info(err.to_string().as_str())); - if let Some((stream, (headers, status_code))) - = create_freeze_frame_stream(&app_state.config, &[], StatusCode::BAD_REQUEST) { - let mut builder = axum::response::Response::builder() - .status(status_code); - for (key, value) in headers { - builder = builder.header(key, value); - } - return builder.body(axum::body::Body::from_stream(stream)) - .unwrap().into_response(); - } - axum::http::StatusCode::NO_CONTENT.into_response() + create_custom_video_stream_response(&app_state.config, &CustomVideoStreamType::ChannelUnavailable).into_response() } } } @@ -69,9 +57,13 @@ async fn hls_api_stream( let (user, target) = try_option_bad_request!( app_state.config.get_target_for_user(¶ms.username, ¶ms.password).await, false, format!("Could not find any user {}", params.username)); - if !user.has_permissions(&app_state).await { + if user.permission_denied(&app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } + if user.connections_exhausted(&app_state).await { + return create_custom_video_stream_response(&app_state.config, &CustomVideoStreamType::UserConnectionsExhausted).into_response(); + } + let Some(hls_entry) = app_state.hls_cache.get_entry(¶ms.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(); }; @@ -79,7 +71,7 @@ async fn hls_api_stream( let virtual_id = params.stream_id; let input = try_option_bad_request!(app_state.config.get_input_by_name(&hls_entry.input_name), true, format!("Cant find input for target {target_name}, context {}, stream_id {virtual_id}", XtreamCluster::Live)); - if hls_url.ends_with(HLS_EXT) { + if is_hls_url(hls_url) { return handle_hls_stream_request(&app_state, &user, hls_url, virtual_id, input, hls_entry.target_type.clone()).await.into_response(); } diff --git a/src/api/endpoints/m3u_api.rs b/src/api/endpoints/m3u_api.rs index 18c1649dc..13dc0286e 100644 --- a/src/api/endpoints/m3u_api.rs +++ b/src/api/endpoints/m3u_api.rs @@ -7,14 +7,14 @@ use crate::model::config::TargetType; use crate::model::playlist::{FieldGetAccessor, PlaylistItemType, XtreamCluster}; use crate::repository::m3u_playlist_iterator::{M3U_RESOURCE_PATH, M3U_STREAM_PATH}; use crate::repository::m3u_repository::{m3u_get_item_for_stream_id, m3u_load_rewrite_playlist}; -use crate::repository::playlist_repository::HLS_EXT; -use crate::utils::network::request::{replace_extension, sanitize_sensitive_info}; +use crate::utils::network::request::{replace_url_extension, sanitize_sensitive_info, HLS_EXT}; use crate::utils::debug_if_enabled; use axum::response::IntoResponse; use bytes::Bytes; use futures::stream; use log::{debug, error}; use std::sync::Arc; +use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, CustomVideoStreamType}; async fn m3u_api( api_req: &UserApiRequest, @@ -69,9 +69,12 @@ async fn m3u_api_stream( let virtual_id: u32 = try_result_bad_request!(action_stream_id.trim().parse()); let Some((user, target)) = get_user_target_by_credentials(&username, &password, &api_req, &app_state).await else { return axum::http::StatusCode::BAD_REQUEST.into_response() }; - if !user.has_permissions(&app_state).await { + if user.permission_denied(&app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } + if user.connections_exhausted(&app_state).await { + return create_custom_video_stream_response(&app_state.config, &CustomVideoStreamType::UserConnectionsExhausted).into_response(); + } if !target.has_output(&TargetType::M3u) { return axum::http::StatusCode::BAD_REQUEST.into_response(); @@ -90,7 +93,7 @@ async fn m3u_api_stream( let is_hls_request = m3u_item.item_type == PlaylistItemType::LiveHls || stream_ext.as_deref() == Some(HLS_EXT); if user.proxy == ProxyType::Redirect { - let redirect_url = if is_hls_request { &replace_extension(&m3u_item.url, "m3u8") } else { &m3u_item.url }; + let redirect_url = if is_hls_request { &replace_url_extension(&m3u_item.url, "m3u8") } else { &m3u_item.url }; // TODO alias processing debug_if_enabled!("Redirecting m3u stream request to {}", sanitize_sensitive_info(redirect_url)); return redirect(redirect_url.as_str()).into_response(); @@ -114,7 +117,7 @@ async fn m3u_api_resource( let Ok(m3u_stream_id) = stream_id.parse::() else { return axum::http::StatusCode::BAD_REQUEST.into_response() }; let Some((user, target)) = get_user_target_by_credentials(&username, &password, &api_req, &app_state).await else { return axum::http::StatusCode::BAD_REQUEST.into_response() }; - if !user.has_permissions(&app_state).await { + if user.permission_denied(&app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } diff --git a/src/api/endpoints/user_api.rs b/src/api/endpoints/user_api.rs index 8d20a31fa..9f4151ec2 100644 --- a/src/api/endpoints/user_api.rs +++ b/src/api/endpoints/user_api.rs @@ -47,7 +47,7 @@ async fn playlist_categories( ) -> impl axum::response::IntoResponse + Send { if let Some(username) = get_username_from_auth_header(&token, &app_state) { if let Some((user, target)) = get_user_target_by_username(username.as_str(), &app_state).await { - if !user.has_permissions(&app_state).await { + if user.permission_denied(&app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } let config = &app_state.config; @@ -105,7 +105,7 @@ async fn save_playlist_bouquet( ) -> impl axum::response::IntoResponse + Send { if let Some(username) = get_username_from_auth_header(&token, &app_state) { if let Some((user, target)) = get_user_target_by_username(username.as_str(), &app_state).await { - if !user.has_permissions(&app_state).await { + if user.permission_denied(&app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } match save_user_bouquet(&app_state.config, &target.name, &username, &bouquet).await { @@ -127,7 +127,7 @@ async fn playlist_bouquet( ) -> impl axum::response::IntoResponse + Send { if let Some(username) = get_username_from_auth_header(&token, &app_state) { if let Some((user, _target)) = get_user_target_by_username(username.as_str(), &app_state).await { - if !user.has_permissions(&app_state).await { + if user.permission_denied(&app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } let xtream = load_user_bouquet_as_json(&app_state.config, &username, TargetType::Xtream).await; diff --git a/src/api/endpoints/xmltv_api.rs b/src/api/endpoints/xmltv_api.rs index a71e359fa..2f762bf5a 100644 --- a/src/api/endpoints/xmltv_api.rs +++ b/src/api/endpoints/xmltv_api.rs @@ -175,7 +175,7 @@ async fn xmltv_api( axum::extract::State(app_state): axum::extract::State>, ) -> impl axum::response::IntoResponse + Send { if let Some((user, target)) = get_user_target(&api_req, &app_state).await { - if !user.has_permissions(&app_state).await { + if user.permission_denied(&app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } match get_epg_path_for_target(&app_state.config, target) { diff --git a/src/api/endpoints/xtream_api.rs b/src/api/endpoints/xtream_api.rs index 88a5b6e26..4ecd45a70 100644 --- a/src/api/endpoints/xtream_api.rs +++ b/src/api/endpoints/xtream_api.rs @@ -21,6 +21,7 @@ use crate::api::endpoints::hls_api::handle_hls_stream_request; use crate::api::endpoints::xmltv_api::get_empty_epg_response; use crate::api::model::app_state::AppState; use crate::api::model::request::UserApiRequest; +use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, CustomVideoStreamType}; use crate::api::model::xtream::XtreamAuthorizationResponse; use crate::m3u_filter_error::{str_to_io_error, M3uFilterError, M3uFilterErrorKind}; use crate::model::api_proxy::{ProxyType, ProxyUserCredentials}; @@ -28,13 +29,13 @@ use crate::model::config::TargetType; use crate::model::config::{Config, ConfigInput, ConfigTarget}; use crate::model::playlist::{get_backdrop_path_value, FieldGetAccessor, PlaylistEntry, PlaylistItemType, XtreamCluster, XtreamPlaylistItem}; use crate::model::xtream::{INFO_RESOURCE_PREFIX, INFO_RESOURCE_PREFIX_EPISODE, PROP_BACKDROP_PATH, SEASON_RESOURCE_PREFIX}; -use crate::repository::playlist_repository::{get_target_id_mapping, HLS_EXT}; +use crate::repository::playlist_repository::{get_target_id_mapping}; use crate::repository::storage::{get_target_storage_path, hex_encode}; use crate::repository::{user_repository, xtream_repository}; use crate::repository::xtream_repository::{TAG_EPISODES, TAG_INFO_DATA, TAG_SEASONS_DATA}; use crate::utils::hash_utils::generate_playlist_uuid; use crate::utils::json_utils::get_u32_from_serde_value; -use crate::utils::network::request::{extract_extension_from_url, sanitize_sensitive_info}; +use crate::utils::network::request::{extract_extension_from_url, sanitize_sensitive_info, HLS_EXT}; use crate::utils::network::xtream::{create_vod_info_from_item, ACTION_GET_LIVE_CATEGORIES, ACTION_GET_LIVE_STREAMS, ACTION_GET_SERIES, ACTION_GET_SERIES_CATEGORIES, ACTION_GET_SERIES_INFO, ACTION_GET_VOD_CATEGORIES, ACTION_GET_VOD_INFO, ACTION_GET_VOD_STREAMS}; use crate::utils::json_utils; use crate::utils::debug_if_enabled; @@ -146,9 +147,13 @@ async fn xtream_player_api_stream( stream_req: XtreamApiStreamRequest<'_>, ) -> impl axum::response::IntoResponse + Send { let (user, target) = try_option_bad_request!(get_user_target_by_credentials(stream_req.username, stream_req.password, api_req, app_state).await, false, format!("Could not find any user {}", stream_req.username)); - if !user.has_permissions(app_state).await { + if user.permission_denied(app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } + if user.connections_exhausted(app_state).await { + return create_custom_video_stream_response(&app_state.config, &CustomVideoStreamType::UserConnectionsExhausted).into_response(); + } + let target_name = &target.name; if !target.has_output(&TargetType::Xtream) { debug!("Target has no xtream output {}", target_name); @@ -337,7 +342,7 @@ async fn xtream_player_api_resource( resource_req: XtreamApiStreamRequest<'_>, ) -> impl axum::response::IntoResponse { let (user, target) = try_option_bad_request!(get_user_target_by_credentials(resource_req.username, resource_req.password, api_req, app_state).await, false, format!("Could not find any user {}", resource_req.username)); - if !user.has_permissions(app_state).await { + if user.permission_denied(app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } let target_name = &target.name; @@ -617,7 +622,7 @@ async fn xtream_player_api( return axum::response::Json(get_user_info(&user, app_state).await).into_response(); } - if !user.has_permissions(app_state).await { + if user.permission_denied(app_state) { return axum::http::StatusCode::FORBIDDEN.into_response(); } @@ -719,24 +724,13 @@ async fn xtream_player_api_get( xtream_player_api(api_req, &app_state).await } -// async fn xtream_player_api_get(req: axum::http::Request, -// axum::extract::Query(api_req): axum::extract::Query, -// axum::extract::State(app_state): axum::extract::State>, -// ) -> impl axum::response::IntoResponse + Send { -// xtream_player_api(&req, api_req.into_inner(), &app_state).await -// } + async fn xtream_player_api_post( axum::extract::State(app_state): axum::extract::State>, axum::extract::Form(api_req): axum::extract::Form, ) -> impl axum::response::IntoResponse + Send { xtream_player_api(api_req, &app_state).await } -// async fn xtream_player_api_post(req: axum::http::Request, -// api_req: web::Form, -// axum::extract::State(app_state): axum::extract::State>, -// ) -> impl axum::response::IntoResponse + Send { -// xtream_player_api(&req, api_req.into_inner(), &app_state).await -// } macro_rules! register_xtream_api { ($router:expr, [$($path:expr),*]) => {{ diff --git a/src/api/model/active_provider_manager.rs b/src/api/model/active_provider_manager.rs index 55e3e500f..3a8d22905 100644 --- a/src/api/model/active_provider_manager.rs +++ b/src/api/model/active_provider_manager.rs @@ -648,7 +648,7 @@ mod tests { assert!(lineup.acquire(false).is_some()); // Release one connection - lineup.release(1); + lineup.release("provider7_1"); // After release, one connection should be available assert!(lineup.acquire(false).is_some()); @@ -721,7 +721,7 @@ mod tests { } // Verify that only the capacity of the provider was utilized (2 connections) - assert_eq!(*lineup.provider.current_connections.read(), 2); + assert_eq!(lineup.provider.current_connections.load(Ordering::SeqCst), 2); } } diff --git a/src/api/model/streams/freeze_frame_stream.rs b/src/api/model/streams/custom_video_stream.rs similarity index 93% rename from src/api/model/streams/freeze_frame_stream.rs rename to src/api/model/streams/custom_video_stream.rs index 8b94a1359..4c51dc11d 100644 --- a/src/api/model/streams/freeze_frame_stream.rs +++ b/src/api/model/streams/custom_video_stream.rs @@ -8,13 +8,13 @@ use std::task::{Context, Poll}; const CHUNK_SIZE: usize = 8192; -pub struct FreezeFrameStream { +pub struct CustomVideoStream { buffer: Arc>, buffer_len: usize, current_pos: usize, // Keep track of the current position in the buffer } -impl FreezeFrameStream { +impl CustomVideoStream { pub fn new(buffer: Arc>) -> Self { let buffer_len = buffer.len(); Self { @@ -25,7 +25,7 @@ impl FreezeFrameStream { } } -impl Stream for FreezeFrameStream { +impl Stream for CustomVideoStream { type Item = Result; fn poll_next( diff --git a/src/api/model/streams/mod.rs b/src/api/model/streams/mod.rs index 886de44e4..1edab8c5c 100644 --- a/src/api/model/streams/mod.rs +++ b/src/api/model/streams/mod.rs @@ -6,4 +6,4 @@ pub(in crate::api) mod active_client_stream; mod timed_client_stream; mod buffered_stream; mod client_stream; -mod freeze_frame_stream; \ No newline at end of file +mod custom_video_stream; \ No newline at end of file diff --git a/src/api/model/streams/provider_stream.rs b/src/api/model/streams/provider_stream.rs index 2e3079a65..e0d102bb9 100644 --- a/src/api/model/streams/provider_stream.rs +++ b/src/api/model/streams/provider_stream.rs @@ -1,7 +1,7 @@ -use crate::api::api_utils::{get_headers_from_request, HeaderFilter}; +use crate::api::api_utils::{get_headers_from_request, HeaderFilter, StreamOptions}; use crate::api::model::model_utils::get_response_headers; use crate::api::model::stream_error::StreamError; -use crate::api::model::streams::freeze_frame_stream::FreezeFrameStream; +use crate::api::model::streams::custom_video_stream::CustomVideoStream; use crate::api::model::streams::provider_stream_factory::{create_provider_stream, BufferStreamOptions}; use crate::model::config::{Config, ConfigInput}; use crate::model::playlist::PlaylistItemType; @@ -13,26 +13,63 @@ use futures::TryStreamExt; use log::{debug, error}; use reqwest::StatusCode; use std::sync::Arc; +use std::time::Duration; use axum::http::HeaderMap; +use axum::response::IntoResponse; use url::Url; type BoxedProviderStream = BoxStream<'static, Result>; type ProviderStreamHeader = Vec<(String, String)>; type ProviderStreamResponse = (Option, Option<(ProviderStreamHeader, StatusCode)>); -pub fn create_freeze_frame_stream(cfg: &Config, headers: &[(String, String)], status: StatusCode ) -> Option<(BoxedProviderStream, (ProviderStreamHeader, StatusCode))> { - if let Some(freeze_frame) = cfg.t_channel_unavailable_file.as_ref() { - debug!("Streaming response freeze frame for status {status}"); +pub enum CustomVideoStreamType { + ChannelUnavailable, + UserConnectionsExhausted, + ProviderConnectionsExhausted, +} + +fn create_video_stream(video: Option<&Arc>>, headers: &[(String, String)], log_message: &str) -> Option<(BoxedProviderStream, (ProviderStreamHeader, StatusCode))> { + if let Some(video) = video { + debug!("{}", log_message); let mut response_headers: Vec<(String, String)> = headers.iter() .filter(|(key, _)| !(key.eq("content-type") || key.eq("content-length") || key.contains("range"))) .map(|(key, value)| (key.to_string(), value.to_string())).collect(); response_headers.push(("content-type".to_string(), "video/mp2t".to_string())); - Some((Box::pin(FreezeFrameStream::new(Arc::clone(freeze_frame))), (response_headers, StatusCode::OK))) - } else { + Some((Box::pin(CustomVideoStream::new(Arc::clone(video))), (response_headers, StatusCode::OK))) + } else { None } } +pub fn create_channel_unavailable_stream(cfg: &Config, headers: &[(String, String)], status: StatusCode) -> Option<(BoxedProviderStream, (ProviderStreamHeader, StatusCode))> { + create_video_stream(cfg.t_channel_unavailable_video.as_ref(), headers, &format!("Streaming response channel unavailable for status {status}")) +} + +pub fn create_user_connections_exhausted_stream(cfg: &Config, headers: &[(String, String)]) -> Option<(BoxedProviderStream, (ProviderStreamHeader, StatusCode))> { + create_video_stream(cfg.t_user_connections_exhausted_video.as_ref(), headers, "Streaming response user connections exhausted") +} + +pub fn create_provider_connections_exhausted_stream(cfg: &Config, headers: &[(String, String)]) -> Option<(BoxedProviderStream, (ProviderStreamHeader, StatusCode))> { + create_video_stream(cfg.t_provider_connections_exhausted_video.as_ref(), headers, "Streaming response provider connections exhausted") +} + +pub fn create_custom_video_stream_response(config: &Config, video_response: &CustomVideoStreamType) -> impl axum::response::IntoResponse + Send { + if let Some((stream, (headers, status_code))) = match video_response { + CustomVideoStreamType::ChannelUnavailable => create_channel_unavailable_stream(config, &[], StatusCode::BAD_REQUEST), + CustomVideoStreamType::UserConnectionsExhausted => create_user_connections_exhausted_stream(config, &[]), + CustomVideoStreamType::ProviderConnectionsExhausted => create_provider_connections_exhausted_stream(config, &[]), + } { + let mut builder = axum::response::Response::builder() + .status(status_code); + for (key, value) in headers { + builder = builder.header(key, value); + } + return builder.body(axum::body::Body::from_stream(stream)).unwrap().into_response(); + } + axum::http::StatusCode::FORBIDDEN.into_response() +} + + pub fn get_header_filter_for_item_type(item_type: PlaylistItemType) -> HeaderFilter { match item_type { PlaylistItemType::Live | PlaylistItemType::LiveUnknown | PlaylistItemType::LiveHls => { @@ -47,7 +84,8 @@ pub async fn get_provider_pipe_stream(cfg: &Config, stream_url: &Url, req_headers: &HeaderMap, input: Option<&ConfigInput>, - item_type: PlaylistItemType) -> ProviderStreamResponse { + item_type: PlaylistItemType, + stream_options: &StreamOptions) -> ProviderStreamResponse { let filter_header = get_header_filter_for_item_type(item_type); let req_headers = get_headers_from_request(req_headers, &filter_header); debug_if_enabled!("Stream requested with headers: {:?}", req_headers.iter().map(|header| (header.0, String::from_utf8_lossy(header.1))).collect::>()); @@ -56,14 +94,20 @@ pub async fn get_provider_pipe_stream(cfg: &Config, // The stream url, we need to clone it because of move to async block. // We merge configured input headers with the headers from the request. let headers = get_request_headers(input_headers.as_ref(), Some(&req_headers)); - let client = http_client.get(stream_url.clone()).headers(headers.clone()); + let client_builder = http_client.get(stream_url.clone()).headers(headers.clone()); + let client = if stream_options.stream_connect_timeout_secs > 0 { + client_builder.timeout(Duration::from_secs(u64::from(stream_options.stream_connect_timeout_secs))) + } else { + client_builder + }; match client.send().await { Ok(response) => { let response_headers = get_response_headers(response.headers()); + // TODO hls handling let status = response.status(); if status.is_success() { (Some(Box::pin(response.bytes_stream().map_err(|err| StreamError::reqwest(&err)))), Some((response_headers, status))) - } else if let Some((boxed_provider_stream, response_info)) = create_freeze_frame_stream(cfg, &response_headers, status) { + } else if let Some((boxed_provider_stream, response_info)) = create_channel_unavailable_stream(cfg, &response_headers, status) { (Some(boxed_provider_stream), Some(response_info)) } else { (None, Some((response_headers, status))) @@ -72,7 +116,7 @@ pub async fn get_provider_pipe_stream(cfg: &Config, Err(err) => { let masked_url = sanitize_sensitive_info(stream_url.as_str()); error!("Failed to open stream {masked_url} {err}"); - if let Some((boxed_provider_stream, response_info)) = create_freeze_frame_stream(cfg, &get_response_headers(&headers), StatusCode::BAD_GATEWAY) { + if let Some((boxed_provider_stream, response_info)) = create_channel_unavailable_stream(cfg, &get_response_headers(&headers), StatusCode::BAD_GATEWAY) { (Some(boxed_provider_stream), Some(response_info)) } else { (None, None) diff --git a/src/api/model/streams/provider_stream_factory.rs b/src/api/model/streams/provider_stream_factory.rs index 8d2a35e56..7c1d3f1f0 100644 --- a/src/api/model/streams/provider_stream_factory.rs +++ b/src/api/model/streams/provider_stream_factory.rs @@ -1,9 +1,10 @@ -use crate::api::api_utils::get_headers_from_request; +use crate::api::api_utils::{get_headers_from_request, StreamOptions}; use crate::api::model::model_utils::get_response_headers; use crate::api::model::stream_error::StreamError; use crate::api::model::streams::buffered_stream::BufferedStream; use crate::api::model::streams::client_stream::ClientStream; -use crate::api::model::streams::provider_stream::{create_freeze_frame_stream, get_header_filter_for_item_type}; +use crate::api::model::streams::provider_stream::{create_channel_unavailable_stream, get_header_filter_for_item_type}; +use crate::api::model::streams::timed_client_stream::{TimeoutClientStream}; use crate::model::config::{Config, ConfigInput}; use crate::model::playlist::PlaylistItemType; use crate::tools::atomic_once_flag::AtomicOnceFlag; @@ -20,7 +21,6 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; use url::Url; -use crate::api::model::streams::timed_client_stream::TimedClientStream; // TODO make this configurable pub const STREAM_QUEUE_SIZE: usize = 4096; // mpsc channel holding messages. with possible 8092byte chunks @@ -43,20 +43,16 @@ pub struct BufferStreamOptions { impl BufferStreamOptions { pub(crate) fn new( item_type: PlaylistItemType, - reconnect_enabled: bool, - force_reconnect_secs: u32, - connect_timeout: u32, - buffer_enabled: bool, - buffer_size: usize, - share_stream: bool + share_stream: bool, + stream_options: &StreamOptions ) -> Self { Self { item_type, - reconnect_enabled, - force_reconnect_secs, - connect_timeout_secs: connect_timeout, - buffer_enabled, - buffer_size, + reconnect_enabled: stream_options.stream_retry, + force_reconnect_secs: stream_options.stream_force_retry_secs, + connect_timeout_secs: stream_options.stream_connect_timeout_secs, + buffer_enabled: stream_options.buffer_enabled, + buffer_size: stream_options.buffer_size, share_stream, } } @@ -197,33 +193,44 @@ fn get_client_stream_request_params( (stream_buffer_size, req_range_start_bytes, options.is_reconnect_enabled(), options.force_reconnect_secs, options.connect_timeout_secs, headers) } -fn prepare_client(request_client: &Arc, url: &Url, headers: &HeaderMap, range_start_bytes_to_request: Option) -> (reqwest::RequestBuilder, bool) { +fn prepare_client(request_client: &Arc, stream_options: &ProviderStreamOptions) -> (reqwest::RequestBuilder, bool) { + let url = stream_options.get_url(); + let range_start = stream_options.get_total_bytes_send(); + let headers = stream_options.get_headers(); let mut client = request_client.get(url.clone()).headers(headers.clone()); - if let Some(range) = range_start_bytes_to_request { - // on reconnect send range header to avoid starting from beginning for vod - let range = format!("bytes={range}-", ); - client = client.header(RANGE, range); - (client, true) // partial content + + let (client_builder, partial) = { + if let Some(range) = range_start { + // on reconnect send range header to avoid starting from beginning for vod + let range = format!("bytes={range}-", ); + client = client.header(RANGE, range); + (client, true) // partial content + } else { + (client, false) + } + }; + + let client = if stream_options.connect_timeout_secs > 0 { + client_builder.timeout(Duration::from_secs(u64::from(stream_options.connect_timeout_secs))) } else { - (client, false) - } + client_builder + }; + (client, partial) } -async fn provider_request(cfg: &Config, request_client: Arc, initial_info: bool, stream_options: &ProviderStreamOptions) -> Result, StatusCode> { - let (client, _partial_content) = prepare_client(&request_client, stream_options.get_url(), stream_options.get_headers(), stream_options.get_total_bytes_send()); +async fn provider_initial_request(cfg: &Config, request_client: Arc, stream_options: &ProviderStreamOptions) -> Result, StatusCode> { + let (client, _partial_content) = prepare_client(&request_client, stream_options); match client.send().await { Ok(mut response) => { let status = response.status(); if status.is_success() { - let response_info = if initial_info { + let response_info = { // Unfortunately, the HEAD request does not work, so we need this workaround. // We need some header information from the provider, we extract the necessary headers and forward them to the client debug_if_enabled!("Provider response status: '{}' headers: {:?}", response.status(), response.headers_mut()); let response_headers: Vec<(String, String)> = get_response_headers(response.headers()); // debug!("First headers {headers:?} {} {}", sanitize_sensitive_info(url.as_str())); Some((response_headers, response.status())) - } else { - None }; return Ok(Some((response.bytes_stream().map_err(|err| { error!("Failed to read response body: {err}"); @@ -231,7 +238,7 @@ async fn provider_request(cfg: &Config, request_client: Arc, in }).boxed(), response_info))); } if let Some((boxed_provider_stream, response_info)) = - create_freeze_frame_stream(cfg, &get_response_headers(response.headers()), status) + create_channel_unavailable_stream(cfg, &get_response_headers(response.headers()), status) { return Ok(Some((boxed_provider_stream, Some(response_info)))); } @@ -239,7 +246,7 @@ async fn provider_request(cfg: &Config, request_client: Arc, in } Err(_err) => { if let Some((boxed_provider_stream, response_info)) = - create_freeze_frame_stream(cfg, &get_response_headers(stream_options.get_headers()), StatusCode::BAD_GATEWAY) + create_channel_unavailable_stream(cfg, &get_response_headers(stream_options.get_headers()), StatusCode::BAD_GATEWAY) { Ok(Some((boxed_provider_stream, Some(response_info)))) } else { @@ -252,17 +259,10 @@ async fn provider_request(cfg: &Config, request_client: Arc, in async fn stream_provider(client: Arc, stream_options: ProviderStreamOptions) -> Option { let url = stream_options.get_url(); - let range_start = stream_options.get_total_bytes_send(); - let headers = stream_options.get_headers(); debug_if_enabled!("stream provider {}", sanitize_sensitive_info(url.as_str())); while stream_options.should_continue() { debug_if_enabled!("Reconnecting stream {}", sanitize_sensitive_info(url.as_str())); - let (client_builder, _) = prepare_client(&client, url, headers, range_start); - let client = if stream_options.connect_timeout_secs > 0 { - client_builder.timeout(Duration::from_secs(u64::from(stream_options.connect_timeout_secs))) - } else { - client_builder - }; + let (client, _) = prepare_client(&client, &stream_options); match client.send().await { Ok(response) => { let status = response.status(); @@ -272,10 +272,10 @@ async fn stream_provider(client: Arc, stream_options: ProviderS StreamError::reqwest(&err) }).boxed(); return if stream_options.reconnect_force_secs > 0 { - Some(TimedClientStream::new(provider_stream, stream_options.reconnect_force_secs).boxed()) + Some(TimeoutClientStream::new(provider_stream, stream_options.reconnect_force_secs).boxed()) } else { Some(provider_stream) - } + }; } if status.is_client_error() { return None; @@ -307,7 +307,7 @@ async fn get_initial_stream(cfg: &Config, client: Arc, stream_o let start = Instant::now(); let mut connect_err: u32 = 1; while stream_options.should_continue() { - match provider_request(cfg, Arc::clone(&client), true, stream_options).await { + match provider_initial_request(cfg, Arc::clone(&client), stream_options).await { Ok(Some(value)) => return Some(value), Ok(None) => { if connect_err > ERR_MAX_RETRY_COUNT { @@ -338,7 +338,7 @@ fn create_provider_stream_options(stream_url: &Url, req_headers: &HeaderMap, input: Option<&ConfigInput>, options: &BufferStreamOptions) -> ProviderStreamOptions { - let (buffer_size, req_range_start_bytes, reconnect, reconnect_force_secs, connect_timeout_secs, headers) + let (buffer_size, req_range_start_bytes, reconnect, reconnect_force_secs, connect_timeout_secs, headers) = get_client_stream_request_params(req_headers, input, options); let url = stream_url.clone(); let range_bytes = Arc::new(req_range_start_bytes.map(AtomicUsize::new)); diff --git a/src/api/model/streams/timed_client_stream.rs b/src/api/model/streams/timed_client_stream.rs index 8d98c4339..0195dcb86 100644 --- a/src/api/model/streams/timed_client_stream.rs +++ b/src/api/model/streams/timed_client_stream.rs @@ -6,18 +6,18 @@ use std::pin::Pin; use std::task::Poll; use std::time::{Duration, Instant}; -pub struct TimedClientStream { +pub struct TimeoutClientStream { inner: ResponseStream, duration: Duration, start_time: Instant, } -impl TimedClientStream { +impl TimeoutClientStream { pub(crate) fn new(inner: ResponseStream, duration: u32) -> Self { Self { inner, duration: Duration::from_secs(u64::from(duration)) , start_time: Instant::now() } } } -impl Stream for TimedClientStream { +impl Stream for TimeoutClientStream { type Item = Result; fn poll_next(mut self: Pin<&mut Self>,cx: &mut std::task::Context<'_>,) -> Poll> { diff --git a/src/api/scheduler.rs b/src/api/scheduler.rs index 8f1846286..bac5c69ae 100644 --- a/src/api/scheduler.rs +++ b/src/api/scheduler.rs @@ -44,7 +44,6 @@ pub async fn start_scheduler(client: Arc, expression: &str, con mod tests { use std::str::FromStr; use std::sync::atomic::{AtomicU8, Ordering}; - use std::time::Instant; use chrono::Local; use cron::Schedule; use crate::api::scheduler::datetime_to_instant; diff --git a/src/foundation/filter.rs b/src/foundation/filter.rs index 3fc7128f9..ac4d5c8d3 100644 --- a/src/foundation/filter.rs +++ b/src/foundation/filter.rs @@ -626,24 +626,24 @@ mod tests { assert_eq!(filtered.len(), 2); assert_eq!( filtered.iter().any(|&chan| { - let group = chan.header.borrow().group.to_string(); - let name = chan.header.borrow().name.to_string(); + let group = chan.header.group.to_string(); + let name = chan.header.name.to_string(); name.eq("24/7: Cars") && group.eq("FR Channels") }), true ); assert_eq!( filtered.iter().any(|&chan| { - let group = chan.header.borrow().group.to_string(); - let name = chan.header.borrow().name.to_string(); + let group = chan.header.group.to_string(); + let name = chan.header.name.to_string(); name.eq("Entertainment") && group.eq("US Channels") }), true ); assert_eq!( filtered.iter().any(|&chan| { - let group = chan.header.borrow().group.to_string(); - let name = chan.header.borrow().name.to_string(); + let group = chan.header.group.to_string(); + let name = chan.header.name.to_string(); name.eq("24/7: Cars") && group.eq("US Channels") }), false diff --git a/src/main.rs b/src/main.rs index 4f2a71149..5895a0168 100644 --- a/src/main.rs +++ b/src/main.rs @@ -145,9 +145,9 @@ fn main() { Err(err) => exit!("{err}"), } - if cfg.t_channel_unavailable_file.is_some() { - info!("Freeze frame video loaded from {:?}", cfg.channel_unavailable_file.as_ref().map_or("?", |v| v.as_str())); - } + // if cfg.t_channel_unavailable_video.is_some() { + // info!("Channel unavailable video loaded from {:?}", cfg.channel_unavailable_file.as_ref().map_or("?", |v| v.as_str())); + // } let rt = tokio::runtime::Runtime::new().unwrap(); let () = rt.block_on(async { diff --git a/src/model/api_proxy.rs b/src/model/api_proxy.rs index 11ef8bf5b..a7f441da8 100644 --- a/src/model/api_proxy.rs +++ b/src/model/api_proxy.rs @@ -171,7 +171,7 @@ impl ProxyUserCredentials { Ok(()) } - pub async fn has_permissions(&self, app_state: &AppState) -> bool { + pub fn has_permissions(&self, app_state: &AppState) -> bool { if app_state.config.user_access_control { if let Some(exp_date) = self.exp_date.as_ref() { let now = Local::now(); @@ -180,12 +180,13 @@ impl ProxyUserCredentials { return false; } } - if let Some(max_connections) = self.max_connections.as_ref() { - if *max_connections < app_state.get_active_connections_for_user(&self.username).await { - debug!("User access denied, too many connections: {}", self.username); - return false; - } - } + // we allow requests with max connection reached, but we should block streaming + // if let Some(max_connections) = self.max_connections.as_ref() { + // if *max_connections < app_state.get_active_connections_for_user(&self.username).await { + // debug!("User access denied, too many connections: {}", self.username); + // return false; + // } + // } if let Some(status) = &self.status { if !matches!(status, ProxyUserStatus::Active | ProxyUserStatus::Trial) { debug!("User access denied, status invalid: {status} for user: {}", self.username); @@ -195,6 +196,29 @@ impl ProxyUserCredentials { } true } + + #[inline] + pub fn permission_denied(&self, app_state: &AppState) -> bool { + !self.has_permissions(app_state) + } + + + pub async fn has_connections_left(&self, app_state: &AppState) -> bool { + if app_state.config.user_access_control { + if let Some(max_connections) = self.max_connections.as_ref() { + if *max_connections < app_state.get_active_connections_for_user(&self.username).await { + debug!("User access denied, too many connections: {}", self.username); + return false; + } + } + } + true + } + + #[inline] + pub async fn connections_exhausted(&self, app_state: &AppState) -> bool { + !self.has_connections_left(app_state).await + } } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] diff --git a/src/model/config.rs b/src/model/config.rs index e065c473c..a6885f48f 100644 --- a/src/model/config.rs +++ b/src/model/config.rs @@ -1152,6 +1152,17 @@ impl ReverseProxyConfig { } } +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)] +#[serde(deny_unknown_fields)] +pub struct CustomStreamResponseConfig { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub channel_unavailable: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub user_connection_exhausted: Option, // user has no more connections + #[serde(default, skip_serializing_if = "Option::is_none")] + pub provider_connection_exhausted: Option, // provider limit reached, has no more connections +} + #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)] #[serde(deny_unknown_fields)] pub struct Config { @@ -1165,7 +1176,7 @@ pub struct Config { #[serde(default, skip_serializing_if = "Option::is_none")] pub user_config_dir: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub channel_unavailable_file: Option, + pub custom_stream_response: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub templates: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -1201,7 +1212,11 @@ pub struct Config { #[serde(skip)] pub file_locks: Arc, #[serde(skip)] - pub t_channel_unavailable_file: Option>>, + pub t_channel_unavailable_video: Option>>, + #[serde(skip)] + pub t_user_connections_exhausted_video: Option>>, + #[serde(skip)] + pub t_provider_connections_exhausted_video: Option>>, } impl Config { @@ -1417,54 +1432,72 @@ impl Config { pub fn prepare(&mut self, resolve_var: bool) -> Result<(), M3uFilterError> { let work_dir = if resolve_var { &config_reader::resolve_env_var(&self.working_dir) } else { &self.working_dir }; self.working_dir = file_utils::get_working_path(work_dir); - - if let Some(channel_unavailable_file) = &self.channel_unavailable_file { - let channel_unavailable = file_utils::make_absolute_path(channel_unavailable_file, &self.working_dir, resolve_var); - match file_utils::read_file_as_bytes(&PathBuf::from(&channel_unavailable)) { - Ok(data) => { - self.t_channel_unavailable_file = Some(Arc::new(data)); - } - Err(err) => { - error!("Failed to load channel unavailable file: {channel_unavailable} {err}"); - } - } - self.channel_unavailable_file = Some(channel_unavailable); - } - - if self.backup_dir.is_none() { - self.backup_dir = Some(PathBuf::from(&self.working_dir).join("backup").clean().to_string_lossy().to_string()); - } else { - self.backup_dir = self.backup_dir.as_ref().map(|backup_dir| { - if resolve_var { - config_reader::resolve_env_var(backup_dir) - } else { - backup_dir.to_owned() - } - }).map(|dir| dir.to_string()); - } - if self.user_config_dir.is_none() { - self.user_config_dir = Some(PathBuf::from(&self.working_dir).join("user_config").clean().to_string_lossy().to_string()); - } else { - self.user_config_dir = self.user_config_dir.as_ref().map(|user_config_dir| { - if resolve_var { - config_reader::resolve_env_var(user_config_dir) - } else { - user_config_dir.to_owned() - } - }).map(|dir| dir.to_string()); - } + self.prepare_custom_stream_response(resolve_var); + self.prepare_directories(resolve_var); if let Some(reverse_proxy) = self.reverse_proxy.as_mut() { reverse_proxy.prepare(&self.working_dir, resolve_var); } + self.prepare_hdhomerun()?; + self.api.prepare(); + self.prepare_api_web_root(resolve_var); + self.prepare_templates()?; + self.check_unique_input_names()?; + let target_names = self.check_unique_target_names()?; + self.check_scheduled_targets(&target_names)?; + self.prepare_sources()?; + self.prepare_video_config()?; + self.prepare_web_auth(resolve_var)?; + Ok(()) + } + + fn prepare_directories(&mut self, resolve_var: bool) { + fn set_directory(path: &mut Option, default_subdir: &str, working_dir: &str, resolve_var: bool) { + *path = Some(match path.as_ref() { + Some(existing) => { + if resolve_var { + resolve_env_var(existing) + } else { + existing.to_owned() + } + } + None => PathBuf::from(working_dir).join(default_subdir).clean().to_string_lossy().to_string(), + }); + } + + set_directory(&mut self.backup_dir, "backup", &self.working_dir, resolve_var); + set_directory(&mut self.user_config_dir, "user_config", &self.working_dir, resolve_var); + } + + fn prepare_hdhomerun(&mut self) -> Result<(), M3uFilterError> { if let Some(hdhomerun) = self.hdhomerun.as_mut() { if hdhomerun.enabled { hdhomerun.prepare(self.api.port)?; } } + Ok(()) + } - self.api.prepare(); - self.prepare_api_web_root(resolve_var); + fn prepare_sources(&mut self) -> Result<(), M3uFilterError> { + // prepare sources and set id's + let mut source_index: u16 = 1; + let mut target_index: u16 = 1; + for source in &mut self.sources { + source_index = source.prepare(source_index)?; + for target in &mut source.targets { + // prepare target templates + let prepare_result = match &self.templates { + Some(templ) => target.prepare(target_index, Some(templ)), + _ => target.prepare(target_index, None) + }; + prepare_result?; + target_index += 1; + } + } + Ok(()) + } + + fn prepare_templates(&mut self) -> Result<(), M3uFilterError> { if let Some(templates) = &mut self.templates { match prepare_templates(templates) { Ok(tmplts) => { @@ -1475,25 +1508,25 @@ impl Config { } } }; - // prepare sources and set id's - let mut source_index: u16 = 1; - let mut target_index: u16 = 1; - self.check_unique_input_names()?; - let target_names = self.check_unique_target_names()?; - self.check_scheduled_targets(&target_names)?; - for source in &mut self.sources { - source_index = source.prepare(source_index)?; - for target in &mut source.targets { - // prepare templates - let prepare_result = match &self.templates { - Some(templ) => target.prepare(target_index, Some(templ)), - _ => target.prepare(target_index, None) - }; - prepare_result?; - target_index += 1; - } + Ok(()) + } + + fn prepare_web_auth(&mut self, resolve_var: bool) -> Result<(), M3uFilterError> { + if !self.web_ui_enabled { + self.web_auth = None; } + if let Some(web_auth) = &mut self.web_auth { + if web_auth.enabled { + web_auth.prepare(&self.t_config_path, resolve_var)?; + } else { + self.web_auth = None; + } + } + Ok(()) + } + + fn prepare_video_config(&mut self) -> Result<(), M3uFilterError> { match &mut self.video { None => { self.video = Some(VideoConfig { @@ -1509,22 +1542,29 @@ impl Config { } } }; - - if !self.web_ui_enabled { - self.web_auth = None; - } - - if let Some(web_auth) = &mut self.web_auth { - if web_auth.enabled { - web_auth.prepare(&self.t_config_path, resolve_var)?; - } else { - self.web_auth = None; - } - } - Ok(()) } + fn prepare_custom_stream_response(&mut self, resolve_var: bool) { + if let Some(custom_stream_response) = self.custom_stream_response.as_ref() { + fn load_and_set_file(path: Option<&String>, working_dir: &str, resolve_var: bool) -> Option>> { + path.as_ref() + .map(|file| file_utils::make_absolute_path(file, working_dir, resolve_var)) + .and_then(|absolute_path| match file_utils::read_file_as_bytes(&PathBuf::from(&absolute_path)) { + Ok(data) => Some(Arc::new(data)), + Err(err) => { + error!("Failed to load file: {absolute_path} {err}"); + None + } + }) + } + + self.t_channel_unavailable_video = load_and_set_file(custom_stream_response.channel_unavailable.as_ref(), &self.working_dir, resolve_var); + self.t_user_connections_exhausted_video = load_and_set_file(custom_stream_response.user_connection_exhausted.as_ref(), &self.working_dir, resolve_var); + self.t_provider_connections_exhausted_video = load_and_set_file(custom_stream_response.provider_connection_exhausted.as_ref(), &self.working_dir, resolve_var); + } + } + fn prepare_api_web_root(&mut self, resolve_var: bool) { if !self.api.web_root.is_empty() { self.api.web_root = file_utils::make_absolute_path(&self.api.web_root, &self.working_dir, resolve_var); diff --git a/src/processing/parser/xmltv.rs b/src/processing/parser/xmltv.rs index 8eb76a568..35d9433b8 100644 --- a/src/processing/parser/xmltv.rs +++ b/src/processing/parser/xmltv.rs @@ -170,8 +170,6 @@ mod tests { use std::collections::HashSet; use std::io; use std::path::PathBuf; - use std::rc::Rc; - use crate::model::xmltv::{TVGuide}; #[test] @@ -181,7 +179,7 @@ mod tests { let tv_guide = TVGuide { file: PathBuf::from(file_path) }; let channel_ids = vec!["channel.1", "channel.2", "channel.3"]; - let channel_ids : HashSet> = channel_ids.into_iter().map(|s| Arc::new(s.to_string())).collect(); + let channel_ids : HashSet = channel_ids.into_iter().map(|s| s.to_string()).collect(); match tv_guide.filter(&channel_ids) { None => assert!(false, "No epg filtered"), diff --git a/src/repository/playlist_repository.rs b/src/repository/playlist_repository.rs index ef6031d6f..900e7d23b 100644 --- a/src/repository/playlist_repository.rs +++ b/src/repository/playlist_repository.rs @@ -11,8 +11,7 @@ use crate::repository::storage::{ensure_target_storage_path, get_target_id_mappi use crate::repository::target_id_mapping::TargetIdMapping; use crate::repository::xtream_repository::xtream_write_playlist; use crate::utils::file::file_lock_manager::FileWriteGuard; - -pub const HLS_EXT: &str = ".m3u8"; +use crate::utils::network::request::is_hls_url; pub async fn persist_playlist(playlist: &mut [PlaylistGroup], epg: Option<&Epg>, target: &ConfigTarget, cfg: &Config) -> Result<(), Vec> { @@ -30,7 +29,7 @@ pub async fn persist_playlist(playlist: &mut [PlaylistGroup], epg: Option<&Epg>, let header = &mut channel.header; let provider_id = header.get_provider_id().unwrap_or_default(); if provider_id == 0 { - header.item_type = match (header.url.ends_with(HLS_EXT), header.item_type) { + header.item_type = match (is_hls_url(&header.url), header.item_type) { (true, _) => PlaylistItemType::LiveHls, (false, PlaylistItemType::Live) => PlaylistItemType::LiveUnknown, _ => header.item_type, diff --git a/src/utils/network/request.rs b/src/utils/network/request.rs index ec0803d20..7221dd5ef 100644 --- a/src/utils/network/request.rs +++ b/src/utils/network/request.rs @@ -26,6 +26,8 @@ use crate::utils::file::file_utils::{get_file_path, persist_file}; use crate::m3u_filter_error::create_m3u_filter_error_result; use crate::utils::debug_if_enabled; +pub const HLS_EXT: &str = ".m3u8"; + pub const fn bytes_to_megabytes(bytes: u64) -> u64 { bytes / 1_048_576 } @@ -426,22 +428,48 @@ pub fn classify_content_type(headers: &[(String, String)]) -> MimeCategory { }) } -pub fn replace_extension(path: &str, new_ext: &str) -> String { - let ext = if let Some(stripped) = new_ext.strip_prefix('.') { stripped } else { new_ext }; - if let Some(pos) = path.rfind('/') { - if let Some(dot_pos) = path[pos..].rfind('.') { - let dot_index = pos + dot_pos; - return format!("{}{}.{}", &path[..dot_index], "", ext); +const HLS_EXT_QUERY: &str = ".m3u8?"; +const HLS_EXT_FRAGMENT: &str = ".m3u8#"; + +pub fn is_hls_url(url: &str) -> bool { + let lc_url = url.to_lowercase(); + lc_url.ends_with(HLS_EXT) || lc_url.contains(HLS_EXT_QUERY) || lc_url.contains(HLS_EXT_FRAGMENT) +} + +pub fn replace_url_extension(url: &str, new_ext: &str) -> String { + let ext = new_ext.strip_prefix('.').unwrap_or(new_ext); // Remove leading dot if exists + + // Split URL into the base part (domain and path) and the suffix (query/fragment) + let (base_url, suffix) = match url.find(['?', '#'].as_ref()) { + Some(pos) => (&url[..pos], &url[pos..]), // Base URL and suffix + None => (url, ""), // No query or fragment + }; + + // Find the last '/' in the base URL, which marks the end of the domain and the beginning of the file path + if let Some(last_slash_pos) = base_url.rfind('/') { + if last_slash_pos < 9 { // protocol slash, return url as is + return url.to_string(); + } + let (path_part, file_name_with_extension) = base_url.split_at(last_slash_pos + 1); + // Find the last dot in the file name to replace the extension + if let Some(dot_pos) = file_name_with_extension.rfind('.') { + return format!( + "{}{}.{}{}", + path_part, + &file_name_with_extension[..dot_pos], // Keep the name part before the dot + ext, // Add the new extension + suffix // Add the query or fragment if any + ); } - } else if let Some(dot_pos) = path.rfind('.') { - return format!("{}{}.{}", &path[..dot_pos], "", ext); } - format!("{path}.{ext}") + + // If no extension is found, add the new extension to the base URL + format!("{}{}.{}{}", base_url, "", ext, suffix) } #[cfg(test)] mod tests { - use crate::utils::network::request::{replace_extension, sanitize_sensitive_info}; + use crate::utils::network::request::{replace_url_extension, sanitize_sensitive_info}; #[test] fn test_url_mask() { @@ -454,17 +482,15 @@ mod tests { #[test] fn test_replace_ext() { let tests = [ - "test.txt", - "folder/test.txt", - "folder/subfolder/file", - "/absolute/path/to/file.tar.gz", - "/home/user/script", - "no_extension", - "some/path/file.with.dots", + ("http://hello.world.com", "http://hello.world.com"), + ("http://hello.world.com/123", "http://hello.world.com/123.mp4"), + ("http://hello.world.com/123.ts?hello=world", "http://hello.world.com/123.mp4?hello=world"), + ("http://hello.world.com/123?hello=world", "http://hello.world.com/123.mp4?hello=world"), + ("http://hello.world.com/123#hello=world", "http://hello.world.com/123.mp4#hello=world") ]; - for test in &tests { - println!("{} -> {}", test, replace_extension(test, ".mp4")); + for (test, expect) in &tests { + assert_eq!(replace_url_extension(test, ".mp4"), *expect); } } }