mirror of
https://github.com/euzu/tuliprox.git
synced 2026-10-13 19:32:36 +02:00
Merge pull request #436 from euzu/feature/async_persist_cache_pipe_and_m3u_write_and_shared_buffer_zero_copy
Feature/async persist cache pipe and m3u write and shared buffer zero copy
This commit is contained in:
euzu
@@ -3,6 +3,10 @@
|
||||
- Avoid blocking the runtime when warming cache
|
||||
- Normalize FileLockManager paths so aliases share the same lock
|
||||
- Use async file operations for playlist persistence to avoid blocking async paths
|
||||
- Async persist cache write pipe so response caching no longer blocks the async runtime
|
||||
- M3U playlist exports now stream async to keep the runtime responsive
|
||||
- Shared stream burst buffer zero copy data buffer to reduce memory usage.
|
||||
|
||||
|
||||
# 3.2.0 (2025-11-14)
|
||||
- Added `name` attribute to Staged Input.
|
||||
|
||||
Generated
+41
-82
@@ -79,22 +79,22 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "anstyle-query"
|
||||
version = "1.1.4"
|
||||
version = "1.1.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9e231f6134f61b71076a3eab506c379d4f36122f2af15a9ff04415ea4c3339e2"
|
||||
checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc"
|
||||
dependencies = [
|
||||
"windows-sys 0.60.2",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "anstyle-wincon"
|
||||
version = "3.0.10"
|
||||
version = "3.0.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3e0633414522a32ffaac8ac6cc8f748e090c5717661fddeea04219e2344f5f2a"
|
||||
checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d"
|
||||
dependencies = [
|
||||
"anstyle",
|
||||
"once_cell_polyfill",
|
||||
"windows-sys 0.60.2",
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -165,9 +165,9 @@ checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
|
||||
|
||||
[[package]]
|
||||
name = "axum"
|
||||
version = "0.8.6"
|
||||
version = "0.8.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8a18ed336352031311f4e0b4dd2ff392d4fbb370777c9d18d7fc9d7359f73871"
|
||||
checksum = "5b098575ebe77cb6d14fc7f32749631a6e44edbef6b796f89b020e99ba20d425"
|
||||
dependencies = [
|
||||
"axum-core",
|
||||
"axum-macros",
|
||||
@@ -357,15 +357,15 @@ checksum = "46c5e41b57b8bba42a04676d81cb89e9ee8e859a1a66f80a5a72e1cb76b34d43"
|
||||
|
||||
[[package]]
|
||||
name = "bytes"
|
||||
version = "1.10.1"
|
||||
version = "1.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d71b6127be86fdcfddb610f7182ac57211d4b18a3e9c82eb2d17662f2227ad6a"
|
||||
checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3"
|
||||
|
||||
[[package]]
|
||||
name = "cc"
|
||||
version = "1.2.45"
|
||||
version = "1.2.46"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "35900b6c8d709fb1d854671ae27aeaa9eec2f8b01b364e1619a40da3e6fe2afe"
|
||||
checksum = "b97463e1064cb1b1c1384ad0a0b9c8abd0988e2a91f52606c80ef14aadb63e36"
|
||||
dependencies = [
|
||||
"find-msvc-tools",
|
||||
"jobserver",
|
||||
@@ -395,7 +395,7 @@ dependencies = [
|
||||
"js-sys",
|
||||
"num-traits",
|
||||
"wasm-bindgen",
|
||||
"windows-link 0.2.1",
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1034,9 +1034,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "find-msvc-tools"
|
||||
version = "0.1.4"
|
||||
version = "0.1.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "52051878f80a721bb68ebfbc930e07b65ba72f2da88968ea5c06fd6ca3d3a127"
|
||||
checksum = "3a3076410a55c90011c298b04d0cfa770b00fa04e1e3c97d3f6c9de105a03844"
|
||||
|
||||
[[package]]
|
||||
name = "flate2"
|
||||
@@ -1054,12 +1054,6 @@ version = "1.0.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
|
||||
|
||||
[[package]]
|
||||
name = "foldhash"
|
||||
version = "0.1.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
|
||||
|
||||
[[package]]
|
||||
name = "foldhash"
|
||||
version = "0.2.0"
|
||||
@@ -1102,7 +1096,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "frontend"
|
||||
version = "3.2.2"
|
||||
version = "3.2.3"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"base64",
|
||||
@@ -1740,9 +1734,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "governor"
|
||||
version = "0.10.1"
|
||||
version = "0.10.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "444405bbb1a762387aa22dd569429533b54a1d8759d35d3b64cb39b0293eaa19"
|
||||
checksum = "6e23d5986fd4364c2fb7498523540618b4b8d92eec6c36a02e565f66748e2f79"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"dashmap",
|
||||
@@ -1750,7 +1744,7 @@ dependencies = [
|
||||
"futures-timer",
|
||||
"futures-util",
|
||||
"getrandom 0.3.4",
|
||||
"hashbrown 0.15.5",
|
||||
"hashbrown 0.16.0",
|
||||
"nonzero_ext",
|
||||
"parking_lot",
|
||||
"portable-atomic",
|
||||
@@ -1808,17 +1802,6 @@ version = "0.14.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
|
||||
|
||||
[[package]]
|
||||
name = "hashbrown"
|
||||
version = "0.15.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
|
||||
dependencies = [
|
||||
"allocator-api2",
|
||||
"equivalent",
|
||||
"foldhash 0.1.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hashbrown"
|
||||
version = "0.16.0"
|
||||
@@ -1827,7 +1810,7 @@ checksum = "5419bdc4f6a9207fbeba6d11b604d481addf78ecd10c11ad51e76c2f6482748d"
|
||||
dependencies = [
|
||||
"allocator-api2",
|
||||
"equivalent",
|
||||
"foldhash 0.2.0",
|
||||
"foldhash",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -1925,9 +1908,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
|
||||
|
||||
[[package]]
|
||||
name = "hyper"
|
||||
version = "1.8.0"
|
||||
version = "1.8.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1744436df46f0bde35af3eda22aeaba453aada65d8f1c171cd8a5f59030bd69f"
|
||||
checksum = "2ab2d4f250c3d7b1c9fcdff1cece94ea4e2dfbec68614f7b87cb205f24ca9d11"
|
||||
dependencies = [
|
||||
"atomic-waker",
|
||||
"bytes",
|
||||
@@ -1994,9 +1977,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "hyper-util"
|
||||
version = "0.1.17"
|
||||
version = "0.1.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3c6995591a8f1380fcb4ba966a252a4b29188d51d2b89e3a252f5305be65aea8"
|
||||
checksum = "52e9a2a24dc5c6821e71a7030e1e14b7b632acac55c40e9d2e082c621261bb56"
|
||||
dependencies = [
|
||||
"base64",
|
||||
"bytes",
|
||||
@@ -2732,7 +2715,7 @@ dependencies = [
|
||||
"libc",
|
||||
"redox_syscall",
|
||||
"smallvec",
|
||||
"windows-link 0.2.1",
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -3438,9 +3421,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "rsa"
|
||||
version = "0.9.8"
|
||||
version = "0.9.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "78928ac1ed176a5ca1d17e578a1825f3d81ca54cf41053a592584b020cfd691b"
|
||||
checksum = "40a0376c50d0358279d9d643e4bf7b7be212f1f4ff1da9070a7b54d22ef75c88"
|
||||
dependencies = [
|
||||
"const-oid",
|
||||
"digest",
|
||||
@@ -3782,7 +3765,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "shared"
|
||||
version = "3.2.2"
|
||||
version = "3.2.3"
|
||||
dependencies = [
|
||||
"base64",
|
||||
"bitflags 2.10.0",
|
||||
@@ -4331,7 +4314,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
|
||||
|
||||
[[package]]
|
||||
name = "tuliprox"
|
||||
version = "3.2.2"
|
||||
version = "3.2.3"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"async-compression",
|
||||
@@ -4758,9 +4741,9 @@ checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb"
|
||||
dependencies = [
|
||||
"windows-implement",
|
||||
"windows-interface",
|
||||
"windows-link 0.2.1",
|
||||
"windows-result 0.4.1",
|
||||
"windows-strings 0.5.1",
|
||||
"windows-link",
|
||||
"windows-result",
|
||||
"windows-strings",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4785,12 +4768,6 @@ dependencies = [
|
||||
"syn 2.0.110",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-link"
|
||||
version = "0.1.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a"
|
||||
|
||||
[[package]]
|
||||
name = "windows-link"
|
||||
version = "0.2.1"
|
||||
@@ -4799,22 +4776,13 @@ checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
|
||||
|
||||
[[package]]
|
||||
name = "windows-registry"
|
||||
version = "0.5.3"
|
||||
version = "0.6.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5b8a9ed28765efc97bbc954883f4e6796c33a06546ebafacbabee9696967499e"
|
||||
checksum = "02752bf7fbdcce7f2a27a742f798510f3e5ad88dbe84871e5168e2120c3d5720"
|
||||
dependencies = [
|
||||
"windows-link 0.1.3",
|
||||
"windows-result 0.3.4",
|
||||
"windows-strings 0.4.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-result"
|
||||
version = "0.3.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6"
|
||||
dependencies = [
|
||||
"windows-link 0.1.3",
|
||||
"windows-link",
|
||||
"windows-result",
|
||||
"windows-strings",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4823,16 +4791,7 @@ version = "0.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5"
|
||||
dependencies = [
|
||||
"windows-link 0.2.1",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "windows-strings"
|
||||
version = "0.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57"
|
||||
dependencies = [
|
||||
"windows-link 0.1.3",
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4841,7 +4800,7 @@ version = "0.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091"
|
||||
dependencies = [
|
||||
"windows-link 0.2.1",
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4877,7 +4836,7 @@ version = "0.61.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
|
||||
dependencies = [
|
||||
"windows-link 0.2.1",
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4902,7 +4861,7 @@ version = "0.53.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3"
|
||||
dependencies = [
|
||||
"windows-link 0.2.1",
|
||||
"windows-link",
|
||||
"windows_aarch64_gnullvm 0.53.1",
|
||||
"windows_aarch64_msvc 0.53.1",
|
||||
"windows_i686_gnu 0.53.1",
|
||||
|
||||
+2
-2
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "tuliprox"
|
||||
version = "3.2.2"
|
||||
version = "3.2.3"
|
||||
edition = "2021"
|
||||
|
||||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
@@ -40,7 +40,7 @@ fastrand = "2"
|
||||
rpassword = "7.4"
|
||||
flate2 = "1.1"
|
||||
blake3 = "1.8"
|
||||
bytes = "1.10"
|
||||
bytes = "1.11"
|
||||
tokio-stream = { version = "0.1", features = ["sync"] }
|
||||
tokio = { version = "1.48", features = ["rt-multi-thread", "parking_lot", "fs"] }
|
||||
#tokio = { version = "1.46", features = ["rt-multi-thread", "parking_lot", "fs", "tracing"] }
|
||||
|
||||
@@ -32,7 +32,7 @@ use shared::utils::{
|
||||
};
|
||||
use std::borrow::Cow;
|
||||
use std::collections::HashMap;
|
||||
use std::io::BufWriter;
|
||||
use tokio::io::BufWriter;
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
@@ -1076,7 +1076,7 @@ async fn build_stream_response(
|
||||
None
|
||||
};
|
||||
if let Some(resource_path) = cache_resource_path {
|
||||
if let Ok(file) = create_new_file_for_write(&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);
|
||||
|
||||
@@ -1,45 +1,34 @@
|
||||
use std::io::Write;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::task::{Context, Poll};
|
||||
use crate::api::model::StreamError;
|
||||
use bytes::Bytes;
|
||||
use log::error;
|
||||
use std::collections::VecDeque;
|
||||
use std::pin::Pin;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::task::{Context, Poll};
|
||||
use tokio::io::AsyncWrite;
|
||||
use tokio_stream::Stream;
|
||||
use crate::api::model::StreamError;
|
||||
|
||||
/// `PersistPipeStream`
|
||||
///
|
||||
/// A stream wrapper that pipes data from an input stream to a writer while tracking
|
||||
/// the total number of bytes processed. Upon completion, it triggers a user-provided
|
||||
/// callback with the total size of the data written.
|
||||
/// - `callback`: A user-provided function that is called with the total size of the processed data once the stream is completed.
|
||||
///
|
||||
/// # Stream Implementation
|
||||
/// - Implements the `Stream` trait to poll the underlying stream for data.
|
||||
/// - For each chunk of data:
|
||||
/// - Writes it to the writer.
|
||||
/// - Updates the size tracker.
|
||||
/// - When the stream is exhausted:
|
||||
/// - Calls `on_complete()` to finalize the operation and trigger the callback.
|
||||
/// Pipes bytes from an upstream stream to an async writer while tracking total size.
|
||||
/// Once the stream completes and the writer is flushed, the provided callback is invoked
|
||||
/// with the total number of bytes written.
|
||||
pub struct PersistPipeStream<S, W> {
|
||||
inner: S,
|
||||
completed: bool,
|
||||
writer: W,
|
||||
size: AtomicUsize,
|
||||
callback: Arc<dyn Fn(usize) + Send + Sync>,
|
||||
pending_writes: VecDeque<Bytes>,
|
||||
current_offset: usize,
|
||||
}
|
||||
|
||||
impl<S, W> PersistPipeStream<S, W>
|
||||
where
|
||||
S: Stream + Unpin,
|
||||
W: Write + Unpin + 'static,
|
||||
W: AsyncWrite + Unpin + 'static,
|
||||
{
|
||||
/// - Creates a new `PersistPipeStream` instance.
|
||||
/// - Arguments:
|
||||
/// - `inner`: The input stream providing the data.
|
||||
/// - `writer`: The writer to which the data is written.
|
||||
/// - `callback`: A callback function to be called with the total size upon stream completion.
|
||||
pub fn new(inner: S, writer: W, callback: Arc<dyn Fn(usize) + Send + Sync>) -> Self {
|
||||
Self {
|
||||
inner,
|
||||
@@ -47,47 +36,101 @@ where
|
||||
writer,
|
||||
size: AtomicUsize::new(0),
|
||||
callback,
|
||||
pending_writes: VecDeque::new(),
|
||||
current_offset: 0,
|
||||
}
|
||||
}
|
||||
|
||||
fn on_complete(&mut self) {
|
||||
fn enqueue_chunk(&mut self, bytes: Bytes) {
|
||||
self.pending_writes.push_back(bytes);
|
||||
}
|
||||
|
||||
fn poll_pending_writes(&mut self, cx: &mut Context<'_>) -> Poll<()> {
|
||||
while let Some(chunk) = self.pending_writes.front() {
|
||||
if self.current_offset >= chunk.len() {
|
||||
if let Some(finished) = self.pending_writes.pop_front() {
|
||||
self.size.fetch_add(finished.len(), Ordering::SeqCst);
|
||||
}
|
||||
self.current_offset = 0;
|
||||
continue;
|
||||
}
|
||||
|
||||
let remaining = &chunk[self.current_offset..];
|
||||
match Pin::new(&mut self.writer).poll_write(cx, remaining) {
|
||||
Poll::Pending => return Poll::Pending,
|
||||
Poll::Ready(Ok(written)) => {
|
||||
if written == 0 {
|
||||
// Avoid tight loop if writer makes no progress.
|
||||
return Poll::Pending;
|
||||
}
|
||||
self.current_offset += written;
|
||||
}
|
||||
Poll::Ready(Err(err)) => {
|
||||
error!("Error writing to resource file: {err}");
|
||||
self.pending_writes.pop_front();
|
||||
self.current_offset = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
self.current_offset = 0;
|
||||
Poll::Ready(())
|
||||
}
|
||||
|
||||
fn poll_flush(&mut self, cx: &mut Context<'_>) -> Poll<()> {
|
||||
match Pin::new(&mut self.writer).poll_flush(cx) {
|
||||
Poll::Pending => Poll::Pending,
|
||||
Poll::Ready(Ok(())) => Poll::Ready(()),
|
||||
Poll::Ready(Err(err)) => {
|
||||
error!("Error flushing resource file: {err}");
|
||||
Poll::Ready(())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn finalize(&mut self) {
|
||||
if !self.completed {
|
||||
self.completed = true;
|
||||
let size = self.size.load(Ordering::SeqCst);
|
||||
if self.writer.flush().is_ok() {
|
||||
(self.callback)(size);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn on_data(&mut self, data: &Result<Bytes, StreamError>) {
|
||||
if let Ok(bytes) = data {
|
||||
self.size.fetch_add(bytes.len(), Ordering::SeqCst);
|
||||
let bytes_to_write = bytes.clone();
|
||||
if let Err(e) = self.writer.write_all(&bytes_to_write) {
|
||||
error!("Error writing to resource file: {e}");
|
||||
}
|
||||
(self.callback)(size);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<S, W> Stream for PersistPipeStream<S, W>
|
||||
where
|
||||
S: Stream<Item = Result<bytes::Bytes, StreamError>> + Unpin,
|
||||
W: Write + Unpin + 'static,
|
||||
S: Stream<Item=Result<bytes::Bytes, StreamError>> + Unpin,
|
||||
W: AsyncWrite + Unpin + 'static,
|
||||
{
|
||||
type Item = Result<Bytes, StreamError>;
|
||||
|
||||
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
|
||||
let this = self.get_mut();
|
||||
|
||||
if !this.pending_writes.is_empty() && this.poll_pending_writes(cx).is_pending() {
|
||||
return Poll::Pending;
|
||||
}
|
||||
|
||||
match Pin::new(&mut this.inner).poll_next(cx) {
|
||||
Poll::Pending => Poll::Pending,
|
||||
Poll::Ready(None) => {
|
||||
this.on_complete();
|
||||
if this.poll_pending_writes(cx).is_pending() {
|
||||
return Poll::Pending;
|
||||
}
|
||||
if this.poll_flush(cx).is_pending() {
|
||||
return Poll::Pending;
|
||||
}
|
||||
this.finalize();
|
||||
Poll::Ready(None)
|
||||
}
|
||||
Poll::Ready(Some(item)) => {
|
||||
this.on_data(&item);
|
||||
if let Ok(bytes) = &item {
|
||||
this.enqueue_chunk(bytes.clone());
|
||||
}
|
||||
// Try to drain pending bytes after queuing new data.
|
||||
if this.poll_pending_writes(cx).is_pending() {
|
||||
// fall through: we still return the chunk to the caller even if persistence is pending
|
||||
}
|
||||
Poll::Ready(Some(item))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -67,7 +67,7 @@ type SubscriberId = SocketAddr;
|
||||
|
||||
|
||||
struct BurstBuffer {
|
||||
buffer: VecDeque<Bytes>,
|
||||
buffer: VecDeque<Arc<Bytes>>,
|
||||
buffer_size: usize,
|
||||
current_bytes: usize,
|
||||
}
|
||||
@@ -91,11 +91,11 @@ impl BurstBuffer {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn snapshot(&self) -> VecDeque<Bytes> {
|
||||
self.buffer.iter().cloned().collect::<VecDeque<Bytes>>()
|
||||
pub fn snapshot(&self) -> VecDeque<Arc<Bytes>> {
|
||||
self.buffer.iter().cloned().collect::<VecDeque<Arc<Bytes>>>()
|
||||
}
|
||||
|
||||
pub fn push(&mut self, packet: &Bytes) {
|
||||
pub fn push(&mut self, packet: Arc<Bytes>) {
|
||||
while self.current_bytes > self.buffer_size {
|
||||
if let Some(popped) = self.buffer.pop_front() {
|
||||
self.current_bytes -= popped.len();
|
||||
@@ -105,18 +105,18 @@ impl BurstBuffer {
|
||||
}
|
||||
}
|
||||
self.current_bytes += packet.len();
|
||||
self.buffer.push_back(packet.clone());
|
||||
self.buffer.push_back(packet);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
async fn send_burst_buffer(
|
||||
start_buffer: &VecDeque<Bytes>,
|
||||
start_buffer: &VecDeque<Arc<Bytes>>,
|
||||
client_tx: &Sender<Bytes>,
|
||||
cancellation_token: &CancellationToken) {
|
||||
for buf in start_buffer {
|
||||
if cancellation_token.is_cancelled() { return; }
|
||||
if let Err(err) = client_tx.send(buf.clone()).await {
|
||||
if let Err(err) = client_tx.send(buf.as_ref().clone()).await {
|
||||
debug!("Error sending current chunk: {err}");
|
||||
return; // stop on send error
|
||||
}
|
||||
@@ -244,12 +244,13 @@ impl SharedStreamState {
|
||||
item = source_stream.next() => {
|
||||
match item {
|
||||
Some(Ok(data)) => {
|
||||
let arc_data = Arc::new(data);
|
||||
{
|
||||
let mut buffer = burst_buffer.lock().await;
|
||||
buffer.push(&data);
|
||||
buffer.push(arc_data.clone());
|
||||
}
|
||||
|
||||
match sender.send(data) {
|
||||
match sender.send(arc_data.as_ref().clone()) {
|
||||
Ok(clients) => {
|
||||
if clients == 0 {
|
||||
debug_if_enabled!("No shared stream subscribers closing {}", sanitize_sensitive_info(&streaming_url));
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
use crate::api::model::{AppState};
|
||||
use crate::api::model::AppState;
|
||||
use crate::model::{AppConfig, ProxyUserCredentials};
|
||||
use crate::model::{Config, ConfigTarget, M3uTargetOutput};
|
||||
use crate::repository::indexed_document::{IndexedDocumentDirectAccess, IndexedDocumentIterator, IndexedDocumentWriter};
|
||||
@@ -6,14 +6,17 @@ use crate::repository::m3u_playlist_iterator::M3uPlaylistM3uTextIterator;
|
||||
use crate::repository::storage::get_target_storage_path;
|
||||
use crate::repository::storage_const;
|
||||
use crate::utils;
|
||||
use log::error;
|
||||
use shared::error::{create_tuliprox_error, info_err};
|
||||
use shared::error::{str_to_io_error, TuliproxError, TuliproxErrorKind};
|
||||
use shared::model::PlaylistItemType;
|
||||
use shared::model::{M3uPlaylistItem, PlaylistGroup, PlaylistItem};
|
||||
use std::fs::File;
|
||||
use std::io::{Error, Write};
|
||||
use std::io::Error;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use log::error;
|
||||
use tokio::fs;
|
||||
use tokio::io::{AsyncWriteExt, BufWriter as AsyncBufWriter};
|
||||
use tokio::task;
|
||||
|
||||
macro_rules! cant_write_result {
|
||||
($path:expr, $err:expr) => {
|
||||
@@ -32,51 +35,88 @@ pub fn m3u_get_epg_file_path(target_path: &Path) -> PathBuf {
|
||||
utils::add_prefix_to_filename(&path, "epg_", Some("xml"))
|
||||
}
|
||||
|
||||
fn persist_m3u_playlist_as_text(cfg: &Config, target: &ConfigTarget, target_output: &M3uTargetOutput, m3u_playlist: &Vec<M3uPlaylistItem>) {
|
||||
if let Some(filename) = target_output.filename.as_ref() {
|
||||
if let Some(m3u_filename) = utils::get_file_path(&cfg.working_dir, Some(PathBuf::from(filename))) {
|
||||
match File::create(&m3u_filename) {
|
||||
Ok(file) => {
|
||||
let mut buf_writer = utils::file_writer(&file);
|
||||
let _ = buf_writer.write(b"#EXTM3U\n");
|
||||
for m3u in m3u_playlist {
|
||||
let _ = buf_writer.write(m3u.to_m3u(target.options.as_ref(), false).clone().as_bytes());
|
||||
let _ = buf_writer.write(b"\n");
|
||||
}
|
||||
}
|
||||
Err(_) => {
|
||||
error!("Can't write m3u plain playlist {}", &m3u_filename.display());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
macro_rules! await_playlist_write {
|
||||
($expr:expr, $fmt:literal $(, $args:expr)* ) => {{
|
||||
$expr.await.map_err(|err| {
|
||||
create_tuliprox_error!(TuliproxErrorKind::Notify, $fmt $(, $args)*, err)
|
||||
})?
|
||||
}};
|
||||
}
|
||||
|
||||
pub async fn m3u_write_playlist(cfg: &AppConfig, target: &ConfigTarget, target_output: &M3uTargetOutput, target_path: &Path, new_playlist: &[PlaylistGroup]) -> Result<(), TuliproxError> {
|
||||
if !new_playlist.is_empty() {
|
||||
let (m3u_path, idx_path) = m3u_get_file_paths(target_path);
|
||||
let m3u_playlist = new_playlist.iter()
|
||||
async fn persist_m3u_playlist_as_text(
|
||||
cfg: &Config,
|
||||
target: &ConfigTarget,
|
||||
target_output: &M3uTargetOutput,
|
||||
m3u_playlist: Arc<Vec<M3uPlaylistItem>>,
|
||||
) -> Result<(), TuliproxError> {
|
||||
let Some(filename) = target_output.filename.as_ref() else { return Ok(()); };
|
||||
let Some(m3u_filename) = utils::get_file_path(&cfg.working_dir, Some(PathBuf::from(filename))) else { return Ok(()); };
|
||||
|
||||
let file = await_playlist_write!(fs::File::create(&m3u_filename), "Can't write m3u plain playlist {} - {}", m3u_filename.display());
|
||||
let mut writer = AsyncBufWriter::new(file);
|
||||
await_playlist_write!(writer.write_all(b"#EXTM3U\n"), "Failed to write header to {} - {}", m3u_filename.display());
|
||||
|
||||
for m3u in m3u_playlist.iter() {
|
||||
let line = m3u.to_m3u(target.options.as_ref(), false);
|
||||
await_playlist_write!(writer.write_all(line.as_bytes()), "Failed to write entry to {} - {}", m3u_filename.display());
|
||||
await_playlist_write!(writer.write_all(b"\n"), "Failed to write newline to {} - {}", m3u_filename.display());
|
||||
}
|
||||
|
||||
await_playlist_write!(writer.flush(), "Failed to flush {} - {}", m3u_filename.display());
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn m3u_write_playlist(
|
||||
cfg: &AppConfig,
|
||||
target: &ConfigTarget,
|
||||
target_output: &M3uTargetOutput,
|
||||
target_path: &Path,
|
||||
new_playlist: &[PlaylistGroup],
|
||||
) -> Result<(), TuliproxError> {
|
||||
if new_playlist.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let (m3u_path, idx_path) = m3u_get_file_paths(target_path);
|
||||
let m3u_playlist = Arc::new(
|
||||
new_playlist
|
||||
.iter()
|
||||
.flat_map(|pg| &pg.channels)
|
||||
.filter(|&pli| pli.header.item_type != PlaylistItemType::SeriesInfo)
|
||||
.map(PlaylistItem::to_m3u).collect::<Vec<M3uPlaylistItem>>();
|
||||
.map(PlaylistItem::to_m3u)
|
||||
.collect::<Vec<M3uPlaylistItem>>(),
|
||||
);
|
||||
|
||||
persist_m3u_playlist_as_text(&cfg.config.load(), target, target_output, &m3u_playlist);
|
||||
{
|
||||
let _file_lock = cfg.file_locks.write_lock(&m3u_path).await;
|
||||
match IndexedDocumentWriter::new(m3u_path.clone(), idx_path) {
|
||||
Ok(mut writer) => {
|
||||
for m3u in m3u_playlist {
|
||||
match writer.write_doc(m3u.virtual_id, &m3u) {
|
||||
Ok(()) => {}
|
||||
Err(err) => return Err(cant_write_result!(&m3u_path, err))
|
||||
}
|
||||
}
|
||||
writer.store().map_err(|err| cant_write_result!(&m3u_path, err))?;
|
||||
}
|
||||
Err(err) => return Err(cant_write_result!(&m3u_path, err))
|
||||
}
|
||||
}
|
||||
let file_lock = cfg.file_locks.write_lock(&m3u_path).await;
|
||||
|
||||
if let Err(err) = persist_m3u_playlist_as_text(&cfg.config.load(), target, target_output, Arc::clone(&m3u_playlist)).await {
|
||||
error!("Persisting m3u playlist failed: {err}");
|
||||
}
|
||||
|
||||
let playlist = Arc::clone(&m3u_playlist);
|
||||
let m3u_path_clone = m3u_path.clone();
|
||||
let idx_path_clone = idx_path.clone();
|
||||
|
||||
task::spawn_blocking(move || -> Result<(), TuliproxError> {
|
||||
let _guard = file_lock;
|
||||
match IndexedDocumentWriter::new(m3u_path_clone.clone(), idx_path_clone) {
|
||||
Ok(mut writer) => {
|
||||
for m3u in playlist.iter() {
|
||||
writer
|
||||
.write_doc(m3u.virtual_id, m3u)
|
||||
.map_err(|err| cant_write_result!(&m3u_path_clone, err))?;
|
||||
}
|
||||
writer
|
||||
.store()
|
||||
.map_err(|err| cant_write_result!(&m3u_path_clone, err))
|
||||
}
|
||||
Err(err) => Err(cant_write_result!(&m3u_path_clone, err)),
|
||||
}
|
||||
})
|
||||
.await
|
||||
.map_err(|err| create_tuliprox_error!(TuliproxErrorKind::Notify, "failed to write m3u playlist: {} - {err}", m3u_path.display()))??;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -122,4 +162,4 @@ pub async fn iter_raw_m3u_playlist(config: &AppConfig, target: &ConfigTarget) ->
|
||||
Ok(reader) => Some((file_lock, reader)),
|
||||
Err(_) => None
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -10,6 +10,7 @@ use crate::utils::debug_if_enabled;
|
||||
use shared::utils::{API_PROXY_FILE, CONFIG_FILE, CONFIG_PATH, MAPPING_FILE, SOURCE_FILE, USER_FILE};
|
||||
use log::{debug, error};
|
||||
use path_clean::PathClean;
|
||||
use tokio::fs as tokio_fs;
|
||||
|
||||
pub fn file_writer<W>(w: W) -> BufWriter<W>
|
||||
where
|
||||
@@ -198,8 +199,13 @@ pub fn append_or_crate_file(path: &Path) -> std::io::Result<File> {
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub fn create_new_file_for_write(path: &Path) -> std::io::Result<File> {
|
||||
OpenOptions::new().write(true).create(true).truncate(true).open(path)
|
||||
pub async fn create_new_file_for_write(path: &Path) -> tokio::io::Result<tokio_fs::File> {
|
||||
tokio_fs::OpenOptions::new()
|
||||
.write(true)
|
||||
.create(true)
|
||||
.truncate(true)
|
||||
.open(path)
|
||||
.await
|
||||
}
|
||||
|
||||
#[inline]
|
||||
|
||||
@@ -217,12 +217,12 @@ async fn decode_local_file_bytes(content: Vec<u8>) -> Result<String, Error> {
|
||||
}
|
||||
}
|
||||
|
||||
fn local_file_not_found(path: &PathBuf) -> Error {
|
||||
fn local_file_not_found(path: &Path) -> Error {
|
||||
let file_str = path.to_str().unwrap_or("?");
|
||||
Error::new(ErrorKind::InvalidData, format!("Cant find file {file_str}"))
|
||||
}
|
||||
|
||||
pub async fn get_local_file_content(file_path: &PathBuf) -> Result<String, Error> {
|
||||
pub async fn get_local_file_content(file_path: &Path) -> Result<String, Error> {
|
||||
match tokio::fs::read(file_path).await {
|
||||
Ok(content) => decode_local_file_bytes(content).await,
|
||||
Err(_) => Err(local_file_not_found(file_path)),
|
||||
|
||||
+6
-14
@@ -16,8 +16,8 @@ print_usage() {
|
||||
echo " branch Git branch name (only 'master' and 'develop' are supported)"
|
||||
echo
|
||||
echo "Examples:"
|
||||
echo " $(basename "$0") master # Builds and pushes with :latest tag"
|
||||
echo " $(basename "$0") develop # Builds and pushes with :dev tag"
|
||||
echo " $(basename "$0") master # Builds and pushes with :VERSION and :latest tags"
|
||||
echo " $(basename "$0") develop # Builds and pushes with :VERSION and :dev tags"
|
||||
exit 1
|
||||
}
|
||||
|
||||
@@ -161,18 +161,10 @@ for IMAGE_NAME in "${!MULTI_PLATFORM_IMAGES[@]}"; do
|
||||
echo "🎯 Building multi-platform image: ${IMAGE_NAME} with target ${BUILD_TARGET}"
|
||||
|
||||
# Prepare tags based on branch
|
||||
DOCKER_TAGS=""
|
||||
if [ "$BRANCH" = "master" ]; then
|
||||
# For master branch: create both version and latest tags
|
||||
DOCKER_TAGS="-t ghcr.io/euzu/${IMAGE_NAME}:${VERSION} -t ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}"
|
||||
BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${VERSION}")
|
||||
BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}")
|
||||
elif [ "$BRANCH" = "develop" ]; then
|
||||
# For develop branch: create only dev tag (no version tag)
|
||||
DOCKER_TAGS="-t ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}"
|
||||
BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}")
|
||||
fi
|
||||
|
||||
DOCKER_TAGS="-t ghcr.io/euzu/${IMAGE_NAME}:${VERSION} -t ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}"
|
||||
BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${VERSION}")
|
||||
BUILT_IMAGES+=("ghcr.io/euzu/${IMAGE_NAME}:${TAG_SUFFIX}")
|
||||
|
||||
# Build and push multi-platform image directly with cache
|
||||
BUILDX_CACHE_ARGS=()
|
||||
[ -n "${BUILDX_CACHE_FROM:-}" ] && BUILDX_CACHE_ARGS+=(--cache-from "${BUILDX_CACHE_FROM}")
|
||||
|
||||
+2
-2
@@ -1,10 +1,10 @@
|
||||
[package]
|
||||
name = "frontend"
|
||||
version = "3.2.2"
|
||||
version = "3.2.3"
|
||||
edition = "2021"
|
||||
|
||||
[dependencies]
|
||||
shared = { version = "3.2.2", path = "../shared" }
|
||||
shared = { version = "3.2.3", path = "../shared" }
|
||||
chrono = "0"
|
||||
yew = "0.21"
|
||||
yew-router = "0.18"
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "shared"
|
||||
version = "3.2.2"
|
||||
version = "3.2.3"
|
||||
edition = "2021"
|
||||
|
||||
[dependencies]
|
||||
|
||||
Reference in New Issue
Block a user