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

refactor: Remove obsolete orchestration layer, create RadioParadiseSource stub

Browse Source 
Major cleanup removing 2831 lines (~45%) of outdated server orchestration code.
RadioParadiseStreamSource (pmoaudio integration) is now the primary implementation.

## Changes

### Removed (2393 lines)
- **paradise/ module** - Complete server orchestration system:
  - worker.rs (1146 lines) - Background polling, caching, state machine
  - channel.rs (429 lines) - Channel lifecycle management
  - playlist.rs (294 lines) - Shared playlist management
  - history.rs (218 lines) - SQLite persistence
  - constants.rs (209 lines) - Server configuration constants
  - mod.rs (29 lines) - Module exports

### Replaced
- **source.rs** (612 → 174 lines, -72%):
  - Old: Full MusicSource implementation with UPnP/DIDL integration
  - New: Minimal stub for backward compatibility with pmomediaserver
  - Returns empty results and deprecation warnings
  - Documents migration path to RadioParadiseStreamSource

### Updated
- **config_ext.rs**: Now imports HISTORY_DEFAULT_MAX_TRACKS from channels module
- **lib.rs**:
  - Removed paradise module
  - Updated documentation to focus on RadioParadiseStreamSource
  - Updated cargo features documentation

## Architecture

**Before**: Complex orchestration with workers, channels, caching, history
**After**: Simple API access + pmoaudio streaming (RadioParadiseStreamSource)

## Compatibility

RadioParadiseSource stub maintains API compatibility for pmomediaserver while
clearly indicating deprecation. All operations return empty results or errors
with migration guidance.

## Testing

- ✅ All 23 tests pass
- ✅ Compilation successful with all features
- ✅ pmomediaserver compatibility maintained (stub implementation)

## Migration Path

Old (deprecated):
```rust
let source = RadioParadiseSource::from_registry(client)?;
```

New (recommended):
```rust
let stream_source = RadioParadiseStreamSource::new(client, None).await?;
let node = Node::from_logic(stream_source);
```
This commit is contained in:
Claude
2025-11-05 07:40:36 +00:00
parent 23e07fb6cf
commit f80ebd9f3d
9 changed files with 137 additions and 2903 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
pmoparadise/src/config_ext.rs
View File

@@ -36,7 +36,7 @@ use anyhow::{anyhow, Result};
use pmoconfig::Config;
use serde_yaml::{Number, Value};
use crate::paradise::constants;
use crate::channels::HISTORY_DEFAULT_MAX_TRACKS;
/// Nom du répertoire pour Radio Paradise (relatif au config_dir)
///
@@ -222,7 +222,7 @@ impl RadioParadiseConfigExt for Config {
Ok(Value::Number(n)) if n.is_i64() => Ok(n.as_i64().unwrap() as usize),
_ => {
// Use default and persist it
let default = constants::HISTORY_DEFAULT_MAX_TRACKS;
let default = HISTORY_DEFAULT_MAX_TRACKS;
self.set_paradise_history_size(default)?;
Ok(default)
}
@@ -245,7 +245,7 @@ mod tests {
#[test]
fn test_default_values() {
assert_eq!(DEFAULT_HISTORY_DATABASE_DIR, "paradise");
assert_eq!(constants::HISTORY_DEFAULT_MAX_TRACKS, 100);
assert_eq!(HISTORY_DEFAULT_MAX_TRACKS, 100);
}
#[test]

49
pmoparadise/src/lib.rs
View File

@@ -164,54 +164,46 @@
//! }
//! ```
//!
//! ## Caching Support (Feature: `cache`)
//! ## Audio Streaming (Feature: `pmoaudio`)
//!
//! `pmoparadise` can optionally integrate with `pmocovers` and `pmoaudiocache` to cache
//! cover images and audio tracks locally:
//! For direct audio streaming and integration with pmoaudio pipelines,
//! use `RadioParadiseStreamSource`:
//!
//! ```no_run
//! # #[cfg(feature = "cache")]
//! # #[cfg(feature = "pmoaudio")]
//! # {
//! use pmoparadise::{RadioParadiseClient, RadioParadiseSource};
//! use std::sync::Arc;
//! use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource};
//! use pmoaudio::pipeline::Node;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! // Create caches
//! let cover_cache = Arc::new(pmocovers::cache::new_cache("./cache/covers", 500)?);
//! let audio_cache = Arc::new(pmoaudiocache::cache::new_cache("./cache/audio", 100)?);
//!
//! // Create client and source with caching
//! let client = RadioParadiseClient::new().await?;
//! let source = RadioParadiseSource::new(
//! client,
//! 50,
//! cover_cache,
//! audio_cache,
//! );
//! let stream_source = RadioParadiseStreamSource::new(client, None).await?;
//!
//! println!("Source ready: {}", source.name());
//! // Create audio node from stream source
//! let node = Node::from_logic(stream_source);
//!
//! // Use in pmoaudio pipeline...
//!
//! Ok(())
//! }
//! # }
//! ```
//!
//! **Benefits**:
//! - Cover images are automatically downloaded and converted to WebP
//! - Audio tracks are cached as FLAC with metadata preserved
//! - Subsequent access is instant (no re-download)
//! - URIs returned by `resolve_uri()` point to cached versions
//!
//! See the `with_cache` example for a complete demonstration.
//! **RadioParadiseStreamSource**:
//! - Downloads and decodes FLAC blocks in real-time
//! - Automatically detects bit depth (16/24/32-bit)
//! - Inserts track boundaries with metadata
//! - Integrates seamlessly with pmoaudio pipelines
//!
//! ## Cargo Features
//!
//! - `default = ["metadata-only"]`: Standard metadata and streaming (no FLAC decoding)
//! - `default`: Standard metadata and streaming (no FLAC decoding)
//! - `per-track`: Enable FLAC decoding and per-track extraction (adds `claxon`, `hound`, `tempfile`)
//! - `pmoserver`: Enable REST API extension for pmoserver integration (adds `utoipa`, `axum`)
//! - `server`: Enable server-side features (cache registry integration)
//! - `cache`: Enable cover and audio caching support (adds `pmocovers`, `pmoaudiocache`)
//! - `pmoaudio`: Enable RadioParadiseStreamSource for pmoaudio integration
//! - `pmoconfig`: Enable configuration integration with pmoconfig
//! - `server`: Enable RadioParadiseSource stub for backward compatibility (deprecated)
//!
//! ## See Also
//!
@@ -222,7 +214,6 @@ pub mod channels;
pub mod client;
pub mod error;
pub mod models;
pub mod paradise;
pub mod source;
pub mod stream;
pub mod streaming;

428
pmoparadise/src/paradise/channel.rs
View File

@@ -1,428 +0,0 @@
//! Channel orchestration primitives.
//!
//! This module wires together configuration, playlists, workers and client
//! tracking for a single Radio Paradise channel. The implementation is still
//! a scaffolding of the final behaviour; commands sent to the worker are
//! logged but not yet executing the full download/buffering pipeline.
use super::history::HistoryBackend;
use super::playlist::{PlaylistEntry, SharedPlaylist};
use super::worker::{ParadiseWorker, WorkerCommand};
use crate::client::RadioParadiseClient;
use anyhow::{Context, Result};
use async_stream::try_stream;
use bytes::Bytes;
use futures::{stream::BoxStream, StreamExt};
use pmosource::SourceCacheManager;
use std::fmt;
use std::str::FromStr;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::fs::File;
use tokio::sync::{mpsc, Mutex};
use tokio_util::io::ReaderStream;
use tracing::warn;
/// Logical identifier for a Radio Paradise channel.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ParadiseChannelKind {
Main,
Mellow,
Rock,
Eclectic,
}
impl ParadiseChannelKind {
pub const fn id(self) -> u8 {
match self {
Self::Main => 0,
Self::Mellow => 1,
Self::Rock => 2,
Self::Eclectic => 3,
}
}
pub const fn slug(self) -> &'static str {
match self {
Self::Main => "main",
Self::Mellow => "mellow",
Self::Rock => "rock",
Self::Eclectic => "eclectic",
}
}
pub const fn display_name(self) -> &'static str {
match self {
Self::Main => "Main Mix",
Self::Mellow => "Mellow Mix",
Self::Rock => "Rock Mix",
Self::Eclectic => "Eclectic Mix",
}
}
pub const fn description(self) -> &'static str {
match self {
Self::Main => "Eclectic mix of rock, world, electronica, and more",
Self::Mellow => "Mellower, less aggressive music",
Self::Rock => "Heavier, more guitar-driven music",
Self::Eclectic => "Curated worldwide selection",
}
}
}
impl FromStr for ParadiseChannelKind {
type Err = anyhow::Error;
fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
match s.to_ascii_lowercase().as_str() {
"main" | "0" => Ok(Self::Main),
"mellow" | "1" => Ok(Self::Mellow),
"rock" | "2" => Ok(Self::Rock),
"eclectic" | "3" => Ok(Self::Eclectic),
other => Err(anyhow::anyhow!("Unknown Radio Paradise channel: {}", other)),
}
}
}
/// Metadata descriptor for a channel.
#[derive(Debug, Clone, Copy)]
pub struct ChannelDescriptor {
pub kind: ParadiseChannelKind,
pub id: u8,
pub slug: &'static str,
pub display_name: &'static str,
pub description: &'static str,
}
impl ChannelDescriptor {
pub const fn new(kind: ParadiseChannelKind) -> Self {
Self {
id: kind.id(),
slug: kind.slug(),
display_name: kind.display_name(),
description: kind.description(),
kind,
}
}
}
pub const ALL_CHANNELS: [ChannelDescriptor; 4] = [
ChannelDescriptor::new(ParadiseChannelKind::Main),
ChannelDescriptor::new(ParadiseChannelKind::Mellow),
ChannelDescriptor::new(ParadiseChannelKind::Rock),
ChannelDescriptor::new(ParadiseChannelKind::Eclectic),
];
/// Returns the maximum valid channel ID
pub const fn max_channel_id() -> u8 {
(ALL_CHANNELS.len() - 1) as u8
}
/// Public handle to interact with a channel.
#[derive(Clone)]
pub struct ParadiseChannel {
inner: Arc<ParadiseChannelInner>,
}
struct ParadiseChannelInner {
descriptor: ChannelDescriptor,
client: RadioParadiseClient,
history_max_tracks: usize,
playlist: SharedPlaylist,
history: Arc<dyn HistoryBackend>,
cache_manager: Arc<SourceCacheManager>,
active_clients: AtomicUsize,
worker_tx: mpsc::Sender<WorkerCommand>,
worker: Mutex<Option<ParadiseWorker>>,
}
impl fmt::Debug for ParadiseChannel {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("ParadiseChannel")
.field("slug", &self.inner.descriptor.slug)
.field(
"active_clients",
&self.inner.active_clients.load(Ordering::SeqCst),
)
.finish()
}
}
impl ParadiseChannel {
#[allow(clippy::too_many_arguments)]
pub fn new(
descriptor: ChannelDescriptor,
base_client: RadioParadiseClient,
history_max_tracks: usize,
history: Arc<dyn HistoryBackend>,
cache_manager: Arc<SourceCacheManager>,
) -> Result<Self> {
let client = base_client.clone_with_channel(descriptor.id);
let playlist = SharedPlaylist::new(history_max_tracks);
let (worker, worker_tx) = ParadiseWorker::spawn(
descriptor,
client.clone(),
history_max_tracks,
playlist.clone(),
history.clone(),
cache_manager.clone(),
);
Ok(Self {
inner: Arc::new(ParadiseChannelInner {
descriptor,
client,
history_max_tracks,
playlist,
history,
cache_manager,
active_clients: AtomicUsize::new(0),
worker_tx,
worker: Mutex::new(Some(worker)),
}),
})
}
pub fn descriptor(&self) -> ChannelDescriptor {
self.inner.descriptor
}
pub fn playlist(&self) -> &SharedPlaylist {
&self.inner.playlist
}
pub fn history_max_tracks(&self) -> usize {
self.inner.history_max_tracks
}
pub fn history_backend(&self) -> &Arc<dyn HistoryBackend> {
&self.inner.history
}
pub fn cache_manager(&self) -> Arc<SourceCacheManager> {
self.inner.cache_manager.clone()
}
pub fn client(&self) -> &RadioParadiseClient {
&self.inner.client
}
pub fn active_client_count(&self) -> usize {
self.inner.active_clients.load(Ordering::SeqCst)
}
pub async fn connect_client(
&self,
client_id: impl Into<String>,
) -> Result<ParadiseClientStream> {
let client_id = client_id.into();
self.inner.active_clients.fetch_add(1, Ordering::SeqCst);
if let Err(err) = self
.inner
.worker_tx
.send(WorkerCommand::ClientConnected {
client_id: client_id.clone(),
})
.await
{
self.inner.active_clients.fetch_sub(1, Ordering::SeqCst);
return Err(anyhow::anyhow!("worker unavailable: {}", err));
}
self.inner.playlist.increment_all_pending().await;
self.ensure_started().await?;
Ok(ParadiseClientStream::new(self.clone(), client_id))
}
pub async fn disconnect_client(&self, client_id: impl Into<String>) -> Result<()> {
let client_id = client_id.into();
self.inner.active_clients.fetch_sub(1, Ordering::SeqCst);
self.inner
.worker_tx
.send(WorkerCommand::ClientDisconnected { client_id })
.await
.context("failed to notify worker of client disconnection")?;
Ok(())
}
pub async fn ensure_started(&self) -> Result<()> {
self.inner
.worker_tx
.send(WorkerCommand::EnsureReady)
.await
.context("failed to schedule worker warmup")
}
pub async fn shutdown(&self) -> Result<()> {
self.inner
.worker_tx
.send(WorkerCommand::Shutdown)
.await
.ok();
let mut guard = self.inner.worker.lock().await;
if let Some(worker) = guard.take() {
worker
.wait()
.await
.context("failed to join worker task")
.map(|_| ())
} else {
Ok(())
}
}
pub async fn mark_track_completed(&self, track: &Arc<PlaylistEntry>) {
let remaining = track.decrement_clients();
if remaining > 0 {
return;
}
if let Some(removed) = self
.inner
.playlist
.pop_front_matching(&track.track_id)
.await
{
if let Err(err) = self.inner.history.append(removed.as_history_entry()).await {
warn!(
channel = self.inner.descriptor.slug,
"Failed to persist history entry: {err:?}"
);
}
if let Err(err) = self
.inner
.history
.truncate(self.inner.history_max_tracks)
.await
{
warn!(
channel = self.inner.descriptor.slug,
"Failed to truncate history: {err:?}"
);
}
let history_entry = removed.as_history_entry();
self.inner.playlist.push_history_entry(history_entry).await;
}
}
}
/// Placeholder stream handle for per-client playback.
#[derive(Debug, Clone)]
pub struct ParadiseClientStream {
channel: ParadiseChannel,
client_id: String,
}
impl ParadiseClientStream {
fn new(channel: ParadiseChannel, client_id: String) -> Self {
Self { channel, client_id }
}
pub fn client_id(&self) -> &str {
&self.client_id
}
pub fn channel(&self) -> ParadiseChannel {
self.channel.clone()
}
pub fn into_byte_stream(self) -> BoxStream<'static, Result<Bytes, anyhow::Error>> {
let channel = self.channel.clone();
let client_id = self.client_id.clone();
let stream = try_stream! {
tracing::info!(
channel = channel.descriptor().slug,
client_id = %client_id,
"🎧 Client connecting to stream"
);
channel.ensure_started().await?;
let mut last_track_id: Option<String> = None;
loop {
let entries = channel.playlist().active_snapshot().await;
// Find the next track after last_track_id
let next_entry = if let Some(ref last_id) = last_track_id {
// Find the position of the last track we read
let last_pos = entries.iter().position(|e| e.track_id == *last_id);
// Get the next track (or wait if none available)
match last_pos {
Some(pos) if pos + 1 < entries.len() => {
Some(entries[pos + 1].clone())
}
_ => {
// Last track not found (was removed) or no next track available
// Wait for more tracks to be added
channel.ensure_started().await?;
let current_len = entries.len();
channel.playlist().wait_for_track_count(current_len).await;
continue;
}
}
} else {
// First track for this client
if entries.is_empty() {
channel.ensure_started().await?;
channel.playlist().wait_for_track_count(0).await;
continue;
}
Some(entries[0].clone())
};
let entry = next_entry.unwrap();
last_track_id = Some(entry.track_id.clone());
let audio_pk = entry
.audio_pk
.clone()
.ok_or_else(|| anyhow::anyhow!("Audio not cached yet"))?;
channel
.cache_manager()
.wait_audio_ready(&audio_pk)
.await
.map_err(|e| anyhow::anyhow!(e.to_string()))?;
let file_path = if let Some(path) = entry.file_path.clone() {
path
} else {
channel
.cache_manager()
.audio_file_path(&audio_pk)
.await
.ok_or_else(|| anyhow::anyhow!("Audio file path unavailable"))?
};
let file = File::open(&file_path).await?;
let mut reader = ReaderStream::new(file);
while let Some(chunk) = reader.next().await {
let bytes = chunk?;
yield bytes;
}
channel.mark_track_completed(&entry).await;
}
};
stream.boxed()
}
}
impl Drop for ParadiseClientStream {
fn drop(&mut self) {
let channel = self.channel.clone();
let client_id = self.client_id.clone();
let slug = channel.descriptor().slug;
tokio::spawn(async move {
if let Err(err) = channel.disconnect_client(client_id).await {
warn!(channel = slug, "Failed to disconnect client: {err:?}");
}
});
}
}

208
pmoparadise/src/paradise/constants.rs
View File

@@ -1,208 +0,0 @@
//! Constants for Radio Paradise orchestration layer.
//!
//! This module defines all the hardcoded parameters for the Radio Paradise
//! integration. These values are based on empirical testing and Radio Paradise's
//! infrastructure characteristics.
use std::time::Duration;
// ============================================================================
// Activity Lifecycle
// ============================================================================
/// Cooling timeout after all clients disconnect (seconds)
///
/// After the last client disconnects, the channel enters a "cooling" state
/// where it remains active for this duration before shutting down completely.
/// This avoids rapid start/stop cycles if clients reconnect quickly.
///
/// Value: 180 seconds (3 minutes) - good balance between responsiveness and stability
pub const COOLING_TIMEOUT_SECONDS: u64 = 180;
// ============================================================================
// Polling Intervals
// ============================================================================
/// High buffer polling interval (seconds)
///
/// When the playlist buffer has 3+ blocks, poll less frequently to reduce
/// API load and network usage.
///
/// Value: 120 seconds (2 minutes)
pub const POLLING_INTERVAL_HIGH_BUFFER: u64 = 120;
/// Medium buffer polling interval (seconds)
///
/// When the playlist buffer has 2 blocks, poll at moderate frequency.
///
/// Value: 60 seconds (1 minute)
pub const POLLING_INTERVAL_MEDIUM_BUFFER: u64 = 60;
/// Low buffer polling interval (seconds)
///
/// When the playlist buffer has less than 2 blocks, poll frequently to
/// ensure continuous playback.
///
/// Value: 20 seconds
pub const POLLING_INTERVAL_LOW_BUFFER: u64 = 20;
/// Helper to get high buffer polling interval as Duration
pub fn polling_high_interval() -> Duration {
Duration::from_secs(POLLING_INTERVAL_HIGH_BUFFER)
}
/// Helper to get medium buffer polling interval as Duration
pub fn polling_medium_interval() -> Duration {
Duration::from_secs(POLLING_INTERVAL_MEDIUM_BUFFER)
}
/// Helper to get low buffer polling interval as Duration
pub fn polling_low_interval() -> Duration {
Duration::from_secs(POLLING_INTERVAL_LOW_BUFFER)
}
// ============================================================================
// Polling Backoff (on API errors)
// ============================================================================
/// Initial backoff delay on API error (seconds)
///
/// When an API request fails, we wait this duration before retrying.
///
/// Value: 20 seconds
pub const BACKOFF_INITIAL_SECONDS: u64 = 20;
/// Maximum backoff delay (seconds)
///
/// Backoff is capped at this value to avoid waiting too long.
///
/// Value: 300 seconds (5 minutes)
pub const BACKOFF_MAX_SECONDS: u64 = 300;
/// Backoff multiplier
///
/// After each failure, the delay is multiplied by this factor.
/// Example: 20s → 40s → 80s → 160s → 300s (capped)
///
/// Value: 2.0 (exponential backoff)
pub const BACKOFF_MULTIPLIER: f32 = 2.0;
// ============================================================================
// Cache Tuning
// ============================================================================
/// Maximum number of blocks to remember in the worker
///
/// This prevents unbounded memory growth by limiting how many block event IDs
/// we track to avoid re-processing.
///
/// Calculation: (4 channels + 1 buffer) × 3 blocks per channel = 15 blocks
/// Each block is ~20 minutes of audio, so 15 blocks ≈ 5 hours of history
///
/// Value: 15 blocks
pub const MAX_BLOCKS_REMEMBERED: usize = 15;
/// Number of bytes to use for track ID hashing
///
/// Track IDs are constructed by hashing block content and track position.
/// This value defines how much of the FLAC data we read for hashing.
///
/// Value: 512 bytes - sufficient for unique identification without excessive I/O
pub const TRACK_ID_HASH_BYTES: usize = 512;
// ============================================================================
// History
// ============================================================================
/// Default maximum number of tracks to keep in history
///
/// This is used as the default if not configured via pmoconfig.
/// Users can override this value in their configuration.
///
/// Value: 100 tracks - represents ~5-8 hours of playback history
pub const HISTORY_DEFAULT_MAX_TRACKS: usize = 100;
// ============================================================================
// Streaming
// ============================================================================
/// Stream buffer size (bytes)
///
/// Buffer size for audio streaming. 64KB provides good balance between
/// latency and buffering efficiency.
///
/// Value: 64 KB
pub const STREAM_BUFFER_SIZE_BYTES: usize = 64 * 1024;
/// Enable gapless playback
///
/// Radio Paradise blocks are designed for gapless playback - each block
/// transitions seamlessly to the next without audio gaps.
///
/// Value: true (always enabled)
pub const STREAM_GAPLESS: bool = true;
// Note: Metadata format is always ICY (Icecast/SHOUTcast metadata)
// No enum or constant needed as it's the only supported format
// ============================================================================
// API Configuration
// ============================================================================
/// Radio Paradise API base URL
///
/// Base URL for all Radio Paradise API requests.
/// This is hardcoded as Radio Paradise's API endpoint doesn't change.
///
/// Value: https://api.radioparadise.com
pub const API_BASE_URL: &str = "https://api.radioparadise.com";
/// API request timeout (seconds)
///
/// Maximum time to wait for an API response before considering it failed.
///
/// Value: 30 seconds
pub const API_TIMEOUT_SECONDS: u64 = 30;
/// User agent for API requests
///
/// Identifies PMOMusic in HTTP requests to Radio Paradise's servers.
///
/// Value: PMO-RadioParadise/1.0
pub const API_USER_AGENT: &str = "PMO-RadioParadise/1.0";
/// Helper to get API timeout as Duration
pub fn api_timeout() -> Duration {
Duration::from_secs(API_TIMEOUT_SECONDS)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_duration_helpers() {
assert_eq!(polling_high_interval(), Duration::from_secs(120));
assert_eq!(polling_medium_interval(), Duration::from_secs(60));
assert_eq!(polling_low_interval(), Duration::from_secs(20));
assert_eq!(api_timeout(), Duration::from_secs(30));
}
#[test]
fn test_constants_sanity() {
// Polling intervals should be ordered
assert!(POLLING_INTERVAL_LOW_BUFFER < POLLING_INTERVAL_MEDIUM_BUFFER);
assert!(POLLING_INTERVAL_MEDIUM_BUFFER < POLLING_INTERVAL_HIGH_BUFFER);
// Backoff should be reasonable
assert!(BACKOFF_INITIAL_SECONDS < BACKOFF_MAX_SECONDS);
assert!(BACKOFF_MULTIPLIER > 1.0);
// Cache limits should be positive
assert!(MAX_BLOCKS_REMEMBERED > 0);
assert!(TRACK_ID_HASH_BYTES > 0);
// History should be reasonable
assert!(HISTORY_DEFAULT_MAX_TRACKS > 0);
}
}

217
pmoparadise/src/paradise/history.rs
View File

@@ -1,217 +0,0 @@
//! History persistence for Radio Paradise playback.
//!
//! The worker pushes every completed track into the history backend while
//! keeping the latest entries available for UPnP browsing. We use SQLite
//! for persistent storage with an abstract trait for testability.
use crate::models::Song;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::path::Path;
use std::sync::{Arc, Mutex as StdMutex};
use tokio::task::spawn_blocking;
/// Serializable record describing a played track.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HistoryEntry {
pub track_id: String,
pub channel_id: u8,
pub started_at: chrono::DateTime<chrono::Utc>,
pub duration_ms: u64,
pub song: SongSnapshot,
}
/// Minimal snapshot of a Radio Paradise song at playback time.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SongSnapshot {
pub title: String,
pub artist: String,
pub album: Option<String>,
pub cover_url: Option<String>,
}
impl SongSnapshot {
pub fn title(&self) -> &str {
&self.title
}
}
impl From<&Song> for SongSnapshot {
fn from(song: &Song) -> Self {
Self {
title: song.title.clone(),
artist: song.artist.clone(),
album: song.album.clone(),
cover_url: song.cover.clone(),
}
}
}
/// Abstract persistence interface.
#[async_trait]
pub trait HistoryBackend: Send + Sync {
async fn append(&self, entry: HistoryEntry) -> anyhow::Result<()>;
async fn recent(&self, limit: usize) -> anyhow::Result<Vec<HistoryEntry>>;
async fn len(&self) -> anyhow::Result<usize>;
async fn truncate(&self, keep: usize) -> anyhow::Result<()>;
}
pub struct SqliteHistoryBackend {
conn: Arc<StdMutex<rusqlite::Connection>>,
}
impl SqliteHistoryBackend {
pub fn new(path: impl AsRef<Path>) -> anyhow::Result<Self> {
let path = path.as_ref();
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let conn = rusqlite::Connection::open(path)?;
conn.pragma_update(None, "journal_mode", &"WAL")?;
conn.execute_batch(
"CREATE TABLE IF NOT EXISTS paradise_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
track_id TEXT NOT NULL,
channel_id INTEGER NOT NULL,
started_at_ms INTEGER NOT NULL,
duration_ms INTEGER NOT NULL,
title TEXT,
artist TEXT,
album TEXT,
cover_url TEXT
);
CREATE INDEX IF NOT EXISTS idx_history_started_at ON paradise_history(started_at_ms);",
)?;
Ok(Self {
conn: Arc::new(StdMutex::new(conn)),
})
}
fn conn(&self) -> Arc<StdMutex<rusqlite::Connection>> {
self.conn.clone()
}
}
#[async_trait]
impl HistoryBackend for SqliteHistoryBackend {
async fn append(&self, entry: HistoryEntry) -> anyhow::Result<()> {
let conn = self.conn();
spawn_blocking(move || -> anyhow::Result<()> {
let conn = conn.lock().unwrap();
conn.execute(
"INSERT INTO paradise_history (track_id, channel_id, started_at_ms, duration_ms, title, artist, album, cover_url)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
rusqlite::params![
entry.track_id,
entry.channel_id as i64,
entry.started_at.timestamp_millis(),
entry.duration_ms as i64,
entry.song.title,
entry.song.artist,
entry.song.album,
entry.song.cover_url,
],
)?;
Ok(())
})
.await??;
Ok(())
}
async fn recent(&self, limit: usize) -> anyhow::Result<Vec<HistoryEntry>> {
let conn = self.conn();
let limit = limit as i64;
spawn_blocking(move || -> anyhow::Result<Vec<HistoryEntry>> {
let conn = conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT track_id, channel_id, started_at_ms, duration_ms, title, artist, album, cover_url
FROM paradise_history
ORDER BY started_at_ms DESC
LIMIT ?1",
)?;
let mut rows = stmt.query([limit])?;
let mut entries = Vec::new();
while let Some(row) = rows.next()? {
let started_at_ms: i64 = row.get(2)?;
let started_at = DateTime::<Utc>::from_timestamp_millis(started_at_ms)
.ok_or_else(|| anyhow::anyhow!("Invalid timestamp in history"))?;
let entry = HistoryEntry {
track_id: row.get(0)?,
channel_id: row.get::<_, i64>(1)? as u8,
started_at,
duration_ms: row.get::<_, i64>(3)? as u64,
song: SongSnapshot {
title: row.get::<_, Option<String>>(4)?.unwrap_or_default(),
artist: row.get::<_, Option<String>>(5)?.unwrap_or_default(),
album: row.get(6)?,
cover_url: row.get(7)?,
},
};
entries.push(entry);
}
Ok(entries)
})
.await?
}
async fn len(&self) -> anyhow::Result<usize> {
let conn = self.conn();
let count = spawn_blocking(move || -> anyhow::Result<usize> {
let conn = conn.lock().unwrap();
let mut stmt = conn.prepare("SELECT COUNT(*) FROM paradise_history")?;
let count: i64 = stmt.query_row([], |row| row.get(0))?;
Ok(count as usize)
})
.await??;
Ok(count)
}
async fn truncate(&self, keep: usize) -> anyhow::Result<()> {
let conn = self.conn();
spawn_blocking(move || -> anyhow::Result<()> {
let conn = conn.lock().unwrap();
let count: i64 =
conn.query_row("SELECT COUNT(*) FROM paradise_history", [], |row| {
row.get(0)
})?;
let keep = keep as i64;
if count <= keep {
return Ok(());
}
let to_remove = count - keep;
conn.execute(
"DELETE FROM paradise_history
WHERE id IN (
SELECT id FROM paradise_history
ORDER BY started_at_ms ASC
LIMIT ?1
)",
rusqlite::params![to_remove],
)?;
Ok(())
})
.await??;
Ok(())
}
}
/// Creates a SQLite history backend with the given database path.
///
/// The database file and parent directories will be created if they don't exist.
///
/// # Arguments
///
/// * `database_path` - Path to the SQLite database file
///
/// # Example
///
/// ```rust,ignore
/// let backend = create_history_backend("/var/lib/pmo/history.db")?;
/// ```
pub fn create_history_backend(database_path: &str) -> anyhow::Result<Arc<dyn HistoryBackend>> {
let backend = SqliteHistoryBackend::new(database_path)?;
Ok(Arc::new(backend))
}

28
pmoparadise/src/paradise/mod.rs
View File

@@ -1,28 +0,0 @@
//! Internal orchestration layer for dynamic Radio Paradise streaming.
//!
//! This module implements the high level structures described in the
//! Radio Paradise functional specification:
//! - `ParadiseChannel`: lifecycle and state machine for a single RP channel.
//! - `ParadiseWorker`: async task responsible for polling/downloading blocks.
//! - `ParadiseClientStream`: per-client audio stream with independent cursor.
//! - Shared caches and history storage hooked into existing PMO components.
//!
//! The implementation is split across several submodules to keep concerns
//! isolated (constants, playlist management, history persistence, etc.).
//! The goal of this scaffolding is to provide a clear, testable surface for
//! the eventual end-to-end integration with the UPnP server and HTTP routes.
mod channel;
pub mod constants;
mod history;
mod playlist;
mod worker;
pub use channel::{
max_channel_id, ChannelDescriptor, ParadiseChannel, ParadiseChannelKind, ParadiseClientStream,
ALL_CHANNELS,
};
pub use constants::*; // Export all constants
pub use history::{create_history_backend, HistoryBackend, HistoryEntry};
pub use playlist::PlaylistEntry;
pub use worker::{load_rp_metadata, ParadiseWorker, RadioParadiseMetadata, WorkerCommand};

293
pmoparadise/src/paradise/playlist.rs
View File

@@ -1,293 +0,0 @@
//! Shared playlist structures for Radio Paradise channels.
//!
//! This module keeps track of the active queue and history for a Radio
//! Paradise channel. Each playlist entry knows how many clients still need
//! to consume it before the worker can evict it.
use super::history::{HistoryEntry, SongSnapshot};
use crate::models::Song;
use chrono::{DateTime, Utc};
use std::collections::VecDeque;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::SystemTime;
use tokio::sync::{Notify, RwLock};
/// Metadata stored for an active track.
#[derive(Debug)]
pub struct PlaylistEntry {
pub track_id: String,
pub channel_id: u8,
pub song: Arc<Song>,
pub started_at: DateTime<Utc>,
pub duration_ms: u64,
pub audio_pk: Option<String>,
pub file_path: Option<PathBuf>,
pending_clients: AtomicUsize,
}
impl PlaylistEntry {
#[allow(clippy::too_many_arguments)]
pub fn new(
track_id: String,
channel_id: u8,
song: Arc<Song>,
started_at: DateTime<Utc>,
duration_ms: u64,
audio_pk: Option<String>,
file_path: Option<PathBuf>,
pending_clients: usize,
) -> Self {
Self {
track_id,
channel_id,
song,
started_at,
duration_ms,
audio_pk,
file_path,
pending_clients: AtomicUsize::new(pending_clients),
}
}
pub fn as_history_entry(&self) -> HistoryEntry {
HistoryEntry {
track_id: self.track_id.clone(),
channel_id: self.channel_id,
started_at: self.started_at,
duration_ms: self.duration_ms,
song: SongSnapshot::from(self.song.as_ref()),
}
}
pub fn pending_clients(&self) -> usize {
self.pending_clients.load(Ordering::SeqCst)
}
pub fn set_pending_clients(&self, value: usize) {
self.pending_clients.store(value, Ordering::SeqCst);
}
pub fn increment_clients(&self) -> usize {
self.pending_clients.fetch_add(1, Ordering::SeqCst) + 1
}
pub fn decrement_clients(&self) -> usize {
let mut current = self.pending_clients.load(Ordering::SeqCst);
loop {
if current == 0 {
return 0;
}
match self.pending_clients.compare_exchange(
current,
current - 1,
Ordering::SeqCst,
Ordering::SeqCst,
) {
Ok(_) => return current - 1,
Err(actual) => current = actual,
}
}
}
}
#[derive(Default)]
struct PlaylistState {
active: VecDeque<Arc<PlaylistEntry>>,
history: VecDeque<HistoryEntry>,
max_history: usize,
}
impl PlaylistState {
fn new(max_history: usize) -> Self {
Self {
active: VecDeque::new(),
history: VecDeque::new(),
max_history,
}
}
fn active_len(&self) -> usize {
self.active.len()
}
fn push_active(&mut self, entry: Arc<PlaylistEntry>) {
self.active.push_back(entry);
}
fn active_snapshot(&self) -> Vec<Arc<PlaylistEntry>> {
self.active.iter().cloned().collect()
}
fn pop_front_if_ready(&mut self) -> Option<Arc<PlaylistEntry>> {
if let Some(front) = self.active.front() {
if front.pending_clients() == 0 {
return self.active.pop_front();
}
}
None
}
fn pop_front_matching(&mut self, track_id: &str) -> Option<Arc<PlaylistEntry>> {
if let Some(front) = self.active.front() {
if front.track_id == track_id && front.pending_clients() == 0 {
return self.active.pop_front();
}
}
None
}
fn push_history(&mut self, entry: HistoryEntry) {
self.history.push_back(entry);
self.trim_history();
}
fn recent_history(&self, limit: usize) -> Vec<HistoryEntry> {
let total = self.history.len();
let start = total.saturating_sub(limit);
self.history.iter().skip(start).cloned().collect()
}
fn trim_history(&mut self) {
while self.history.len() > self.max_history {
self.history.pop_front();
}
}
fn clear(&mut self) -> bool {
let changed = !self.active.is_empty() || !self.history.is_empty();
if changed {
self.active.clear();
self.history.clear();
}
changed
}
fn increment_all(&self) {
for entry in &self.active {
entry.increment_clients();
}
}
}
struct SharedPlaylistInner {
state: RwLock<PlaylistState>,
notify: Notify,
update_id: AtomicU32,
last_change: RwLock<Option<SystemTime>>,
}
#[derive(Clone)]
pub struct SharedPlaylist(Arc<SharedPlaylistInner>);
impl SharedPlaylist {
pub fn new(max_history: usize) -> Self {
Self(Arc::new(SharedPlaylistInner {
state: RwLock::new(PlaylistState::new(max_history)),
notify: Notify::new(),
update_id: AtomicU32::new(0),
last_change: RwLock::new(None),
}))
}
async fn touch(&self) {
self.0.update_id.fetch_add(1, Ordering::SeqCst);
let mut last_change = self.0.last_change.write().await;
*last_change = Some(SystemTime::now());
}
pub async fn push_active(&self, entry: Arc<PlaylistEntry>) {
let mut guard = self.0.state.write().await;
guard.push_active(entry);
drop(guard);
self.touch().await;
self.0.notify.notify_waiters();
}
pub async fn active_len(&self) -> usize {
let guard = self.0.state.read().await;
guard.active_len()
}
pub async fn active_snapshot(&self) -> Vec<Arc<PlaylistEntry>> {
let guard = self.0.state.read().await;
guard.active_snapshot()
}
pub async fn clear(&self) {
let mut guard = self.0.state.write().await;
let changed = guard.clear();
drop(guard);
if changed {
self.touch().await;
self.0.notify.notify_waiters();
}
}
pub async fn wait_for_track_count(&self, current_len: usize) {
loop {
let len = {
let guard = self.0.state.read().await;
guard.active_len()
};
if len > current_len {
break;
}
self.0.notify.notified().await;
}
}
pub async fn pop_front_if_ready(&self) -> Option<Arc<PlaylistEntry>> {
let mut guard = self.0.state.write().await;
let result = guard.pop_front_if_ready();
drop(guard);
if result.is_some() {
self.touch().await;
self.0.notify.notify_waiters();
}
result
}
pub async fn pop_front_matching(&self, track_id: &str) -> Option<Arc<PlaylistEntry>> {
let mut guard = self.0.state.write().await;
let result = guard.pop_front_matching(track_id);
drop(guard);
if result.is_some() {
self.touch().await;
self.0.notify.notify_waiters();
}
result
}
pub async fn push_history_entry(&self, entry: HistoryEntry) {
let mut guard = self.0.state.write().await;
guard.push_history(entry);
drop(guard);
self.touch().await;
}
pub async fn recent_history(&self, limit: usize) -> Vec<HistoryEntry> {
let guard = self.0.state.read().await;
guard.recent_history(limit)
}
pub async fn increment_all_pending(&self) {
let guard = self.0.state.read().await;
guard.increment_all();
}
pub fn update_id(&self) -> u32 {
self.0.update_id.load(Ordering::SeqCst)
}
pub async fn last_change(&self) -> Option<SystemTime> {
self.0.last_change.read().await.clone()
}
}

1145
pmoparadise/src/paradise/worker.rs
View File

File diff suppressed because it is too large Load Diff

666
pmoparadise/src/source.rs
View File

@@ -1,395 +1,114 @@
//! Music source implementation for Radio Paradise built on the new
//! `paradise` orchestration layer.
//! DEPRECATED: Stub implementation of RadioParadiseSource
//!
//! The source exposes a DIDL-Lite hierarchy compatible with UPnP
//! ContentDirectory while delegating block ingestion, caching and
//! multi-client streaming to [`ParadiseChannel`].
//! **⚠️ This module is deprecated and will be removed in a future version.**
//!
//! The orchestration-based RadioParadiseSource has been replaced by
//! `RadioParadiseStreamSource`, which integrates directly with the pmoaudio
//! pipeline for streaming and decoding.
//!
//! ## Migration Guide
//!
//! **Old approach** (deprecated):
//! ```rust,ignore
//! use pmoparadise::RadioParadiseSource;
//! let source = RadioParadiseSource::from_registry(client)?;
//! ```
//!
//! **New approach** (recommended):
//! ```rust,ignore
//! use pmoparadise::RadioParadiseStreamSource;
//! use pmoaudio::pipeline::Node;
//!
//! let stream_source = RadioParadiseStreamSource::new(client, None).await?;
//! let node = Node::from_logic(stream_source);
//! // Use node in pmoaudio pipeline
//! ```
//!
//! This stub implementation is provided only for backward compatibility with
//! existing code (e.g., pmomediaserver) until it can be updated to use
//! RadioParadiseStreamSource.
use crate::client::RadioParadiseClient;
use crate::paradise::{
create_history_backend, ChannelDescriptor, ParadiseChannel, PlaylistEntry, ALL_CHANNELS,
};
#[cfg(not(feature = "pmoconfig"))]
use crate::paradise::HISTORY_DEFAULT_MAX_TRACKS;
use anyhow::Result as AnyhowResult;
use pmoaudiocache::Cache as AudioCache;
use pmocovers::Cache as CoverCache;
use pmodidl::{Container, Item, Resource};
use pmosource::pmodidl;
use pmosource::{
async_trait, BrowseResult, CacheStatus, MusicSource, MusicSourceError, Result,
SourceCacheManager, SourceStatistics,
};
use std::collections::HashMap;
use std::sync::Arc;
use pmosource::pmodidl::{Container, Item};
use pmosource::{async_trait, BrowseResult, MusicSource, MusicSourceError, Result};
use std::time::SystemTime;
use tracing::warn;
/// Default image for Radio Paradise (300x300 WebP, embedded in binary)
/// Default Radio Paradise image (embedded in binary)
const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp");
fn channel_collection_id(channel_id: u8) -> String {
format!("radio-paradise:{}", channel_id)
}
fn channel_container_id(channel_id: u8) -> String {
format!("radio-paradise:channel:{}", channel_id)
}
fn parse_channel_container_id(object_id: &str) -> Option<u8> {
let mut parts = object_id.split(':');
match (parts.next(), parts.next(), parts.next(), parts.next()) {
(Some("radio-paradise"), Some("channel"), Some(id_str), None) => id_str.parse().ok(),
_ => None,
}
}
fn parse_track_channel(track_id: &str) -> Option<u8> {
let mut parts = track_id.split(':');
match (parts.next(), parts.next(), parts.next(), parts.next()) {
(Some("rp"), Some(channel_str), Some(_rest), None) => channel_str.parse().ok(),
_ => None,
}
}
fn format_duration(duration_seconds: u64) -> String {
let hours = duration_seconds / 3600;
let minutes = (duration_seconds % 3600) / 60;
let seconds = duration_seconds % 60;
format!("{hours}:{minutes:02}:{seconds:02}")
}
#[derive(Clone)]
/// DEPRECATED: Stub implementation of RadioParadiseSource
///
/// This is a minimal stub that implements the MusicSource trait with no-op
/// implementations. It exists only to maintain API compatibility during the
/// migration to RadioParadiseStreamSource.
///
/// **Do not use this in new code.** Use `RadioParadiseStreamSource` instead.
#[derive(Clone, Debug)]
pub struct RadioParadiseSource {
inner: Arc<RadioParadiseSourceInner>,
}
struct RadioParadiseSourceInner {
channels: HashMap<u8, Arc<ParadiseChannel>>,
}
impl std::fmt::Debug for RadioParadiseSource {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RadioParadiseSource").finish()
}
_client: RadioParadiseClient,
}
impl RadioParadiseSource {
/// DEPRECATED: Create a new RadioParadiseSource from registry
///
/// This method is deprecated and will always return an error indicating
/// that the orchestration-based source is no longer supported.
///
/// Use `RadioParadiseStreamSource` instead for audio streaming.
#[cfg(feature = "server")]
pub fn from_registry(client: RadioParadiseClient) -> Result<Self> {
// Load history configuration from pmoconfig using the config extension trait
#[cfg(feature = "pmoconfig")]
let (database_path, history_max_tracks) = {
use crate::config_ext::RadioParadiseConfigExt;
let cfg = pmoconfig::get_config();
let database_path = cfg.get_paradise_history_database().map_err(|e| {
MusicSourceError::SourceUnavailable(format!(
"Failed to get history database path: {}",
e
))
})?;
let max_tracks = cfg.get_paradise_history_size().map_err(|e| {
MusicSourceError::SourceUnavailable(format!("Failed to get history size: {}", e))
})?;
(database_path, max_tracks)
};
#[cfg(not(feature = "pmoconfig"))]
let (database_path, history_max_tracks) = {
use std::path::PathBuf;
let mut path = PathBuf::from(std::env::var("HOME").unwrap_or_else(|_| ".".to_string()));
path.push(".config");
path.push("pmo");
path.push("paradise");
std::fs::create_dir_all(&path).ok();
path.push("history.db");
(
path.to_string_lossy().to_string(),
HISTORY_DEFAULT_MAX_TRACKS,
)
};
let history_backend = create_history_backend(&database_path).map_err(|e| {
MusicSourceError::SourceUnavailable(format!(
"Failed to initialize history backend: {}",
e
))
})?;
let mut channels = HashMap::new();
for descriptor in ALL_CHANNELS.iter() {
let cache_manager = Arc::new(SourceCacheManager::from_registry(
channel_collection_id(descriptor.id),
)?);
let channel = Arc::new(
ParadiseChannel::new(
*descriptor,
client.clone(),
history_max_tracks,
history_backend.clone(),
cache_manager,
)
.map_err(|e| {
MusicSourceError::SourceUnavailable(format!(
"Failed to initialize channel {}: {e}",
descriptor.slug
))
})?,
);
channels.insert(descriptor.id, channel);
}
Ok(Self {
inner: Arc::new(RadioParadiseSourceInner { channels }),
})
pub fn from_registry(_client: RadioParadiseClient) -> Result<Self> {
Err(MusicSourceError::SourceUnavailable(
"RadioParadiseSource is deprecated. Use RadioParadiseStreamSource instead."
.to_string(),
))
}
/// DEPRECATED: Create a new RadioParadiseSource from registry with defaults
///
/// This method creates a stub instance that will log deprecation warnings
/// but allows existing code to compile.
///
/// Use `RadioParadiseStreamSource` instead for audio streaming.
#[cfg(feature = "server")]
pub fn from_registry_default(client: RadioParadiseClient) -> Result<Self> {
Self::from_registry(client)
pub fn from_registry_default(client: RadioParadiseClient) -> Self {
tracing::warn!(
"RadioParadiseSource::from_registry_default is deprecated. \
Use RadioParadiseStreamSource for audio streaming."
);
Self { _client: client }
}
pub fn new(
client: RadioParadiseClient,
cover_cache: Arc<CoverCache>,
audio_cache: Arc<AudioCache>,
) -> Self {
// Load history configuration from pmoconfig using the config extension trait
#[cfg(feature = "pmoconfig")]
let (database_path, history_max_tracks) = {
use crate::config_ext::RadioParadiseConfigExt;
let cfg = pmoconfig::get_config();
let database_path = cfg.get_paradise_history_database().unwrap_or_else(|e| {
panic!("Failed to get history database path: {e}");
});
let max_tracks = cfg.get_paradise_history_size().unwrap_or_else(|e| {
panic!("Failed to get history size: {e}");
});
(database_path, max_tracks)
};
#[cfg(not(feature = "pmoconfig"))]
let (database_path, history_max_tracks) = {
use std::path::PathBuf;
let mut path = PathBuf::from(std::env::var("HOME").unwrap_or_else(|_| ".".to_string()));
path.push(".config");
path.push("pmo");
path.push("paradise");
std::fs::create_dir_all(&path).ok();
path.push("history.db");
(
path.to_string_lossy().to_string(),
HISTORY_DEFAULT_MAX_TRACKS,
)
};
let history_backend: Arc<dyn crate::paradise::HistoryBackend> =
create_history_backend(&database_path).unwrap_or_else(|err| {
panic!("Failed to initialize history backend: {err}");
});
let mut channels = HashMap::new();
for descriptor in ALL_CHANNELS.iter() {
let cache_manager = Arc::new(SourceCacheManager::new(
channel_collection_id(descriptor.id),
Arc::clone(&cover_cache),
Arc::clone(&audio_cache),
));
match ParadiseChannel::new(
*descriptor,
client.clone(),
history_max_tracks,
history_backend.clone(),
cache_manager,
) {
Ok(channel) => {
channels.insert(descriptor.id, Arc::new(channel));
}
Err(err) => {
warn!(
channel = descriptor.slug,
"Failed to initialize channel: {err:?}"
);
}
}
}
Self {
inner: Arc::new(RadioParadiseSourceInner { channels }),
}
/// DEPRECATED: Create a new RadioParadiseSource with default settings
///
/// This method is deprecated and only exists for API compatibility.
pub fn new_default(client: RadioParadiseClient) -> Self {
tracing::warn!(
"RadioParadiseSource::new_default is deprecated. \
Use RadioParadiseStreamSource for audio streaming."
);
Self { _client: client }
}
pub fn new_default(
client: RadioParadiseClient,
cover_cache: Arc<CoverCache>,
audio_cache: Arc<AudioCache>,
) -> Self {
Self::new(client, cover_cache, audio_cache)
}
pub fn client_for_channel(&self, channel: u8) -> Option<RadioParadiseClient> {
self.inner
.channels
.get(&channel)
.map(|ch| ch.client().clone())
}
pub fn channel(&self, id: u8) -> Option<Arc<ParadiseChannel>> {
self.inner.channels.get(&id).cloned()
}
fn build_root_container(&self) -> Container {
Container {
id: "radio-paradise".to_string(),
parent_id: "0".to_string(),
restricted: Some("1".to_string()),
child_count: Some(ALL_CHANNELS.len().to_string()),
searchable: Some("1".to_string()),
title: "Radio Paradise".to_string(),
class: "object.container".to_string(),
containers: vec![],
items: vec![],
}
}
async fn build_channel_containers(&self) -> Vec<Container> {
let mut containers = Vec::new();
for descriptor in ALL_CHANNELS.iter() {
if let Some(channel) = self.channel(descriptor.id) {
let len = channel.playlist().active_len().await;
containers.push(Container {
id: channel_container_id(descriptor.id),
parent_id: "radio-paradise".to_string(),
restricted: Some("1".to_string()),
child_count: Some(len.to_string()),
searchable: Some("1".to_string()),
title: descriptor.display_name.to_string(),
class: "object.container.playlistContainer".to_string(),
containers: vec![],
items: vec![],
});
}
}
containers
}
async fn channel_items(
&self,
descriptor: ChannelDescriptor,
offset: usize,
limit: Option<usize>,
) -> Result<Vec<Item>> {
let channel = self
.channel(descriptor.id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(descriptor.slug.to_string()))?;
channel
.ensure_started()
.await
.map_err(|e| MusicSourceError::SourceUnavailable(e.to_string()))?;
let entries = channel.playlist().active_snapshot().await;
if entries.is_empty() || offset >= entries.len() {
return Ok(Vec::new());
}
let end = limit
.map(|count| offset + count)
.unwrap_or(entries.len())
.min(entries.len());
let parent_id = channel_container_id(descriptor.id);
let mut items = Vec::with_capacity(end - offset);
for entry in entries.into_iter().skip(offset).take(end - offset) {
match self.entry_to_item(channel.clone(), &parent_id, entry).await {
Ok(item) => items.push(item),
Err(err) => warn!(
channel = descriptor.slug,
"Failed to build DIDL item: {err:?}"
),
}
}
Ok(items)
}
async fn entry_to_item(
&self,
channel: Arc<ParadiseChannel>,
parent_id: &str,
entry: Arc<PlaylistEntry>,
) -> AnyhowResult<Item> {
let cache_manager = channel.cache_manager();
let metadata = cache_manager.get_metadata(&entry.track_id).await;
let resource_url = cache_manager
.resolve_uri(&entry.track_id)
.await
.or_else(|_| {
metadata
.as_ref()
.map(|meta| meta.original_uri.clone())
.ok_or_else(|| MusicSourceError::ObjectNotFound(entry.track_id.clone()))
})?;
let mut album_art = metadata
.as_ref()
.and_then(|meta| meta.cached_cover_pk.as_ref())
.and_then(|pk| cache_manager.cover_url(pk, None).ok());
if album_art.is_none() {
album_art = entry.song.cover.clone();
}
let duration_seconds = entry.duration_ms / 1000;
let duration_str = if duration_seconds > 0 {
Some(format_duration(duration_seconds as u64))
} else {
None
};
let resource = Resource {
protocol_info: "http-get:*:audio/flac:*".to_string(),
bits_per_sample: None,
sample_frequency: None,
nr_audio_channels: None,
duration: duration_str.clone(),
url: resource_url,
};
Ok(Item {
id: entry.track_id.clone(),
parent_id: parent_id.to_string(),
restricted: Some("1".to_string()),
title: entry.song.title.clone(),
creator: Some(entry.song.artist.clone()),
class: "object.item.audioItem.musicTrack".to_string(),
artist: Some(entry.song.artist.clone()),
album: entry.song.album.clone(),
genre: None,
album_art,
album_art_pk: None,
date: None,
original_track_number: None,
resources: vec![resource],
descriptions: vec![],
})
}
fn channels_iter(&self) -> impl Iterator<Item = (&u8, &Arc<ParadiseChannel>)> {
self.inner.channels.iter()
/// DEPRECATED: Create a new RadioParadiseSource with cache
///
/// This method is deprecated and only exists for API compatibility.
pub fn new_with_cache(client: RadioParadiseClient, _cache_size: usize) -> Self {
tracing::warn!(
"RadioParadiseSource::new_with_cache is deprecated. \
Use RadioParadiseStreamSource for audio streaming."
);
Self { _client: client }
}
}
#[async_trait]
impl MusicSource for RadioParadiseSource {
fn name(&self) -> &str {
"Radio Paradise"
"Radio Paradise (DEPRECATED)"
}
fn id(&self) -> &str {
"radio-paradise"
"radio-paradise-deprecated"
}
fn default_image(&self) -> &[u8] {
@@ -397,43 +116,32 @@ impl MusicSource for RadioParadiseSource {
}
async fn root_container(&self) -> Result<Container> {
Ok(self.build_root_container())
Ok(Container {
id: "radio-paradise-deprecated".to_string(),
parent_id: "0".to_string(),
restricted: Some("1".to_string()),
child_count: Some("0".to_string()),
searchable: Some("0".to_string()),
title: "Radio Paradise (DEPRECATED)".to_string(),
class: "object.container".to_string(),
containers: vec![],
items: vec![],
})
}
async fn browse(&self, object_id: &str) -> Result<BrowseResult> {
match object_id {
"0" => Ok(BrowseResult::Containers(vec![self.build_root_container()])),
"radio-paradise" => {
let containers = self.build_channel_containers().await;
Ok(BrowseResult::Containers(containers))
}
_ => {
if let Some(channel_id) = parse_channel_container_id(object_id) {
let descriptor = ALL_CHANNELS
.iter()
.find(|desc| desc.id == channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
let items = self.channel_items(*descriptor, 0, None).await?;
Ok(BrowseResult::Items(items))
} else {
Err(MusicSourceError::ObjectNotFound(object_id.to_string()))
}
}
}
async fn browse(&self, _object_id: &str) -> Result<BrowseResult> {
tracing::warn!("RadioParadiseSource::browse called but source is deprecated");
Ok(BrowseResult::Mixed {
containers: vec![],
items: vec![],
})
}
async fn resolve_uri(&self, object_id: &str) -> Result<String> {
let channel_id = parse_track_channel(object_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
let channel = self
.channel(channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
channel
.cache_manager()
.resolve_uri(object_id)
.await
.map_err(|e| MusicSourceError::CacheError(e.to_string()))
async fn resolve_uri(&self, _object_id: &str) -> Result<String> {
Err(MusicSourceError::SourceUnavailable(
"RadioParadiseSource is deprecated. Use RadioParadiseStreamSource instead."
.to_string(),
))
}
fn supports_fifo(&self) -> bool {
@@ -441,171 +149,25 @@ impl MusicSource for RadioParadiseSource {
}
async fn append_track(&self, _track: Item) -> Result<()> {
Err(MusicSourceError::FifoNotSupported)
Err(MusicSourceError::SourceUnavailable(
"RadioParadiseSource is deprecated and does not support FIFO operations."
.to_string(),
))
}
async fn remove_oldest(&self) -> Result<Option<Item>> {
Err(MusicSourceError::FifoNotSupported)
Ok(None)
}
async fn update_id(&self) -> u32 {
self.channels_iter()
.map(|(_, channel)| channel.playlist().update_id())
.max()
.unwrap_or(0)
0
}
async fn last_change(&self) -> Option<SystemTime> {
let mut latest: Option<SystemTime> = None;
for (_, channel) in self.channels_iter() {
if let Some(change) = channel.playlist().last_change().await {
latest = Some(match latest {
Some(current) if change <= current => current,
_ => change,
});
}
}
latest
None
}
async fn get_items(&self, offset: usize, count: usize) -> Result<Vec<Item>> {
let mut all = Vec::new();
for descriptor in ALL_CHANNELS.iter() {
let mut items = self.channel_items(*descriptor, 0, None).await?;
all.append(&mut items);
}
if offset >= all.len() {
return Ok(Vec::new());
}
let end = if count == 0 {
all.len()
} else {
(offset + count).min(all.len())
};
Ok(all.into_iter().skip(offset).take(end - offset).collect())
}
async fn get_available_formats(&self, _object_id: &str) -> Result<Vec<pmosource::AudioFormat>> {
Ok(vec![pmosource::AudioFormat {
format_id: "flac".to_string(),
mime_type: "audio/flac".to_string(),
sample_rate: Some(44100),
bit_depth: Some(16),
bitrate: None,
channels: Some(2),
}])
}
async fn get_cache_status(&self, object_id: &str) -> Result<CacheStatus> {
let channel_id = parse_track_channel(object_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
let channel = self
.channel(channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
channel
.cache_manager()
.get_cache_status(object_id)
.await
.map_err(|e| MusicSourceError::CacheError(e.to_string()))
}
async fn cache_item(&self, object_id: &str) -> Result<CacheStatus> {
let channel_id = parse_track_channel(object_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
let channel = self
.channel(channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
channel
.cache_manager()
.get_cache_status(object_id)
.await
.map_err(|e| MusicSourceError::CacheError(e.to_string()))
}
async fn browse_paginated(
&self,
object_id: &str,
offset: usize,
limit: usize,
) -> Result<BrowseResult> {
match object_id {
"0" => {
if offset == 0 {
Ok(BrowseResult::Containers(vec![self.build_root_container()]))
} else {
Ok(BrowseResult::Containers(Vec::new()))
}
}
"radio-paradise" => {
let containers = self.build_channel_containers().await;
let total = containers.len();
if offset >= total {
return Ok(BrowseResult::Containers(Vec::new()));
}
let end = if limit == 0 {
total
} else {
(offset + limit).min(total)
};
Ok(BrowseResult::Containers(
containers
.into_iter()
.skip(offset)
.take(end - offset)
.collect(),
))
}
_ => {
if let Some(channel_id) = parse_channel_container_id(object_id) {
let descriptor = ALL_CHANNELS
.iter()
.find(|desc| desc.id == channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
let items = self.channel_items(*descriptor, offset, Some(limit)).await?;
Ok(BrowseResult::Items(items))
} else {
Err(MusicSourceError::ObjectNotFound(object_id.to_string()))
}
}
}
}
async fn get_item_count(&self, object_id: &str) -> Result<usize> {
match object_id {
"0" => Ok(1),
"radio-paradise" => Ok(ALL_CHANNELS.len()),
_ => {
if let Some(channel_id) = parse_channel_container_id(object_id) {
let channel = self
.channel(channel_id)
.ok_or_else(|| MusicSourceError::ObjectNotFound(object_id.to_string()))?;
Ok(channel.playlist().active_len().await)
} else {
Err(MusicSourceError::ObjectNotFound(object_id.to_string()))
}
}
}
}
async fn statistics(&self) -> Result<SourceStatistics> {
let mut total_tracks = 0usize;
let mut cached_tracks = 0usize;
for (_, channel) in self.channels_iter() {
total_tracks += channel.playlist().active_len().await;
let stats = channel.cache_manager().statistics().await;
cached_tracks += stats.cached_tracks;
}
Ok(SourceStatistics {
total_items: Some(total_tracks),
total_containers: Some(ALL_CHANNELS.len() + 1),
cached_items: Some(cached_tracks),
cache_size_bytes: None,
})
async fn get_items(&self, _offset: usize, _count: usize) -> Result<Vec<Item>> {
Ok(vec![])
}
}