diff --git a/CHANGELOG.md b/CHANGELOG.md index b4f3bf658..24c1c0302 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/Cargo.lock b/Cargo.lock index b817c70be..b39068668 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 77ab5377e..2a13255e6 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -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"] } diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index fbdd7f9a0..a086fb0b3 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -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); diff --git a/backend/src/api/model/streams/persist_pipe_stream.rs b/backend/src/api/model/streams/persist_pipe_stream.rs index 5864958e9..550384350 100644 --- a/backend/src/api/model/streams/persist_pipe_stream.rs +++ b/backend/src/api/model/streams/persist_pipe_stream.rs @@ -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 { inner: S, completed: bool, writer: W, size: AtomicUsize, callback: Arc, + pending_writes: VecDeque, + current_offset: usize, } impl PersistPipeStream 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) -> 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) { - 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 Stream for PersistPipeStream where - S: Stream> + Unpin, - W: Write + Unpin + 'static, + S: Stream> + Unpin, + W: AsyncWrite + Unpin + 'static, { type Item = Result; fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { 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)) } } diff --git a/backend/src/api/model/streams/shared_stream_manager.rs b/backend/src/api/model/streams/shared_stream_manager.rs index 752610d6f..6c55132c3 100644 --- a/backend/src/api/model/streams/shared_stream_manager.rs +++ b/backend/src/api/model/streams/shared_stream_manager.rs @@ -67,7 +67,7 @@ type SubscriberId = SocketAddr; struct BurstBuffer { - buffer: VecDeque, + buffer: VecDeque>, buffer_size: usize, current_bytes: usize, } @@ -91,11 +91,11 @@ impl BurstBuffer { } } - pub fn snapshot(&self) -> VecDeque { - self.buffer.iter().cloned().collect::>() + pub fn snapshot(&self) -> VecDeque> { + self.buffer.iter().cloned().collect::>>() } - pub fn push(&mut self, packet: &Bytes) { + pub fn push(&mut self, packet: Arc) { 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, + start_buffer: &VecDeque>, client_tx: &Sender, 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)); diff --git a/backend/src/repository/m3u_repository.rs b/backend/src/repository/m3u_repository.rs index f494f60d6..7d97e0f3c 100644 --- a/backend/src/repository/m3u_repository.rs +++ b/backend/src/repository/m3u_repository.rs @@ -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) { - 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>, +) -> 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::>(); + .map(PlaylistItem::to_m3u) + .collect::>(), + ); - 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 } -} +} \ No newline at end of file diff --git a/backend/src/utils/file/file_utils.rs b/backend/src/utils/file/file_utils.rs index a52092805..d59fc75c0 100644 --- a/backend/src/utils/file/file_utils.rs +++ b/backend/src/utils/file/file_utils.rs @@ -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) -> BufWriter where @@ -198,8 +199,13 @@ pub fn append_or_crate_file(path: &Path) -> std::io::Result { } #[inline] -pub fn create_new_file_for_write(path: &Path) -> std::io::Result { - 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::OpenOptions::new() + .write(true) + .create(true) + .truncate(true) + .open(path) + .await } #[inline] diff --git a/backend/src/utils/network/request.rs b/backend/src/utils/network/request.rs index e9652ff87..46d613bf5 100644 --- a/backend/src/utils/network/request.rs +++ b/backend/src/utils/network/request.rs @@ -217,12 +217,12 @@ async fn decode_local_file_bytes(content: Vec) -> Result { } } -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 { +pub async fn get_local_file_content(file_path: &Path) -> Result { match tokio::fs::read(file_path).await { Ok(content) => decode_local_file_bytes(content).await, Err(_) => Err(local_file_not_found(file_path)), diff --git a/bin/build_docker.sh b/bin/build_docker.sh index 67c8ad936..e35eb0915 100755 --- a/bin/build_docker.sh +++ b/bin/build_docker.sh @@ -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}") diff --git a/frontend/Cargo.toml b/frontend/Cargo.toml index 12a475134..123f5a7a3 100644 --- a/frontend/Cargo.toml +++ b/frontend/Cargo.toml @@ -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" diff --git a/shared/Cargo.toml b/shared/Cargo.toml index 0faab76aa..3e30a1de4 100644 --- a/shared/Cargo.toml +++ b/shared/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "shared" -version = "3.2.2" +version = "3.2.3" edition = "2021" [dependencies]