eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

🗑️ Remove unused imports, macros and dead code

Browse Source 
- Drop `Path`/``State``` from unused Axum imports in config.rs and registry
- Mark `_position_sec` field as `#[allow(dead_code)]`` in PositionUpdateRequest and PlayerStateReport
- Remove unused macro rules (`add_action_arg!`, `add_action!``, `` add_var!)``
- Delete unused PlayerReport struct and related handler code
This commit is contained in:
Eric Coissac
2026-04-05 10:52:54 +02:00
parent 340c69cb2b
commit 546e8a782f
21 changed files with 1479 additions and 451 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
pmoaudio-ext/src/sinks/streaming_flac_sink.rs
View File

@@ -533,6 +533,9 @@ impl StreamingFlacSink {
current_timestamp: Arc::new(RwLock::new(0.0)),
pending_track_duration: None,
pending_total_samples: None,
is_paused: Arc::new(AtomicBool::new(false)),
stream_type: Arc::new(RwLock::new(pmoaudio::StreamType::Finite)),
last_track_metadata: Arc::new(RwLock::new(None)),
},
};

188
pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs
View File

@@ -61,7 +61,7 @@ use async_trait::async_trait;
use bytes::Bytes;
use pmoaudio::{
pipeline::{AudioPipelineNode, Node, NodeLogic, PipelineHandle, StopReason},
AudioError, AudioSegment, SyncMarker, TypeRequirement, TypedAudioNode, _AudioSegment,
AudioChunk, AudioError, AudioSegment, SyncMarker, TypeRequirement, TypedAudioNode, _AudioSegment,
};
use pmoflac::{EncoderOptions, FlacEncodedStream};
use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
@@ -109,6 +109,23 @@ impl OggFlacStreamHandle {
pub fn set_auto_stop(&self, enabled: bool) {
self.inner.auto_stop.store(enabled, Ordering::SeqCst);
}
/// Pause playback - sends silence (zeros) to maintain client connections
pub fn pause(&self) {
self.inner.is_paused.store(true, Ordering::SeqCst);
debug!("StreamingOggFlacSink paused");
}
/// Resume playback - continues sending actual audio
pub fn resume(&self) {
self.inner.is_paused.store(false, Ordering::SeqCst);
debug!("StreamingOggFlacSink resumed");
}
/// Check if currently paused
pub fn is_paused(&self) -> bool {
self.inner.is_paused.load(Ordering::SeqCst)
}
}
/// OGG-FLAC client stream (implements AsyncRead).
@@ -163,16 +180,98 @@ impl NodeLogic for StreamingOggFlacSinkLogic {
debug!("StreamingOggFlacSink started");
// TODO: Implement OGG-FLAC encoding logic
// For now, just process segments without encoding
let mut was_paused = false;
let mut silence_timestamp: f64 = 0.0;
loop {
// Check pause state BEFORE select to avoid receiving chunks when Finite + paused
{
let is_paused = self.ctx.is_paused.load(Ordering::SeqCst);
let stream_type = *self.ctx.stream_type.read().await;
// For Finite + paused: loop WITHOUT receiving chunks
if is_paused && stream_type == pmoaudio::StreamType::Finite {
if self.ctx.sample_rate.is_none() {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
continue;
}
let sample_rate = self.ctx.sample_rate.unwrap();
let frames_per_50ms = (sample_rate as f64 * 0.05) as usize;
// Create proper silence chunk using AudioChunk::silence
let silence_chunk = AudioChunk::silence(frames_per_50ms, sample_rate);
let pcm_bytes = chunk_to_pcm_bytes(&silence_chunk, self.ctx.bits_per_sample)?;
trace!("Sending silence during pause (finite) - no recv");
let pcm_chunk = PcmChunk {
bytes: pcm_bytes,
timestamp_sec: silence_timestamp,
duration_sec: 0.05,
};
if let Some(tx) = &self.ctx.pcm_tx {
let _ = tx.send(pcm_chunk).await;
}
silence_timestamp += 0.05;
continue; // Loop again WITHOUT going to select
}
}
tokio::select! {
_ = stop_token.cancelled() => {
debug!("StreamingOggFlacSink stopped by cancellation");
break;
}
// Check for pause state changes periodically
_ = tokio::time::sleep(std::time::Duration::from_millis(50)) => {
let is_paused = self.ctx.is_paused.load(Ordering::SeqCst);
if is_paused != was_paused {
if is_paused {
// Entering pause - restart encoder to close OGG segment
debug!("StreamingOggFlacSink: entering pause - restarting encoder");
let metadata = self.ctx.last_track_metadata.read().await.clone();
if let Some(ref meta) = metadata {
if let Err(e) = self.ctx.prepare_encoder_options_for_track(meta).await {
error!("Failed to prepare encoder options for pause: {}", e);
}
}
if self.ctx.encoder_state.is_some() {
if let Err(e) = self.ctx.restart_encoder_for_new_track(
|flac_stream, broadcast, header, current_timestamp, current_duration, max_lead, _sample_rate, timestamp_offset_sec| {
broadcast_ogg_flac_stream(flac_stream, broadcast, header, current_timestamp, current_duration, max_lead, timestamp_offset_sec)
},
).await {
warn!("Failed to restart encoder on pause: {}", e);
}
}
} else {
// Exiting pause - restart encoder to open new OGG segment
debug!("StreamingOggFlacSink: exiting pause - restarting encoder");
let metadata = self.ctx.last_track_metadata.read().await.clone();
if let Some(ref meta) = metadata {
if let Err(e) = self.ctx.prepare_encoder_options_for_track(meta).await {
error!("Failed to prepare encoder options for resume: {}", e);
}
}
if self.ctx.encoder_state.is_some() {
if let Err(e) = self.ctx.restart_encoder_for_new_track(
|flac_stream, broadcast, header, current_timestamp, current_duration, max_lead, _sample_rate, timestamp_offset_sec| {
broadcast_ogg_flac_stream(flac_stream, broadcast, header, current_timestamp, current_duration, max_lead, timestamp_offset_sec)
},
).await {
warn!("Failed to restart encoder on resume: {}", e);
}
}
}
was_paused = is_paused;
}
continue; // Skip processing - this is just a check
}
segment = input.recv() => {
match segment {
Some(seg) => {
@@ -241,7 +340,33 @@ impl NodeLogic for StreamingOggFlacSinkLogic {
duration_sec
);
// Send to FLAC encoder with timestamp and duration
// Check pause state
let was_paused = self.ctx.is_paused.load(Ordering::SeqCst);
// Get stream type
let stream_type = *self.ctx.stream_type.read().await;
// Handle pause:
// - Always send silence to hear something
// - Continuous: receive chunks but drop them
// - Finite: don't receive chunks (backpressure)
if was_paused {
trace!("StreamingOggFlacSink: sending silence during pause ({})",
if stream_type == pmoaudio::StreamType::Continuous { "continuous" } else { "finite" });
let silence_bytes = vec![0u8; pcm_bytes.len()];
let silence_chunk = PcmChunk {
bytes: silence_bytes,
timestamp_sec: seg.timestamp_sec,
duration_sec,
};
if let Some(tx) = &self.ctx.pcm_tx {
let _ = tx.send(silence_chunk).await;
}
// Always continue (skip sending real audio)
continue;
}
// Normal case: send PCM to encoder
let pcm_chunk = PcmChunk {
bytes: pcm_bytes,
timestamp_sec: seg.timestamp_sec,
@@ -265,7 +390,13 @@ impl NodeLogic for StreamingOggFlacSinkLogic {
_AudioSegment::Sync(marker) => {
match marker.as_ref() {
SyncMarker::TrackBoundary { metadata, .. } => {
SyncMarker::TrackBoundary { metadata, stream_type } => {
// Store the last TrackBoundary metadata for pause/resume
*self.ctx.last_track_metadata.write().await = Some(metadata.clone());
// Update stream type
*self.ctx.stream_type.write().await = *stream_type;
// Inject per-track metadata and duration into the next FLAC header.
if let Err(e) =
self.ctx.prepare_encoder_options_for_track(metadata).await
@@ -442,30 +573,33 @@ impl StreamingOggFlacSink {
let logic = StreamingOggFlacSinkLogic {
ctx: SharedSinkContext {
encoder_options,
bits_per_sample,
enable_total_samples: options.enable_total_samples,
restart_encoder_on_track_boundary: options.restart_encoder_on_track_boundary,
default_title: options.default_title.clone(),
default_artist: options.default_artist.clone(),
use_only_default_metadata: options.use_only_default_metadata,
pcm_tx: Some(pcm_tx),
pcm_rx: Some(pcm_rx),
metadata,
broadcast,
header,
encoder_state: None,
sample_rate: None,
broadcast_max_lead_time: broadcast_max_lead_time.max(0.0),
first_chunk_timestamp_checked: false,
timestamp_offset_sec: 0.0,
current_timestamp: Arc::new(RwLock::new(0.0)),
pending_track_duration: None,
pending_total_samples: None,
},
encoder_options,
bits_per_sample,
enable_total_samples: options.enable_total_samples,
restart_encoder_on_track_boundary: options.restart_encoder_on_track_boundary,
default_title: options.default_title.clone(),
default_artist: options.default_artist.clone(),
use_only_default_metadata: options.use_only_default_metadata,
pcm_tx: Some(pcm_tx),
pcm_rx: Some(pcm_rx),
metadata,
broadcast,
header,
encoder_state: None,
sample_rate: None,
broadcast_max_lead_time: broadcast_max_lead_time.max(0.0),
first_chunk_timestamp_checked: false,
timestamp_offset_sec: 0.0,
current_timestamp: Arc::new(RwLock::new(0.0)),
pending_track_duration: None,
pending_total_samples: None,
is_paused: Arc::new(AtomicBool::new(false)),
stream_type: Arc::new(RwLock::new(pmoaudio::StreamType::Finite)),
last_track_metadata: Arc::new(RwLock::new(None)),
},
};
let sink = Self {
let sink = StreamingOggFlacSink {
inner: Node::new_with_input(logic, 16),
};

12
pmoaudio-ext/src/sinks/streaming_sink_common.rs
View File

@@ -118,6 +118,10 @@ pub struct SharedStreamHandleInner {
pub stop_token: CancellationToken,
pub header: Arc<RwLock<Option<Bytes>>>,
pub auto_stop: Arc<AtomicBool>,
/// Pause state - true if paused, false if playing
pub is_paused: Arc<AtomicBool>,
/// Stream type (Continuous for radio, Finite for tracks)
pub stream_type: Arc<RwLock<pmoaudio::StreamType>>,
}
impl SharedStreamHandleInner {
@@ -135,6 +139,8 @@ impl SharedStreamHandleInner {
stop_token,
header,
auto_stop,
is_paused: Arc::new(AtomicBool::new(false)),
stream_type: Arc::new(RwLock::new(pmoaudio::StreamType::Finite)),
}
}
@@ -294,6 +300,12 @@ pub struct SharedSinkContext {
pub current_timestamp: Arc<RwLock<f64>>,
pub pending_track_duration: Option<Duration>,
pub pending_total_samples: Option<u64>,
/// Pause state - true if paused, false if playing
pub is_paused: Arc<AtomicBool>,
/// Stream type (Continuous for radio, Finite for tracks)
pub stream_type: Arc<RwLock<pmoaudio::StreamType>>,
/// Last TrackBoundary metadata received (for pause/resume)
pub last_track_metadata: Arc<RwLock<Option<Arc<RwLock<dyn TrackMetadata>>>>>,
}
impl SharedSinkContext {