diff --git a/pmoaudiocache/src/metadata_ext.rs b/pmoaudiocache/src/metadata_ext.rs index 49fc5724..3ddd25a2 100644 --- a/pmoaudiocache/src/metadata_ext.rs +++ b/pmoaudiocache/src/metadata_ext.rs @@ -3,8 +3,8 @@ //! Ce module utilise la macro `define_metadata_properties!` de pmocache //! pour générer automatiquement des méthodes d'accès typées aux métadonnées audio. -use pmocache::define_metadata_properties; use crate::AudioConfig; +use pmocache::define_metadata_properties; // Génération automatique du trait AudioMetadataExt avec toutes les propriétés audio define_metadata_properties! { diff --git a/pmoaudiocache/src/streaming.rs b/pmoaudiocache/src/streaming.rs index bce56175..803f566a 100644 --- a/pmoaudiocache/src/streaming.rs +++ b/pmoaudiocache/src/streaming.rs @@ -150,7 +150,10 @@ async fn stream_flac_to_flac( .await .map_err(|e| format!("Encoder/Decoder error: {}", e))?; - tracing::debug!("Streaming FLAC conversion complete: {} bytes", total_written); + tracing::debug!( + "Streaming FLAC conversion complete: {} bytes", + total_written + ); Ok(()) } @@ -182,9 +185,7 @@ async fn buffer_and_convert_to_flac( // 2. Si c'est déjà du FLAC, on l'écrit directement if buffer.len() >= 4 && &buffer[0..4] == b"fLaC" { tracing::debug!("Input is already FLAC, writing directly"); - file.write_all(&buffer) - .await - .map_err(|e| e.to_string())?; + file.write_all(&buffer).await.map_err(|e| e.to_string())?; file.flush().await.map_err(|e| e.to_string())?; progress(buffer.len() as u64); return Ok(()); @@ -193,11 +194,10 @@ async fn buffer_and_convert_to_flac( tracing::debug!("Converting to FLAC with Symphonia + pmoflac"); // 3. Décoder l'audio avec Symphonia (dans un blocking task car c'est CPU-intensive) - let (samples, channels, sample_rate, bits_per_sample) = tokio::task::spawn_blocking(move || { - decode_with_symphonia_sync(buffer) - }) - .await - .map_err(|e| format!("Decode task panicked: {}", e))??; + let (samples, channels, sample_rate, bits_per_sample) = + tokio::task::spawn_blocking(move || decode_with_symphonia_sync(buffer)) + .await + .map_err(|e| format!("Decode task panicked: {}", e))??; tracing::debug!( "Decoded {} samples, {} channels, {} Hz, {} bits", @@ -341,9 +341,7 @@ fn decode_with_symphonia_sync(buffer: Vec) -> Result<(Vec, usize, u32, decoder.reset(); continue; } - Err(SymphoniaError::IoError(e)) - if e.kind() == std::io::ErrorKind::UnexpectedEof => - { + Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => { break; } Err(e) => { @@ -490,10 +488,7 @@ impl tokio::io::AsyncRead for StreamToAsyncRead { self.chunk_offset = 0; } Poll::Ready(Some(Err(e))) => { - return Poll::Ready(Err(std::io::Error::new( - std::io::ErrorKind::Other, - e, - ))); + return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::Other, e))); } Poll::Ready(None) => { // Stream terminé diff --git a/pmocache/src/cache_trait.rs b/pmocache/src/cache_trait.rs index 48ec1977..536c3a0b 100644 --- a/pmocache/src/cache_trait.rs +++ b/pmocache/src/cache_trait.rs @@ -141,8 +141,7 @@ pub trait FileCache: Send + Sync { /// /// `true` si l'entrée existe en base de données et que le fichier est présent fn is_valid_pk(&self, pk: &str) -> bool { - self.get_database().get(pk, false).is_ok() - && self.file_path(pk).exists() + self.get_database().get(pk, false).is_ok() && self.file_path(pk).exists() } } @@ -176,4 +175,3 @@ pub fn pk_from_content_header(header: &[u8]) -> String { let result = hasher.finalize(); hex::encode(&result[..16]) // 16 octets = 32 caractères hex } - diff --git a/pmoflac/src/autodetect.rs b/pmoflac/src/autodetect.rs index 8bac1bf6..cd27242b 100644 --- a/pmoflac/src/autodetect.rs +++ b/pmoflac/src/autodetect.rs @@ -27,9 +27,9 @@ pub enum DecodeAudioError { #[error("MP3 decode error: {0}")] Mp3(#[from] Mp3Error), #[error("Ogg/Vorbis decode error: {0}")] - Vorbis(#[from] OggError), + Vorbis(OggError), #[error("Ogg/Opus decode error: {0}")] - Opus(#[from] OggOpusError), + Opus(OggOpusError), #[error("WAV decode error: {0}")] Wav(#[from] WavError), #[error("AIFF decode error: {0}")] @@ -67,11 +67,15 @@ where DecodedAudioStream::Mp3(stream) } DetectedFormat::OggVorbis => { - let stream = decode_ogg_vorbis_stream(prefixed).await?; + let stream = decode_ogg_vorbis_stream(prefixed) + .await + .map_err(DecodeAudioError::Vorbis)?; DecodedAudioStream::OggVorbis(stream) } DetectedFormat::OggOpus => { - let stream = decode_ogg_opus_stream(prefixed).await?; + let stream = decode_ogg_opus_stream(prefixed) + .await + .map_err(DecodeAudioError::Opus)?; DecodedAudioStream::OggOpus(stream) } DetectedFormat::Wav => { @@ -113,10 +117,10 @@ impl DecodedAudioStream { DecodedAudioStream::Flac(inner) => inner.wait().await.map_err(DecodeAudioError::from), DecodedAudioStream::Mp3(inner) => inner.wait().await.map_err(DecodeAudioError::from), DecodedAudioStream::OggVorbis(inner) => { - inner.wait().await.map_err(DecodeAudioError::from) + inner.wait().await.map_err(DecodeAudioError::Vorbis) } DecodedAudioStream::OggOpus(inner) => { - inner.wait().await.map_err(DecodeAudioError::from) + inner.wait().await.map_err(DecodeAudioError::Opus) } DecodedAudioStream::Wav(inner) => inner.wait().await.map_err(DecodeAudioError::from), DecodedAudioStream::Aiff(inner) => inner.wait().await.map_err(DecodeAudioError::from), @@ -184,8 +188,8 @@ impl DecodedReader { match self { DecodedReader::Flac(inner) => inner.wait().await.map_err(DecodeAudioError::from), DecodedReader::Mp3(inner) => inner.wait().await.map_err(DecodeAudioError::from), - DecodedReader::OggVorbis(inner) => inner.wait().await.map_err(DecodeAudioError::from), - DecodedReader::OggOpus(inner) => inner.wait().await.map_err(DecodeAudioError::from), + DecodedReader::OggVorbis(inner) => inner.wait().await.map_err(DecodeAudioError::Vorbis), + DecodedReader::OggOpus(inner) => inner.wait().await.map_err(DecodeAudioError::Opus), DecodedReader::Wav(inner) => inner.wait().await.map_err(DecodeAudioError::from), DecodedReader::Aiff(inner) => inner.wait().await.map_err(DecodeAudioError::from), } diff --git a/pmoflac/src/lib.rs b/pmoflac/src/lib.rs index 8332763b..6c9c5963 100644 --- a/pmoflac/src/lib.rs +++ b/pmoflac/src/lib.rs @@ -104,6 +104,7 @@ pub mod encoder; pub mod error; pub mod mp3; pub mod ogg; +mod ogg_common; pub mod opus; mod pcm; mod stream; diff --git a/pmoflac/src/ogg.rs b/pmoflac/src/ogg.rs index 3add0354..854230e5 100644 --- a/pmoflac/src/ogg.rs +++ b/pmoflac/src/ogg.rs @@ -102,8 +102,7 @@ //! ``` use std::{ - collections::VecDeque, - io::{self, Read}, + io, pin::Pin, task::{Context, Poll}, }; @@ -121,56 +120,23 @@ use tokio::{ use crate::{ common::ChannelReader, decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE}, + ogg_common::{OggContainerError, OggPacketReader, OggReaderOptions}, pcm::StreamInfo, stream::ManagedAsyncReader, }; -/// Maximum number of bytes to scan when searching for Ogg sync pattern. -/// -/// This prevents unbounded memory growth when processing streams with -/// large amounts of garbage data before the first valid Ogg page. -const MAX_SYNC_SEARCH: usize = 64 * 1024; +/// Shared error alias for the Vorbis decoder. +pub type OggError = OggContainerError; -/// Errors that can occur while decoding Ogg/Vorbis data. -#[derive(thiserror::Error, Debug, Clone)] -pub enum OggError { - #[error("I/O error ({kind:?}): {message}")] - Io { - kind: io::ErrorKind, - message: String, - }, - #[error("Ogg/Vorbis decode error: {0}")] - Decode(String), - #[error("internal channel closed unexpectedly")] - ChannelClosed, - #[error("{role} task failed: {details}")] - TaskJoin { role: &'static str, details: String }, -} - -impl From for OggError { - fn from(err: io::Error) -> Self { - OggError::Io { - kind: err.kind(), - message: err.to_string(), - } - } -} - -impl From for OggError { +impl From for OggContainerError { fn from(err: header::HeaderReadError) -> Self { - OggError::Decode(err.to_string()) + OggContainerError::Decode(err.to_string()) } } -impl From for OggError { +impl From for OggContainerError { fn from(err: audio::AudioReadError) -> Self { - OggError::Decode(err.to_string()) - } -} - -impl From for OggError { - fn from(value: String) -> Self { - OggError::Decode(value) + OggContainerError::Decode(err.to_string()) } } @@ -262,8 +228,8 @@ where let (info_tx, info_rx) = oneshot::channel::>(); let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), OggError> { - let channel_reader = ChannelReader::::new(ingest_rx); - let mut packet_reader = StreamingPacketReader::new(channel_reader); + let channel_reader = ChannelReader::::new(ingest_rx); + let mut packet_reader = OggPacketReader::new(channel_reader, OggReaderOptions::default()); // Read Vorbis headers (3 packets: identification, comment, setup) let ident_packet = packet_reader @@ -346,314 +312,3 @@ where Ok(OggDecodedStream { info, reader }) } - -/// Streaming packet reader that assembles Vorbis packets from Ogg pages. -/// -/// This reader parses Ogg pages manually without requiring seek operations, -/// making it suitable for truly streaming scenarios. It handles: -/// - Searching for Ogg sync pattern ("OggS") -/// - Parsing page headers and segment tables -/// - Validating CRC32 checksums -/// - Assembling multi-page packets -/// - Detecting end-of-stream -struct StreamingPacketReader -where - E: std::error::Error + std::fmt::Display, -{ - reader: ChannelReader, - current_packet: Vec, - queue: VecDeque>, - finished: bool, - eos_seen: bool, - stream_serial: Option, - sync_buffer: Vec, - synced: bool, -} - -impl StreamingPacketReader -where - E: std::error::Error + std::fmt::Display, -{ - fn new(reader: ChannelReader) -> Self { - Self { - reader, - current_packet: Vec::new(), - queue: VecDeque::new(), - finished: false, - eos_seen: false, - stream_serial: None, - sync_buffer: Vec::new(), - synced: false, - } - } - - /// Returns the next complete Vorbis packet, or None if the stream has ended. - fn next_packet(&mut self) -> Result>, OggError> { - loop { - if let Some(packet) = self.queue.pop_front() { - return Ok(Some(packet)); - } - if self.finished { - return Ok(None); - } - self.read_page()?; - } - } - - /// Reads bytes, first from sync_buffer then from the underlying reader. - fn read_bytes(&mut self, buf: &mut [u8]) -> io::Result { - if buf.is_empty() { - return Ok(0); - } - - let mut total = 0; - - // First, consume from sync_buffer - if !self.sync_buffer.is_empty() { - let to_copy = buf.len().min(self.sync_buffer.len()); - buf[..to_copy].copy_from_slice(&self.sync_buffer[..to_copy]); - self.sync_buffer.drain(..to_copy); - total += to_copy; - if total == buf.len() { - return Ok(total); - } - } - - // Then read from underlying reader - while total < buf.len() { - match Read::read(&mut self.reader, &mut buf[total..])? { - 0 => break, - n => total += n, - } - } - - Ok(total) - } - - /// Reads exactly buf.len() bytes or returns error. - fn read_exact_from_source(&mut self, buf: &mut [u8]) -> Result { - let mut offset = 0; - while offset < buf.len() { - let n = self.read_bytes(&mut buf[offset..])?; - if n == 0 { - return if offset == 0 { - Ok(false) // Clean EOF - } else { - Err(OggError::Decode("unexpected EOF while reading page".into())) - }; - } - offset += n; - } - Ok(true) - } - - /// Searches for the Ogg sync pattern ("OggS") in the stream. - /// - /// This is called before reading the first page to handle streams that - /// have garbage bytes at the beginning (e.g., HTTP headers, ID3 tags). - /// It buffers up to MAX_SYNC_SEARCH bytes while searching. - fn find_sync(&mut self) -> Result<(), OggError> { - if self.synced { - return Ok(()); - } - - while self.sync_buffer.len() < MAX_SYNC_SEARCH { - let mut chunk = [0u8; 1024]; - let n = Read::read(&mut self.reader, &mut chunk)?; - if n == 0 { - return Err(OggError::Decode( - "EOF reached while searching for Ogg sync pattern".into(), - )); - } - self.sync_buffer.extend_from_slice(&chunk[..n]); - - // Search for "OggS" pattern - if let Some(pos) = self - .sync_buffer - .windows(4) - .position(|window| window == b"OggS") - { - // Found sync! Remove garbage bytes before it - self.sync_buffer.drain(..pos); - self.synced = true; - return Ok(()); - } - - // If buffer is getting large and still no sync, keep only last 3 bytes - // (in case "OggS" is split across chunk boundary) - if self.sync_buffer.len() >= MAX_SYNC_SEARCH { - let keep_len = 3.min(self.sync_buffer.len()); - self.sync_buffer.drain(..self.sync_buffer.len() - keep_len); - } - } - - Err(OggError::Decode(format!( - "No Ogg sync pattern found in first {} bytes", - MAX_SYNC_SEARCH - ))) - } - - /// Reads a single Ogg page and processes its packets. - /// - /// This method: - /// 1. Ensures we're synced to "OggS" pattern - /// 2. Reads the 27-byte page header - /// 3. Validates the CRC32 checksum - /// 4. Reads the segment table - /// 5. Reads the page data - /// 6. Assembles packets from segments - fn read_page(&mut self) -> Result<(), OggError> { - // Ensure we've found the sync pattern - self.find_sync()?; - - // Read 27-byte page header - let mut header = [0u8; 27]; - if !self.read_exact_from_source(&mut header)? { - self.finished = true; - return Ok(()); - } - - // Validate Ogg page header - if &header[0..4] != b"OggS" { - return Err(OggError::Decode("invalid Ogg capture pattern".into())); - } - if header[4] != 0 { - return Err(OggError::Decode("unsupported Ogg version".into())); - } - - let header_type = header[5]; - let bitstream_serial = u32::from_le_bytes([header[14], header[15], header[16], header[17]]); - - // Enforce single bitstream - if let Some(serial) = self.stream_serial { - if serial != bitstream_serial { - return Err(OggError::Decode( - "multiple logical streams are not supported".into(), - )); - } - } else { - self.stream_serial = Some(bitstream_serial); - } - - // Read segment table - let page_segments = header[26] as usize; - let mut segment_table = vec![0u8; page_segments]; - self.read_exact_from_source(&mut segment_table)?; - - // Calculate page data length - let data_len: usize = segment_table.iter().map(|&v| v as usize).sum(); - let mut data = vec![0u8; data_len]; - self.read_exact_from_source(&mut data)?; - - // Validate CRC32 - let expected_crc = u32::from_le_bytes([header[22], header[23], header[24], header[25]]); - let mut crc_header = header; - crc_header[22..26].copy_from_slice(&[0, 0, 0, 0]); // Zero out CRC field - - let mut crc = crc::vorbis_crc32_update(0, &crc_header); - crc = crc::vorbis_crc32_update(crc, &segment_table); - crc = crc::vorbis_crc32_update(crc, &data); - - if crc != expected_crc { - return Err(OggError::Decode(format!( - "CRC32 mismatch: expected 0x{:08x}, got 0x{:08x}", - expected_crc, crc - ))); - } - - // Validate continuation flags - if header_type & 0x01 != 0 && self.current_packet.is_empty() { - return Err(OggError::Decode( - "unexpected continuation flag without existing packet".into(), - )); - } - if header_type & 0x01 == 0 && !self.current_packet.is_empty() { - return Err(OggError::Decode( - "dangling packet without continuation flag".into(), - )); - } - - // Assemble packets from segments - let mut offset: usize = 0; - for &seg_len in &segment_table { - let len = seg_len as usize; - let end = offset - .checked_add(len) - .ok_or_else(|| OggError::Decode("segment length overflow".into()))?; - if end > data.len() { - return Err(OggError::Decode("segment exceeds page data".into())); - } - self.current_packet.extend_from_slice(&data[offset..end]); - offset = end; - - // Packet complete when segment is less than 255 bytes - if seg_len < 255 { - let packet = std::mem::take(&mut self.current_packet); - self.queue.push_back(packet); - } - } - - if offset != data.len() { - return Err(OggError::Decode("page data not fully consumed".into())); - } - - // Check for end-of-stream - if header_type & 0x04 != 0 { - self.eos_seen = true; - if self.current_packet.is_empty() { - self.finished = true; - } - } - - Ok(()) - } -} - -/// CRC32 calculation for Ogg pages. -/// -/// This module implements the CRC32 algorithm used by the Ogg container format. -/// The polynomial is 0x04c11db7 with initial value 0 and no final XOR. -mod crc { - /// Precomputed CRC32 lookup table for Ogg. - /// - /// Generated using the polynomial 0x04c11db7. - const fn get_tbl_elem(idx: u32) -> u32 { - let mut r: u32 = idx << 24; - let mut i = 0; - while i < 8 { - r = (r << 1) ^ (-(((r >> 31) & 1) as i32) as u32 & 0x04c11db7); - i += 1; - } - r - } - - const fn lookup_array() -> [u32; 0x100] { - let mut lup_arr: [u32; 0x100] = [0; 0x100]; - let mut i = 0; - while i < 0x100 { - lup_arr[i] = get_tbl_elem(i as u32); - i += 1; - } - lup_arr - } - - static CRC_LOOKUP_ARRAY: &[u32] = &lookup_array(); - - /// Updates the CRC32 value with new data. - /// - /// # Arguments - /// - /// * `cur` - Current CRC32 value (use 0 for initial call) - /// * `array` - Data to include in CRC calculation - /// - /// # Returns - /// - /// Updated CRC32 value - pub fn vorbis_crc32_update(cur: u32, array: &[u8]) -> u32 { - let mut ret: u32 = cur; - for av in array { - ret = (ret << 8) ^ CRC_LOOKUP_ARRAY[(*av as u32 ^ (ret >> 24)) as usize]; - } - ret - } -} diff --git a/pmoflac/src/ogg_common.rs b/pmoflac/src/ogg_common.rs new file mode 100644 index 00000000..bd1463e0 --- /dev/null +++ b/pmoflac/src/ogg_common.rs @@ -0,0 +1,426 @@ +//! # Common Ogg Container Parsing +//! +//! This module provides shared functionality for parsing Ogg containers, +//! used by both Ogg/Vorbis and Ogg/Opus decoders. +//! +//! ## Features +//! +//! - **Streaming packet assembly**: Reads Ogg pages and assembles multi-page packets +//! - **Optional CRC32 validation**: Can validate page integrity +//! - **Optional sync search**: Can search for "OggS" pattern in streams with garbage +//! - **Shared error type**: Uses `OggContainerError` for consistent error reporting +//! +//! ## Architecture +//! +//! The `OggPacketReader` reads Ogg pages incrementally: +//! 1. Optionally searches for "OggS" sync pattern +//! 2. Reads 27-byte page headers +//! 3. Optionally validates CRC32 checksums +//! 4. Reads segment tables and page data +//! 5. Assembles packets from segments (handling multi-page packets) +//! 6. Returns complete packets to the decoder + +use std::{ + collections::VecDeque, + io::{self, Read}, +}; + +use crate::common::ChannelReader; + +/// Maximum number of bytes to scan when searching for Ogg sync pattern. +/// +/// This prevents unbounded memory growth when processing streams with +/// large amounts of garbage data before the first valid Ogg page. +const MAX_SYNC_SEARCH: usize = 64 * 1024; + +/// Configuration options for Ogg packet reader. +#[derive(Clone, Debug)] +pub struct OggReaderOptions { + /// Whether to validate CRC32 checksums of Ogg pages. + /// + /// Vorbis typically validates CRC, Opus often doesn't. + pub validate_crc: bool, + + /// Whether to search for "OggS" sync pattern at start of stream. + /// + /// Useful for streams that may have garbage bytes before valid data. + pub find_sync: bool, + + /// Maximum bytes to search for sync pattern (only used if find_sync is true). + pub max_sync_search: usize, +} + +impl Default for OggReaderOptions { + fn default() -> Self { + Self { + validate_crc: true, + find_sync: true, + max_sync_search: MAX_SYNC_SEARCH, + } + } +} + +/// Shared error type for Ogg container parsing. +#[derive(thiserror::Error, Debug, Clone)] +pub enum OggContainerError { + #[error("I/O error ({kind:?}): {message}")] + Io { + kind: io::ErrorKind, + message: String, + }, + #[error("ogg container error: {0}")] + Decode(String), + #[error("internal channel closed unexpectedly")] + ChannelClosed, + #[error("{role} task failed: {details}")] + TaskJoin { role: &'static str, details: String }, +} + +impl From for OggContainerError { + fn from(err: io::Error) -> Self { + OggContainerError::Io { + kind: err.kind(), + message: err.to_string(), + } + } +} + +impl From for OggContainerError { + fn from(value: String) -> Self { + OggContainerError::Decode(value) + } +} + +impl From<&str> for OggContainerError { + fn from(value: &str) -> Self { + OggContainerError::Decode(value.into()) + } +} + +/// Streaming Ogg packet reader that assembles packets from Ogg pages. +/// +/// This reader parses Ogg pages manually without requiring seek operations, +/// making it suitable for truly streaming scenarios. +/// +/// # Features +/// +/// - Searches for Ogg sync pattern ("OggS") +/// - Parses page headers and segment tables +/// - Validates CRC32 checksums (optional) +/// - Assembles multi-page packets +/// - Detects end-of-stream +/// - Enforces single logical bitstream +pub struct OggPacketReader { + reader: ChannelReader, + current_packet: Vec, + queue: VecDeque>, + finished: bool, + stream_serial: Option, + sync_buffer: Vec, + synced: bool, + options: OggReaderOptions, +} + +impl OggPacketReader { + /// Creates a new Ogg packet reader with the given options. + pub fn new(reader: ChannelReader, options: OggReaderOptions) -> Self { + Self { + reader, + current_packet: Vec::new(), + queue: VecDeque::new(), + finished: false, + stream_serial: None, + sync_buffer: Vec::new(), + synced: !options.find_sync, // If we don't need to find sync, we're already synced + options, + } + } + + /// Returns the next complete packet, or None if the stream has ended. + /// + /// # Errors + /// + /// Returns an error if: + /// - I/O error occurs + /// - Ogg page structure is invalid + /// - CRC32 validation fails (if enabled) + /// - Multiple logical bitstreams detected + pub fn next_packet(&mut self) -> Result>, OggContainerError> { + loop { + if let Some(packet) = self.queue.pop_front() { + return Ok(Some(packet)); + } + if self.finished { + return Ok(None); + } + self.read_page()?; + } + } + + /// Reads bytes, first from sync_buffer then from the underlying reader. + fn read_bytes(&mut self, buf: &mut [u8]) -> io::Result { + if buf.is_empty() { + return Ok(0); + } + + let mut total = 0; + + // First, consume from sync_buffer + if !self.sync_buffer.is_empty() { + let to_copy = buf.len().min(self.sync_buffer.len()); + buf[..to_copy].copy_from_slice(&self.sync_buffer[..to_copy]); + self.sync_buffer.drain(..to_copy); + total += to_copy; + if total == buf.len() { + return Ok(total); + } + } + + // Then read from underlying reader + while total < buf.len() { + match Read::read(&mut self.reader, &mut buf[total..])? { + 0 => break, + n => total += n, + } + } + + Ok(total) + } + + /// Reads exactly buf.len() bytes or returns error. + fn read_exact_from_source(&mut self, buf: &mut [u8]) -> Result { + let mut offset = 0; + while offset < buf.len() { + let n = self + .read_bytes(&mut buf[offset..]) + .map_err(OggContainerError::from)?; + if n == 0 { + return if offset == 0 { + Ok(false) // Clean EOF + } else { + Err(OggContainerError::Decode( + "unexpected EOF while reading page".into(), + )) + }; + } + offset += n; + } + Ok(true) + } + + /// Searches for the Ogg sync pattern ("OggS") in the stream. + /// + /// This is called before reading the first page to handle streams that + /// have garbage bytes at the beginning (e.g., HTTP headers, ID3 tags). + /// It buffers up to max_sync_search bytes while searching. + fn find_sync(&mut self) -> Result<(), OggContainerError> { + if self.synced { + return Ok(()); + } + + while self.sync_buffer.len() < self.options.max_sync_search { + let mut chunk = [0u8; 1024]; + let n = Read::read(&mut self.reader, &mut chunk).map_err(OggContainerError::from)?; + if n == 0 { + return Err(OggContainerError::Decode( + "EOF reached while searching for Ogg sync pattern".into(), + )); + } + self.sync_buffer.extend_from_slice(&chunk[..n]); + + // Search for "OggS" pattern + if let Some(pos) = self + .sync_buffer + .windows(4) + .position(|window| window == b"OggS") + { + // Found sync! Remove garbage bytes before it + self.sync_buffer.drain(..pos); + self.synced = true; + return Ok(()); + } + + // If buffer is getting large and still no sync, keep only last 3 bytes + // (in case "OggS" is split across chunk boundary) + if self.sync_buffer.len() >= self.options.max_sync_search { + let keep_len = 3.min(self.sync_buffer.len()); + self.sync_buffer.drain(..self.sync_buffer.len() - keep_len); + } + } + + Err(OggContainerError::Decode(format!( + "No Ogg sync pattern found in first {} bytes", + self.options.max_sync_search + ))) + } + + /// Reads a single Ogg page and processes its packets. + /// + /// This method: + /// 1. Ensures we're synced to "OggS" pattern (if enabled) + /// 2. Reads the 27-byte page header + /// 3. Validates the CRC32 checksum (if enabled) + /// 4. Reads the segment table + /// 5. Reads the page data + /// 6. Assembles packets from segments + fn read_page(&mut self) -> Result<(), OggContainerError> { + // Ensure we've found the sync pattern (if required) + if self.options.find_sync { + self.find_sync()?; + } + + // Read 27-byte page header + let mut header = [0u8; 27]; + if !self.read_exact_from_source(&mut header)? { + self.finished = true; + return Ok(()); + } + + // Validate Ogg page header + if &header[0..4] != b"OggS" { + return Err(OggContainerError::Decode( + "invalid Ogg capture pattern".into(), + )); + } + if header[4] != 0 { + return Err(OggContainerError::Decode("unsupported Ogg version".into())); + } + + let header_type = header[5]; + let bitstream_serial = u32::from_le_bytes([header[14], header[15], header[16], header[17]]); + + // Enforce single bitstream + if let Some(serial) = self.stream_serial { + if serial != bitstream_serial { + return Err(OggContainerError::Decode( + "multiple logical streams are not supported".into(), + )); + } + } else { + self.stream_serial = Some(bitstream_serial); + } + + // Read segment table + let page_segments = header[26] as usize; + let mut segment_table = vec![0u8; page_segments]; + self.read_exact_from_source(&mut segment_table)?; + + // Calculate page data length + let data_len: usize = segment_table.iter().map(|&v| v as usize).sum(); + let mut data = vec![0u8; data_len]; + self.read_exact_from_source(&mut data)?; + + // Validate CRC32 if enabled + if self.options.validate_crc { + let expected_crc = u32::from_le_bytes([header[22], header[23], header[24], header[25]]); + let mut crc_header = header; + crc_header[22..26].copy_from_slice(&[0, 0, 0, 0]); // Zero out CRC field + + let mut crc = crc::vorbis_crc32_update(0, &crc_header); + crc = crc::vorbis_crc32_update(crc, &segment_table); + crc = crc::vorbis_crc32_update(crc, &data); + + if crc != expected_crc { + return Err(OggContainerError::Decode(format!( + "CRC32 mismatch: expected 0x{expected_crc:08x}, got 0x{crc:08x}" + ))); + } + } + + // Validate continuation flags + if header_type & 0x01 != 0 && self.current_packet.is_empty() { + return Err(OggContainerError::Decode( + "unexpected continuation flag without existing packet".into(), + )); + } + if header_type & 0x01 == 0 && !self.current_packet.is_empty() { + return Err(OggContainerError::Decode( + "dangling packet without continuation flag".into(), + )); + } + + // Assemble packets from segments + let mut offset: usize = 0; + for &seg_len in &segment_table { + let len = seg_len as usize; + let end = offset + .checked_add(len) + .ok_or_else(|| OggContainerError::Decode("segment length overflow".into()))?; + if end > data.len() { + return Err(OggContainerError::Decode( + "segment exceeds page data".into(), + )); + } + self.current_packet.extend_from_slice(&data[offset..end]); + offset = end; + + // Packet complete when segment is less than 255 bytes + if seg_len < 255 { + let packet = std::mem::take(&mut self.current_packet); + self.queue.push_back(packet); + } + } + + if offset != data.len() { + return Err(OggContainerError::Decode( + "page data not fully consumed".into(), + )); + } + + // Check for end-of-stream + if header_type & 0x04 != 0 && self.current_packet.is_empty() { + self.finished = true; + } + + Ok(()) + } +} + +/// CRC32 calculation for Ogg pages. +/// +/// This module implements the CRC32 algorithm used by the Ogg container format. +/// The polynomial is 0x04c11db7 with initial value 0 and no final XOR. +pub(crate) mod crc { + /// Precomputed CRC32 lookup table for Ogg. + /// + /// Generated using the polynomial 0x04c11db7. + const fn get_tbl_elem(idx: u32) -> u32 { + let mut r: u32 = idx << 24; + let mut i = 0; + while i < 8 { + r = (r << 1) ^ (-(((r >> 31) & 1) as i32) as u32 & 0x04c11db7); + i += 1; + } + r + } + + const fn lookup_array() -> [u32; 0x100] { + let mut lup_arr: [u32; 0x100] = [0; 0x100]; + let mut i = 0; + while i < 0x100 { + lup_arr[i] = get_tbl_elem(i as u32); + i += 1; + } + lup_arr + } + + static CRC_LOOKUP_ARRAY: &[u32] = &lookup_array(); + + /// Updates the CRC32 value with new data. + /// + /// # Arguments + /// + /// * `cur` - Current CRC32 value (use 0 for initial call) + /// * `array` - Data to include in CRC calculation + /// + /// # Returns + /// + /// Updated CRC32 value + pub fn vorbis_crc32_update(cur: u32, array: &[u8]) -> u32 { + let mut ret: u32 = cur; + for av in array { + ret = (ret << 8) ^ CRC_LOOKUP_ARRAY[(*av as u32 ^ (ret >> 24)) as usize]; + } + ret + } +} diff --git a/pmoflac/src/opus.rs b/pmoflac/src/opus.rs index fa9aad0f..640c927c 100644 --- a/pmoflac/src/opus.rs +++ b/pmoflac/src/opus.rs @@ -5,7 +5,7 @@ //! 100% streaming (no seeking or buffering entire files). use std::{ - io::{self, Read}, + io, pin::Pin, task::{Context, Poll}, }; @@ -19,6 +19,7 @@ use tokio::{ use crate::{ common::ChannelReader, decoder_common::{spawn_ingest_task, spawn_writer_task, CHANNEL_CAPACITY, DUPLEX_BUFFER_SIZE}, + ogg_common::{OggContainerError, OggPacketReader, OggReaderOptions}, pcm::StreamInfo, stream::ManagedAsyncReader, }; @@ -26,38 +27,12 @@ use crate::{ /// Maximum number of samples per Opus frame at 48 kHz (120 ms). const MAX_FRAME_SAMPLES: usize = 5760; -/// Errors that can occur while decoding Ogg/Opus data. -#[derive(thiserror::Error, Debug, Clone)] -pub enum OggOpusError { - #[error("I/O error ({kind:?}): {message}")] - Io { - kind: io::ErrorKind, - message: String, - }, - #[error("Ogg/Opus decode error: {0}")] - Decode(String), - #[error("internal channel closed unexpectedly")] - ChannelClosed, -} +/// Shared error alias for the Opus decoder. +pub type OggOpusError = OggContainerError; -impl From for OggOpusError { - fn from(err: io::Error) -> Self { - OggOpusError::Io { - kind: err.kind(), - message: err.to_string(), - } - } -} - -impl From for OggOpusError { +impl From for OggContainerError { fn from(err: OpusError) -> Self { - OggOpusError::Decode(err.to_string()) - } -} - -impl From for OggOpusError { - fn from(value: String) -> Self { - OggOpusError::Decode(value) + OggContainerError::Decode(err.to_string()) } } @@ -107,8 +82,15 @@ where let (info_tx, info_rx) = oneshot::channel::>(); let blocking_handle = tokio::task::spawn_blocking(move || -> Result<(), OggOpusError> { - let channel_reader = ChannelReader::::new(ingest_rx); - let mut packet_reader = StreamingPacketReader::new(channel_reader); + let channel_reader = ChannelReader::::new(ingest_rx); + let mut packet_reader = OggPacketReader::new( + channel_reader, + OggReaderOptions { + validate_crc: false, + find_sync: false, + ..OggReaderOptions::default() + }, + ); let header_packet = packet_reader .next_packet()? @@ -275,141 +257,3 @@ impl OpusTags { Ok(OpusTags) } } - -/// Streaming Ogg packet reader reused for Opus packets. -struct StreamingPacketReader -where - E: std::error::Error + std::fmt::Display, -{ - reader: ChannelReader, - current_packet: Vec, - pending_packets: std::collections::VecDeque>, - finished: bool, - stream_serial: Option, -} - -impl StreamingPacketReader -where - E: std::error::Error + std::fmt::Display, -{ - fn new(reader: ChannelReader) -> Self { - Self { - reader, - current_packet: Vec::new(), - pending_packets: std::collections::VecDeque::new(), - finished: false, - stream_serial: None, - } - } - - fn next_packet(&mut self) -> Result>, OggOpusError> { - loop { - if let Some(packet) = self.pending_packets.pop_front() { - return Ok(Some(packet)); - } - if self.finished { - return Ok(None); - } - self.read_page()?; - } - } - - fn read_page(&mut self) -> Result<(), OggOpusError> { - let mut header = [0u8; 27]; - if !read_exact_or_eof(&mut self.reader, &mut header)? { - self.finished = true; - return Ok(()); - } - - if &header[0..4] != b"OggS" { - return Err(OggOpusError::Decode("invalid Ogg capture pattern".into())); - } - if header[4] != 0 { - return Err(OggOpusError::Decode("unsupported Ogg version".into())); - } - - let header_type = header[5]; - let bitstream_serial = u32::from_le_bytes([header[14], header[15], header[16], header[17]]); - - if let Some(serial) = self.stream_serial { - if serial != bitstream_serial { - return Err(OggOpusError::Decode( - "multiple logical Ogg streams are unsupported".to_string(), - )); - } - } else { - self.stream_serial = Some(bitstream_serial); - } - - let page_segments = header[26] as usize; - let mut segment_table = vec![0u8; page_segments]; - read_exact_checked(&mut self.reader, &mut segment_table)?; - - let data_len: usize = segment_table.iter().map(|&v| v as usize).sum(); - let mut data = vec![0u8; data_len]; - read_exact_checked(&mut self.reader, &mut data)?; - - if header_type & 0x01 != 0 && self.current_packet.is_empty() { - return Err(OggOpusError::Decode( - "continuation flag set without existing packet".into(), - )); - } - if header_type & 0x01 == 0 && !self.current_packet.is_empty() { - return Err(OggOpusError::Decode( - "expected continuation flag for unfinished packet".into(), - )); - } - - let mut offset = 0usize; - for &seg_len in &segment_table { - let len = seg_len as usize; - let end = offset - .checked_add(len) - .ok_or_else(|| OggOpusError::Decode("segment length overflow".into()))?; - if end > data.len() { - return Err(OggOpusError::Decode("segment exceeds page data".into())); - } - self.current_packet.extend_from_slice(&data[offset..end]); - offset = end; - - if seg_len < 255 { - let packet = std::mem::take(&mut self.current_packet); - self.pending_packets.push_back(packet); - } - } - - if offset != data.len() { - return Err(OggOpusError::Decode("page data not fully consumed".into())); - } - - if header_type & 0x04 != 0 && self.current_packet.is_empty() { - self.finished = true; - } - - Ok(()) - } -} - -fn read_exact_or_eof(reader: &mut R, buf: &mut [u8]) -> io::Result { - let mut read = 0; - while read < buf.len() { - match reader.read(&mut buf[read..])? { - 0 if read == 0 => return Ok(false), - 0 => { - return Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "unexpected EOF while reading", - )) - } - n => read += n, - } - } - Ok(true) -} - -fn read_exact_checked(reader: &mut R, buf: &mut [u8]) -> Result<(), OggOpusError> { - if !read_exact_or_eof(reader, buf)? { - return Err(OggOpusError::Decode("unexpected EOF in Ogg stream".into())); - } - Ok(()) -} diff --git a/pmomediaserver/src/content_handler.rs b/pmomediaserver/src/content_handler.rs index b17e420b..cf25ab13 100644 --- a/pmomediaserver/src/content_handler.rs +++ b/pmomediaserver/src/content_handler.rs @@ -28,9 +28,12 @@ fn to_didl_lite(containers: &[Container], items: &[pmodidl::Item]) -> Result{}", body)) + let body = quick_xml::se::to_string(&didl) + .map_err(|e| format!("Failed to serialize DIDL-Lite: {}", e))?; + Ok(format!( + "{}", + body + )) } /// Handler pour le service ContentDirectory @@ -99,8 +102,8 @@ impl ContentHandler { if object_id == "0" { // Retourner le container racine let root = self.build_root_container().await; - let didl = to_didl_lite(&[root], &[])?; - Ok((didl, 1, 1, 1)) + let didl = to_didl_lite(&[root], &[])?; + Ok((didl, 1, 1, 1)) } else { // Essayer de trouver l'objet dans les sources // Vérifier si c'est un container racine d'une source diff --git a/pmoparadise/src/ffmpeg_streaming.rs b/pmoparadise/src/ffmpeg_streaming.rs index db6bcca1..389ed8ff 100644 --- a/pmoparadise/src/ffmpeg_streaming.rs +++ b/pmoparadise/src/ffmpeg_streaming.rs @@ -26,7 +26,7 @@ pub fn init() -> Result<()> { /// PCM chunk with decoded audio data #[derive(Debug, Clone)] pub struct PCMChunk { - pub samples: Vec, // Interleaved 16-bit samples + pub samples: Vec, // Interleaved 16-bit samples pub sample_rate: u32, pub channels: u32, pub position_ms: u64, @@ -52,7 +52,7 @@ impl ProgressiveDecoder { let mut buffer = vec![0u8; 8192]; loop { match stream.read(&mut buffer) { - Ok(0) => break, // EOF + Ok(0) => break, // EOF Ok(n) => { let chunk = Bytes::copy_from_slice(&buffer[..n]); if tx.send(Ok(chunk)).is_err() { diff --git a/pmoparadise/src/paradise/worker.rs b/pmoparadise/src/paradise/worker.rs index 5823204e..c502436e 100644 --- a/pmoparadise/src/paradise/worker.rs +++ b/pmoparadise/src/paradise/worker.rs @@ -881,7 +881,7 @@ fn ms_to_frames(ms: u64, sample_rate: u32) -> usize { fn decode_block_audio(data: Vec) -> anyhow::Result { use symphonia::core::audio::SampleBuffer; - use symphonia::core::codecs::{CODEC_TYPE_NULL, DecoderOptions}; + use symphonia::core::codecs::{DecoderOptions, CODEC_TYPE_NULL}; use symphonia::core::errors::Error as SymphoniaError; use symphonia::core::formats::FormatOptions; use symphonia::core::io::MediaSourceStream; diff --git a/pmoplaylist/examples/basic_usage.rs b/pmoplaylist/examples/basic_usage.rs index 1671186d..642a5228 100644 --- a/pmoplaylist/examples/basic_usage.rs +++ b/pmoplaylist/examples/basic_usage.rs @@ -5,7 +5,7 @@ //! cargo run -p pmoplaylist --example basic_usage //! ``` -use pmoplaylist::{DEFAULT_IMAGE, FifoPlaylist, Track}; +use pmoplaylist::{FifoPlaylist, Track, DEFAULT_IMAGE}; #[tokio::main] async fn main() { diff --git a/pmoplaylist/examples/http_server_integration.rs b/pmoplaylist/examples/http_server_integration.rs index d4aab310..a420f6ca 100644 --- a/pmoplaylist/examples/http_server_integration.rs +++ b/pmoplaylist/examples/http_server_integration.rs @@ -8,7 +8,7 @@ //! cargo run -p pmoplaylist --example http_server_integration //! ``` -use pmoplaylist::{DEFAULT_IMAGE, FifoPlaylist, Track}; +use pmoplaylist::{FifoPlaylist, Track, DEFAULT_IMAGE}; use std::sync::Arc; #[tokio::main] diff --git a/pmoplaylist/examples/radio_streaming.rs b/pmoplaylist/examples/radio_streaming.rs index 69d4ae54..2b3fe2a6 100644 --- a/pmoplaylist/examples/radio_streaming.rs +++ b/pmoplaylist/examples/radio_streaming.rs @@ -10,7 +10,7 @@ //! cargo run -p pmoplaylist --example radio_streaming //! ``` -use pmoplaylist::{DEFAULT_IMAGE, FifoPlaylist, Track}; +use pmoplaylist::{FifoPlaylist, Track, DEFAULT_IMAGE}; use std::time::Duration; use tokio::time::sleep; diff --git a/pmoplaylist/src/error.rs b/pmoplaylist/src/error.rs index 4198d15e..7c8f8562 100644 --- a/pmoplaylist/src/error.rs +++ b/pmoplaylist/src/error.rs @@ -5,31 +5,31 @@ pub enum Error { #[error("Playlist not found: {0}")] PlaylistNotFound(String), - + #[error("Playlist deleted: {0}")] PlaylistDeleted(String), - + #[error("Playlist already exists: {0}")] PlaylistAlreadyExists(String), - + #[error("Playlist is not persistent: {0}")] PlaylistNotPersistent(String), - + #[error("Write lock already held for playlist: {0}")] WriteLockHeld(String), - + #[error("Cache entry not found: {0}")] CacheEntryNotFound(String), - + #[error("Cache error: {0}")] CacheError(String), - + #[error("Persistence error: {0}")] PersistenceError(String), - + #[error("PlaylistManager not initialized")] ManagerNotInitialized, - + #[error(transparent)] Other(#[from] anyhow::Error), } diff --git a/pmoplaylist/src/handle/read.rs b/pmoplaylist/src/handle/read.rs index 8bcc6950..dec1433b 100644 --- a/pmoplaylist/src/handle/read.rs +++ b/pmoplaylist/src/handle/read.rs @@ -22,7 +22,7 @@ impl ReadHandle { cursor: AtomicUsize::new(0), } } - + /// Pop le prochain morceau (avance le curseur) /// /// Skip automatiquement les entrées invalides dans le cache. @@ -31,24 +31,24 @@ impl ReadHandle { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + let pos = self.cursor.load(Ordering::SeqCst); - + let core = self.playlist.core.read().await; - + // Fin de playlist ? if pos >= core.len() { return Ok(None); } - + let record = match core.get(pos) { Some(r) => r, None => return Ok(None), }; - + let cache_pk = record.cache_pk.clone(); drop(core); - + // Vérifier validité dans le cache let cache = crate::manager::audio_cache()?; if cache.is_valid_pk(&cache_pk) { @@ -61,31 +61,33 @@ impl ReadHandle { let mut core = self.playlist.core.write().await; core.remove_by_cache_pk(&cache_pk); drop(core); - + // Sauvegarder si persistante if self.playlist.persistent { if let Some(persistence) = crate::manager::PlaylistManager().persistence() { let title = self.playlist.title().await; let core = self.playlist.core.read().await; - let _ = persistence.save_playlist(&self.playlist.id, &title, &core.config, &core.tracks).await; + let _ = persistence + .save_playlist(&self.playlist.id, &title, &core.config, &core.tracks) + .await; } } - + // Ne pas avancer le curseur, continuer avec la position actuelle continue; } } } - + /// Peek le prochain morceau sans avancer le curseur pub async fn peek(&self) -> Result> { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + let pos = self.cursor.load(Ordering::SeqCst); let core = self.playlist.core.read().await; - + match core.get(pos) { Some(record) => { // Vérifier validité @@ -99,28 +101,28 @@ impl ReadHandle { None => Ok(None), } } - + /// Position actuelle du curseur pub fn position(&self) -> usize { self.cursor.load(Ordering::SeqCst) } - + /// Nombre de morceaux restants (compte uniquement les valides) pub async fn remaining(&self) -> Result { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + let pos = self.cursor.load(Ordering::SeqCst); let core = self.playlist.core.read().await; - + if pos >= core.len() { return Ok(0); } - + let cache = crate::manager::audio_cache()?; let mut count = 0; - + for i in pos..core.len() { if let Some(record) = core.get(i) { if cache.is_valid_pk(&record.cache_pk) { @@ -128,38 +130,38 @@ impl ReadHandle { } } } - + Ok(count) } - + /// Crée un nouveau handle avec cursor à 0 pub fn get_new_handle(&self) -> Result { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + Ok(ReadHandle::new(self.playlist.clone())) } - + /// Vérifie si la playlist est vivante pub fn is_alive(&self) -> bool { self.playlist.is_alive() } - + /// ID de la playlist pub fn id(&self) -> &str { &self.playlist.id } - + /// Génère un Container DIDL-Lite pub async fn to_container(&self) -> Result { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + let title = self.playlist.title().await; let remaining = self.remaining().await?; - + Ok(Container { id: self.playlist.id.clone(), parent_id: "0".to_string(), @@ -172,48 +174,48 @@ impl ReadHandle { items: vec![], }) } - + /// Génère des Items DIDL-Lite depuis la position actuelle pub async fn to_items(&self, limit: usize) -> Result> { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + let pos = self.cursor.load(Ordering::SeqCst); let core = self.playlist.core.read().await; let cache = crate::manager::audio_cache()?; - + // Récupérer base_url depuis le cache (via route_for) let mut items = Vec::new(); let mut idx = 0; - + for i in pos..core.len() { if items.len() >= limit { break; } - + let record = match core.get(i) { Some(r) => r, None => continue, }; - + // Vérifier validité if !cache.is_valid_pk(&record.cache_pk) { continue; } - + // Charger métadonnées let metadata = match pmoaudiocache::get_metadata(&*cache, &record.cache_pk) { Ok(m) => m, Err(_) => continue, }; - + // Construire l'URL via route_for let url = cache.route_for(&record.cache_pk, None); - + // Créer le Resource DIDL let resource = metadata.to_didl_resource(url); - + // Créer l'Item let item = Item { id: format!("{}:{}", self.playlist.id, pos + idx), @@ -232,11 +234,11 @@ impl ReadHandle { resources: vec![resource], descriptions: vec![], }; - + items.push(item); idx += 1; } - + Ok(items) } } diff --git a/pmoplaylist/src/handle/write.rs b/pmoplaylist/src/handle/write.rs index 9806338f..0d02aec9 100644 --- a/pmoplaylist/src/handle/write.rs +++ b/pmoplaylist/src/handle/write.rs @@ -22,41 +22,41 @@ impl WriteHandle { _write_token: write_token, } } - + /// Ajoute un morceau à la playlist pub async fn push(&self, cache_pk: String) -> Result<()> { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + // Vérifier que le pk existe dans le cache let cache = crate::manager::audio_cache()?; if !cache.is_valid_pk(&cache_pk) { return Err(crate::Error::CacheEntryNotFound(cache_pk)); } - + // Ajouter à la playlist let record = Record::new(cache_pk); let mut core = self.playlist.core.write().await; core.push(record); drop(core); - + self.playlist.touch().await; - + // Sauvegarder si persistante if self.playlist.persistent { self.save_to_db().await?; } - + Ok(()) } - + /// Ajoute plusieurs morceaux de manière atomique pub async fn push_set(&self, cache_pks: Vec) -> Result<()> { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + // Vérifier tous les pks d'abord let cache = crate::manager::audio_cache()?; for pk in &cache_pks { @@ -64,195 +64,194 @@ impl WriteHandle { return Err(crate::Error::CacheEntryNotFound(pk.clone())); } } - + // Créer tous les records - let records: Vec = cache_pks.into_iter() - .map(Record::new) - .collect(); - + let records: Vec = cache_pks.into_iter().map(Record::new).collect(); + // Ajouter atomiquement let mut core = self.playlist.core.write().await; core.push_all(records); drop(core); - + self.playlist.touch().await; - + // Une seule sauvegarde pour tout le batch if self.playlist.persistent { self.save_to_db().await?; } - + Ok(()) } - + /// Vide la playlist pub async fn flush(&self) -> Result<()> { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + let mut core = self.playlist.core.write().await; core.clear(); drop(core); - + self.playlist.touch().await; - + if self.playlist.persistent { self.save_to_db().await?; } - + Ok(()) } - + /// Supprime la playlist définitivement pub async fn delete(self) -> Result<()> { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + // Marquer comme supprimée self.playlist.mark_deleted(); - + // Supprimer du manager crate::manager::delete_playlist_internal(&self.playlist.id).await?; - + Ok(()) } - + /// Change le titre pub async fn set_title(&self, title: String) -> Result<()> { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + self.playlist.set_title(title).await; - + if self.playlist.persistent { self.save_to_db().await?; } - + Ok(()) } - + /// Change la capacité maximale pub async fn set_capacity(&self, max_size: Option) -> Result<()> { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + let mut core = self.playlist.core.write().await; core.set_capacity(max_size); drop(core); - + self.playlist.touch().await; - + if self.playlist.persistent { self.save_to_db().await?; } - + Ok(()) } - + /// Change le TTL par défaut pub async fn set_default_ttl(&self, ttl: Option) -> Result<()> { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + let mut core = self.playlist.core.write().await; core.set_default_ttl(ttl); drop(core); - + self.playlist.touch().await; - + if self.playlist.persistent { self.save_to_db().await?; } - + Ok(()) } - + /// Clone vers une nouvelle playlist persistante pub async fn clone_as_persistent(&self, new_id: String) -> Result { if !self.playlist.is_alive() { return Err(crate::Error::PlaylistDeleted(self.playlist.id.clone())); } - + // Récupérer les données actuelles let title = self.playlist.title().await; let core = self.playlist.core.read().await; let config = core.config.clone(); let tracks = core.snapshot(); drop(core); - + // Créer la nouvelle playlist persistante let manager = crate::manager::PlaylistManager(); let mut new_handle = manager.create_persistent_playlist(new_id).await?; - + // Copier le titre et la config new_handle.set_title(title).await?; new_handle.set_capacity(config.max_size).await?; new_handle.set_default_ttl(config.default_ttl).await?; - + // Copier tous les morceaux - let pks: Vec = tracks.iter() - .map(|r| r.cache_pk.clone()) - .collect(); + let pks: Vec = tracks.iter().map(|r| r.cache_pk.clone()).collect(); new_handle.push_set(pks).await?; - + Ok(new_handle) } - + // Métadonnées - + pub fn id(&self) -> &str { &self.playlist.id } - + pub async fn title(&self) -> String { self.playlist.title().await } - + pub fn is_persistent(&self) -> bool { self.playlist.persistent } - + pub async fn capacity(&self) -> Option { let core = self.playlist.core.read().await; core.config.max_size } - + pub async fn default_ttl(&self) -> Option { let core = self.playlist.core.read().await; core.config.default_ttl } - + pub async fn len(&self) -> usize { let core = self.playlist.core.read().await; core.len() } - + pub async fn is_empty(&self) -> bool { let core = self.playlist.core.read().await; core.is_empty() } - + pub async fn last_change(&self) -> SystemTime { self.playlist.last_change().await } - + // Helpers internes - + async fn save_to_db(&self) -> Result<()> { let manager = crate::manager::PlaylistManager(); - let persistence = manager.persistence() + let persistence = manager + .persistence() .ok_or_else(|| crate::Error::PersistenceError("No persistence manager".into()))?; - + let title = self.playlist.title().await; let core = self.playlist.core.read().await; let config = &core.config; let tracks = &core.tracks; - - persistence.save_playlist(&self.playlist.id, &title, config, tracks).await + + persistence + .save_playlist(&self.playlist.id, &title, config, tracks) + .await } } diff --git a/pmoplaylist/src/manager.rs b/pmoplaylist/src/manager.rs index 333ee165..e622b2b1 100644 --- a/pmoplaylist/src/manager.rs +++ b/pmoplaylist/src/manager.rs @@ -70,7 +70,9 @@ impl PlaylistManager { #[cfg(not(feature = "pmoconfig"))] { - PLAYLIST_MANAGER.get().expect("PlaylistManager not initialized. Call init() first.") + PLAYLIST_MANAGER + .get() + .expect("PlaylistManager not initialized. Call init() first.") } } @@ -102,7 +104,8 @@ impl PlaylistManager { if let Some(persistence) = &self.inner.persistence { let title = playlist.title().await; let core = playlist.core.read().await; - persistence.save_playlist(&playlist.id, &title, &core.config, &core.tracks) + persistence + .save_playlist(&playlist.id, &title, &core.config, &core.tracks) .await?; } @@ -185,12 +188,7 @@ impl PlaylistManager { // Reconstruire la playlist let mut playlists = self.inner.playlists.write().await; - let playlist = Arc::new(Playlist::new( - id.to_string(), - title.clone(), - config, - true, - )); + let playlist = Arc::new(Playlist::new(id.to_string(), title.clone(), config, true)); // Restaurer les tracks { @@ -265,7 +263,9 @@ impl PlaylistManager { if let Some(persistence) = &self.inner.persistence { let title = playlist.title().await; let core = playlist.core.read().await; - let _ = persistence.save_playlist(&playlist.id, &title, &core.config, &core.tracks).await; + let _ = persistence + .save_playlist(&playlist.id, &title, &core.config, &core.tracks) + .await; } } } @@ -286,8 +286,7 @@ pub(crate) async fn delete_playlist_internal(id: &str) -> Result<()> { /// Helper pour acc�der au cache audio pub(crate) fn audio_cache() -> Result> { - pmoupnp::get_audio_cache() - .ok_or_else(|| crate::Error::ManagerNotInitialized) + pmoupnp::get_audio_cache().ok_or_else(|| crate::Error::ManagerNotInitialized) } /// Fonction raccourcie pour acc�der au singleton diff --git a/pmoplaylist/src/persistence/mod.rs b/pmoplaylist/src/persistence/mod.rs index 890a4616..5d53b8ba 100644 --- a/pmoplaylist/src/persistence/mod.rs +++ b/pmoplaylist/src/persistence/mod.rs @@ -19,13 +19,15 @@ impl PersistenceManager { pub fn new(db_path: &Path) -> Result { // Créer le répertoire parent si nécessaire if let Some(parent) = db_path.parent() { - std::fs::create_dir_all(parent) - .map_err(|e| crate::Error::PersistenceError(format!("Failed to create directory: {}", e)))?; + std::fs::create_dir_all(parent).map_err(|e| { + crate::Error::PersistenceError(format!("Failed to create directory: {}", e)) + })?; } - - let conn = Connection::open(db_path) - .map_err(|e| crate::Error::PersistenceError(format!("Failed to open database: {}", e)))?; - + + let conn = Connection::open(db_path).map_err(|e| { + crate::Error::PersistenceError(format!("Failed to open database: {}", e)) + })?; + // Créer les tables conn.execute( "CREATE TABLE IF NOT EXISTS playlists ( @@ -37,8 +39,11 @@ impl PersistenceManager { last_modified INTEGER NOT NULL )", [], - ).map_err(|e| crate::Error::PersistenceError(format!("Failed to create playlists table: {}", e)))?; - + ) + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to create playlists table: {}", e)) + })?; + conn.execute( "CREATE TABLE IF NOT EXISTS tracks ( playlist_id TEXT NOT NULL, @@ -48,23 +53,28 @@ impl PersistenceManager { FOREIGN KEY (playlist_id) REFERENCES playlists(id) ON DELETE CASCADE )", [], - ).map_err(|e| crate::Error::PersistenceError(format!("Failed to create tracks table: {}", e)))?; - + ) + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to create tracks table: {}", e)) + })?; + conn.execute( "CREATE INDEX IF NOT EXISTS idx_tracks_playlist ON tracks(playlist_id, added_at)", [], - ).map_err(|e| crate::Error::PersistenceError(format!("Failed to create index: {}", e)))?; - + ) + .map_err(|e| crate::Error::PersistenceError(format!("Failed to create index: {}", e)))?; + conn.execute( "CREATE INDEX IF NOT EXISTS idx_tracks_cache_pk ON tracks(cache_pk)", [], - ).map_err(|e| crate::Error::PersistenceError(format!("Failed to create index: {}", e)))?; - + ) + .map_err(|e| crate::Error::PersistenceError(format!("Failed to create index: {}", e)))?; + Ok(Self { conn: Arc::new(Mutex::new(conn)), }) } - + /// Sauvegarde une playlist complète pub async fn save_playlist( &self, @@ -74,12 +84,12 @@ impl PersistenceManager { tracks: &VecDeque>, ) -> Result<()> { let conn = self.conn.lock().unwrap(); - + let now_nanos = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_nanos() as i64; - + // Upsert playlist metadata conn.execute( "INSERT OR REPLACE INTO playlists (id, title, max_size, default_ttl_secs, created_at, last_modified) @@ -94,13 +104,13 @@ impl PersistenceManager { now_nanos, ], ).map_err(|e| crate::Error::PersistenceError(format!("Failed to save playlist: {}", e)))?; - + // Supprimer les anciens tracks - conn.execute( - "DELETE FROM tracks WHERE playlist_id = ?1", - params![id], - ).map_err(|e| crate::Error::PersistenceError(format!("Failed to delete old tracks: {}", e)))?; - + conn.execute("DELETE FROM tracks WHERE playlist_id = ?1", params![id]) + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to delete old tracks: {}", e)) + })?; + // Insérer les nouveaux tracks for record in tracks { conn.execute( @@ -112,26 +122,34 @@ impl PersistenceManager { &record.cache_pk, record.ttl.map(|d| d.as_secs() as i64), ], - ).map_err(|e| crate::Error::PersistenceError(format!("Failed to insert track: {}", e)))?; + ) + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to insert track: {}", e)) + })?; } - + Ok(()) } - + /// Charge une playlist - pub async fn load_playlist(&self, id: &str) -> Result>)>> { + pub async fn load_playlist( + &self, + id: &str, + ) -> Result>)>> { let conn = self.conn.lock().unwrap(); - + // Charger les métadonnées - let mut stmt = conn.prepare( - "SELECT title, max_size, default_ttl_secs FROM playlists WHERE id = ?1" - ).map_err(|e| crate::Error::PersistenceError(format!("Failed to prepare statement: {}", e)))?; - + let mut stmt = conn + .prepare("SELECT title, max_size, default_ttl_secs FROM playlists WHERE id = ?1") + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to prepare statement: {}", e)) + })?; + let result = stmt.query_row(params![id], |row| { let title: String = row.get(0)?; let max_size: Option = row.get(1)?; let default_ttl_secs: Option = row.get(2)?; - + Ok(( title, PlaylistConfig { @@ -140,76 +158,91 @@ impl PersistenceManager { }, )) }); - + let (title, config) = match result { Ok(data) => data, Err(rusqlite::Error::QueryReturnedNoRows) => return Ok(None), - Err(e) => return Err(crate::Error::PersistenceError(format!("Failed to load playlist: {}", e))), + Err(e) => { + return Err(crate::Error::PersistenceError(format!( + "Failed to load playlist: {}", + e + ))) + } }; - + // Charger les tracks let mut stmt = conn.prepare( "SELECT added_at, cache_pk, ttl_secs FROM tracks WHERE playlist_id = ?1 ORDER BY added_at ASC" ).map_err(|e| crate::Error::PersistenceError(format!("Failed to prepare statement: {}", e)))?; - - let rows = stmt.query_map(params![id], |row| { - let added_at_nanos: i64 = row.get(0)?; - let cache_pk: String = row.get(1)?; - let ttl_secs: Option = row.get(2)?; - - let added_at = UNIX_EPOCH + Duration::from_nanos(added_at_nanos as u64); - let ttl = ttl_secs.map(|s| Duration::from_secs(s as u64)); - - Ok(Record { - cache_pk, - added_at, - ttl, + + let rows = stmt + .query_map(params![id], |row| { + let added_at_nanos: i64 = row.get(0)?; + let cache_pk: String = row.get(1)?; + let ttl_secs: Option = row.get(2)?; + + let added_at = UNIX_EPOCH + Duration::from_nanos(added_at_nanos as u64); + let ttl = ttl_secs.map(|s| Duration::from_secs(s as u64)); + + Ok(Record { + cache_pk, + added_at, + ttl, + }) }) - }).map_err(|e| crate::Error::PersistenceError(format!("Failed to query tracks: {}", e)))?; - + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to query tracks: {}", e)) + })?; + let mut tracks = VecDeque::new(); for row in rows { - let record = row.map_err(|e| crate::Error::PersistenceError(format!("Failed to read track: {}", e)))?; + let record = row.map_err(|e| { + crate::Error::PersistenceError(format!("Failed to read track: {}", e)) + })?; tracks.push_back(Arc::new(record)); } - + Ok(Some((title, config, tracks))) } - + /// Supprime une playlist pub async fn delete_playlist(&self, id: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); - conn.execute( - "DELETE FROM playlists WHERE id = ?1", - params![id], - ).map_err(|e| crate::Error::PersistenceError(format!("Failed to delete playlist: {}", e)))?; + conn.execute("DELETE FROM playlists WHERE id = ?1", params![id]) + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to delete playlist: {}", e)) + })?; Ok(()) } - + /// Liste toutes les playlists persistantes pub async fn list_playlist_ids(&self) -> Result> { let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare("SELECT id FROM playlists") - .map_err(|e| crate::Error::PersistenceError(format!("Failed to prepare statement: {}", e)))?; - - let rows = stmt.query_map([], |row| row.get(0)) - .map_err(|e| crate::Error::PersistenceError(format!("Failed to query playlists: {}", e)))?; - + let mut stmt = conn.prepare("SELECT id FROM playlists").map_err(|e| { + crate::Error::PersistenceError(format!("Failed to prepare statement: {}", e)) + })?; + + let rows = stmt.query_map([], |row| row.get(0)).map_err(|e| { + crate::Error::PersistenceError(format!("Failed to query playlists: {}", e)) + })?; + let mut ids = Vec::new(); for row in rows { - ids.push(row.map_err(|e| crate::Error::PersistenceError(format!("Failed to read id: {}", e)))?); + ids.push(row.map_err(|e| { + crate::Error::PersistenceError(format!("Failed to read id: {}", e)) + })?); } - + Ok(ids) } - + /// Supprime tous les tracks contenant un cache_pk donné pub async fn remove_by_cache_pk(&self, cache_pk: &str) -> Result<()> { let conn = self.conn.lock().unwrap(); - conn.execute( - "DELETE FROM tracks WHERE cache_pk = ?1", - params![cache_pk], - ).map_err(|e| crate::Error::PersistenceError(format!("Failed to remove tracks: {}", e)))?; + conn.execute("DELETE FROM tracks WHERE cache_pk = ?1", params![cache_pk]) + .map_err(|e| { + crate::Error::PersistenceError(format!("Failed to remove tracks: {}", e)) + })?; Ok(()) } } diff --git a/pmoplaylist/src/playlist/core.rs b/pmoplaylist/src/playlist/core.rs index bc75885d..dbc3a2b2 100644 --- a/pmoplaylist/src/playlist/core.rs +++ b/pmoplaylist/src/playlist/core.rs @@ -35,13 +35,13 @@ impl PlaylistCore { config, } } - + /// Ajoute un record et applique l'éviction pub fn push(&mut self, record: Record) { self.tracks.push_back(Arc::new(record)); self.evict(); } - + /// Ajoute plusieurs records de manière atomique pub fn push_all(&mut self, records: Vec) { for record in records { @@ -49,14 +49,13 @@ impl PlaylistCore { } self.evict(); } - + /// Nettoie les morceaux expirés et applique la limite de taille pub fn evict(&mut self) { // 1. Supprimer les morceaux périmés par TTL - self.tracks.retain(|record| { - !record.is_expired(self.config.default_ttl) - }); - + self.tracks + .retain(|record| !record.is_expired(self.config.default_ttl)); + // 2. Appliquer la limite de taille (FIFO) if let Some(max) = self.config.max_size { while self.tracks.len() > max { @@ -64,45 +63,45 @@ impl PlaylistCore { } } } - + /// Vide complètement la playlist pub fn clear(&mut self) { self.tracks.clear(); } - + /// Nombre de morceaux pub fn len(&self) -> usize { self.tracks.len() } - + /// Vérifie si la playlist est vide pub fn is_empty(&self) -> bool { self.tracks.is_empty() } - + /// Récupère un record par index pub fn get(&self, index: usize) -> Option> { self.tracks.get(index).cloned() } - + /// Snapshot de tous les records pub fn snapshot(&self) -> Vec> { self.tracks.iter().cloned().collect() } - + /// Supprime un record par cache_pk (retourne true si supprimé) pub fn remove_by_cache_pk(&mut self, cache_pk: &str) -> bool { let initial_len = self.tracks.len(); self.tracks.retain(|r| r.cache_pk != cache_pk); self.tracks.len() != initial_len } - + /// Met à jour la capacité maximale pub fn set_capacity(&mut self, max_size: Option) { self.config.max_size = max_size; self.evict(); } - + /// Met à jour le TTL par défaut pub fn set_default_ttl(&mut self, ttl: Option) { self.config.default_ttl = ttl; diff --git a/pmoplaylist/src/playlist/mod.rs b/pmoplaylist/src/playlist/mod.rs index d6e11e5e..1e9cc5be 100644 --- a/pmoplaylist/src/playlist/mod.rs +++ b/pmoplaylist/src/playlist/mod.rs @@ -51,49 +51,50 @@ impl Playlist { writer_lock: RwLock::new(None), } } - + /// Vérifie si la playlist est active pub fn is_alive(&self) -> bool { PlaylistState::from(self.state.load(Ordering::SeqCst)) == PlaylistState::Active } - + /// Marque la playlist comme supprimée pub fn mark_deleted(&self) { - self.state.store(PlaylistState::Deleted as u8, Ordering::SeqCst); + self.state + .store(PlaylistState::Deleted as u8, Ordering::SeqCst); } - + /// Met à jour le timestamp de dernière modification pub async fn touch(&self) { *self.last_change.write().await = SystemTime::now(); } - + /// Récupère le titre pub async fn title(&self) -> String { self.title.read().await.clone() } - + /// Change le titre pub async fn set_title(&self, title: String) { *self.title.write().await = title; self.touch().await; } - + /// Timestamp du dernier changement pub async fn last_change(&self) -> SystemTime { *self.last_change.read().await } - + /// Tente d'acquérir le write lock pub async fn acquire_write_lock(&self) -> Result, ()> { let mut guard = self.writer_lock.write().await; - + // Vérifier si un writer existe déjà if let Some(weak) = guard.as_ref() { if weak.strong_count() > 0 { return Err(()); // Lock déjà pris } } - + // Créer un nouveau token let token = Arc::new(()); *guard = Some(Arc::downgrade(&token)); diff --git a/pmoplaylist/src/playlist/record.rs b/pmoplaylist/src/playlist/record.rs index 7dec95b5..521e605d 100644 --- a/pmoplaylist/src/playlist/record.rs +++ b/pmoplaylist/src/playlist/record.rs @@ -10,10 +10,10 @@ use std::time::{Duration, SystemTime}; pub struct Record { /// Clé primaire dans pmoaudiocache pub cache_pk: String, - + /// Timestamp d'ajout à la playlist (en nanosecondes depuis epoch) pub added_at: SystemTime, - + /// Durée de vie optionnelle (surcharge le TTL par défaut) pub ttl: Option, } @@ -27,7 +27,7 @@ impl Record { ttl: None, } } - + /// Crée un record avec un TTL personnalisé pub fn with_ttl(cache_pk: String, ttl: Duration) -> Self { Self { @@ -36,12 +36,12 @@ impl Record { ttl: Some(ttl), } } - + /// Vérifie si le record est expiré pub fn is_expired(&self, default_ttl: Option) -> bool { let now = SystemTime::now(); let age = now.duration_since(self.added_at).unwrap_or_default(); - + if let Some(ttl) = self.ttl { age >= ttl } else if let Some(default_ttl) = default_ttl { @@ -50,7 +50,7 @@ impl Record { false } } - + /// Retourne le timestamp en nanosecondes depuis epoch pub fn added_at_nanos(&self) -> i64 { self.added_at diff --git a/pmoplaylist/src/track.rs b/pmoplaylist/src/track.rs index 883791af..d4439b83 100644 --- a/pmoplaylist/src/track.rs +++ b/pmoplaylist/src/track.rs @@ -1,8 +1,8 @@ //! PlaylistTrack : résultat d'un pop() avec helpers pour accéder au cache use crate::Result; -use pmocache::cache_trait::FileCache; use pmoaudiocache::AudioMetadataExt; +use pmocache::cache_trait::FileCache; use std::path::PathBuf; /// Un morceau récupéré depuis une playlist diff --git a/pmoupnp/src/devices/device_instance.rs b/pmoupnp/src/devices/device_instance.rs index dd8aa27f..e8b9b88c 100644 --- a/pmoupnp/src/devices/device_instance.rs +++ b/pmoupnp/src/devices/device_instance.rs @@ -149,8 +149,7 @@ impl UpnpObject for DeviceInstance { // UDN let mut udn = Element::new("UDN"); - udn.children - .push(XMLNode::Text(self.udn_with_prefix())); + udn.children.push(XMLNode::Text(self.udn_with_prefix())); elem.children.push(XMLNode::Element(udn)); // serviceList