From 97a383c079204fd05ad50119689af9e1b550e0b5 Mon Sep 17 00:00:00 2001 From: Eric Coissac Date: Sat, 15 Nov 2025 15:43:42 +0100 Subject: [PATCH] Tentative de gestion d'un historique --- .DS_Store | Bin 14340 -> 14340 bytes .gitignore | 1 + Cargo.lock | 4 + PMOMusic/src/main.rs | 16 +- pmoapp/webapp/package-lock.json | 4 + .../src/components/RadioParadiseExplorer.vue | 1637 ++++++++++------- .../src/nodes/track_boundary_cover_node.rs | 1 + pmoaudio-ext/src/sinks/broadcast_pacing.rs | 6 +- pmoaudio-ext/src/sinks/flac_cache_sink.rs | 85 +- pmoaudio-ext/src/sinks/streaming_flac_sink.rs | 62 +- .../src/sinks/streaming_ogg_flac_sink.rs | 56 +- pmoaudio-ext/src/sinks/timed_broadcast.rs | 4 +- pmomediaserver/Cargo.toml | 14 +- pmomediaserver/src/lib.rs | 7 + pmomediaserver/src/paradise_streaming.rs | 280 +++ pmomediaserver/src/sources.rs | 13 +- pmomediaserver/src/sources_api.rs | 35 +- pmoparadise/Cargo.toml | 2 +- pmoparadise/examples/serve_channels.rs | 112 +- pmoparadise/src/lib.rs | 6 +- pmoparadise/src/models.rs | 21 + pmoparadise/src/pmoserver_ext.rs | 6 +- .../src/radio_paradise_stream_source.rs | 2 +- pmoparadise/src/source.rs | 416 ++++- pmoparadise/src/stream_channel.rs | 317 +++- 25 files changed, 2195 insertions(+), 912 deletions(-) create mode 100644 pmomediaserver/src/paradise_streaming.rs diff --git a/.DS_Store b/.DS_Store index 250fc4d04128e0cacb3c172170ed5008c87e6340..ad016f38d475ec81ec18cbb85852a532c2bc69a3 100644 GIT binary patch delta 310 zcmZoEXepTBFKW-gz`)GFAi%(o&ydWJ2*eo-sS__MPc|@NVPynLvP^c=5SbjW$us$* zxX8wfX6zFm@NQ=3VBug?*qp6t&p2_S;$|L!E}qGIWYQ)F$;NEF7R)Fv2h_;PpvO?a zkP9?7m!XuQm?4uPd9tCL%;f*#iQ*E~)kcC5J zEDJ8m%gN762b#+`xj;c`vbZ|mVOa+gakzudm&#$NN!$;DTDR2ve73u%t-gW|L_ygjtssbe{ngj)&F$3njY4bq-&eUk*Ep>737v| zk;~d5OE9^W6UME)j%|18*x?9=c^o!`95CL5*!mnVu9@U4<0~VNb&?cV36uyf5L{)) z4$_8xd$bL@^ah3;>9?dA?}I!^rDX=?|j=6Yr;_h(_f*bLj9NvfZ^x$rhy| z&*V81qNHk2kUG~SNcAwyDBq2PfNUSdLvB)Z=FvT)!M+%$eUWGJnq_gYMu^RKoF-PW z6O@2iIhh!izeYM`Dv>^Xi!7ZK*z;Xnr-xgt9Oe>=blG9A;xFSVe;bANG_iR*a-Q2# zx3_)YmA^FZ%HX~-(Mctl|JyRo{mb*y`Z&^#RV-s3V_0N0hB@YWECd};PTVUi+A2Q{ j>-F!wZp4^T+_wH3<@IU?{*@QLnn$jdx(mA(NlkqKGNd8$ diff --git a/.gitignore b/.gitignore index 2eec05c3..345d0102 100644 --- a/.gitignore +++ b/.gitignore @@ -39,3 +39,4 @@ upmpdcli/ test_upnp*.cargo/ .cargo/ setup-env.sh +/cache \ No newline at end of file diff --git a/Cargo.lock b/Cargo.lock index c07fcb80..dc62c18b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3211,11 +3211,14 @@ dependencies = [ name = "pmomediaserver" version = "0.1.0" dependencies = [ + "anyhow", "async-trait", "axum 0.8.7", "bevy_reflect", "once_cell", + "pmoaudiocache", "pmoconfig", + "pmocovers", "pmodidl", "pmoparadise", "pmoqobuz", @@ -3227,6 +3230,7 @@ dependencies = [ "serde_json", "thiserror 1.0.69", "tokio", + "tokio-util", "tracing", "utoipa", ] diff --git a/PMOMusic/src/main.rs b/PMOMusic/src/main.rs index f17b1220..35c2713b 100644 --- a/PMOMusic/src/main.rs +++ b/PMOMusic/src/main.rs @@ -1,6 +1,6 @@ use pmoapp::{WebAppExt, Webapp}; use pmomediarenderer::MEDIA_RENDERER; -use pmomediaserver::{MEDIA_SERVER, sources::SourcesExt}; +use pmomediaserver::{MEDIA_SERVER, sources::SourcesExt, ParadiseStreamingExt}; use pmoserver::Server; use pmosource::MusicSourceExt; use pmoupnp::UpnpServerExt; @@ -36,13 +36,19 @@ async fn main() -> Result<(), Box> { info!("🎵 Registering music sources..."); // // Enregistrer Qobuz - // if let Err(e) = server.register_qobuz().await { + // if let Err(e) = server.write().await.register_qobuz().await { // tracing::warn!("⚠️ Failed to register Qobuz: {}", e); // } - // Enregistrer Radio Paradise (inclut l'initialisation de l'API) - if let Err(e) = server.write().await.register_paradise().await { - tracing::warn!("⚠️ Failed to register Radio Paradise: {}", e); + // Initialiser les canaux de streaming Radio Paradise (pipelines + routes HTTP) + info!("📻 Initializing Radio Paradise streaming channels..."); + if let Err(e) = server.write().await.init_paradise_streaming().await { + tracing::warn!("⚠️ Failed to initialize Paradise streaming: {}", e); + } else { + // Enregistrer la source Radio Paradise UPnP (inclut l'initialisation de l'API) + if let Err(e) = server.write().await.register_paradise().await { + tracing::warn!("⚠️ Failed to register Radio Paradise source: {}", e); + } } // Lister toutes les sources enregistrées diff --git a/pmoapp/webapp/package-lock.json b/pmoapp/webapp/package-lock.json index e8814061..ffa3542e 100644 --- a/pmoapp/webapp/package-lock.json +++ b/pmoapp/webapp/package-lock.json @@ -1230,6 +1230,7 @@ "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==", "dev": true, "license": "MIT", + "peer": true, "engines": { "node": ">=12" }, @@ -1339,6 +1340,7 @@ "integrity": "sha512-p1diW6TqL9L07nNxvRMM7hMMw4c5XOo/1ibL4aAIGmSAt9slTE1Xgw5KWuof2uTOvCg9BY7ZRi+GaF+7sfgPeQ==", "devOptional": true, "license": "Apache-2.0", + "peer": true, "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" @@ -1353,6 +1355,7 @@ "integrity": "sha512-VbA8ScMvAISJNJVbRDTJdCwqQoAareR/wutevKanhR2/1EkoXVZVkkORaYm/tNVCjP/UDTKtcw3bAkwOUdedmA==", "dev": true, "license": "MIT", + "peer": true, "dependencies": { "esbuild": "^0.25.0", "fdir": "^6.5.0", @@ -1434,6 +1437,7 @@ "resolved": "https://registry.npmjs.org/vue/-/vue-3.5.22.tgz", "integrity": "sha512-toaZjQ3a/G/mYaLSbV+QsQhIdMo9x5rrqIpYRObsJ6T/J+RyCSFwN2LHNVH9v8uIcljDNa3QzPVdv3Y6b9hAJQ==", "license": "MIT", + "peer": true, "dependencies": { "@vue/compiler-dom": "3.5.22", "@vue/compiler-sfc": "3.5.22", diff --git a/pmoapp/webapp/src/components/RadioParadiseExplorer.vue b/pmoapp/webapp/src/components/RadioParadiseExplorer.vue index c52ed2cb..1168038b 100644 --- a/pmoapp/webapp/src/components/RadioParadiseExplorer.vue +++ b/pmoapp/webapp/src/components/RadioParadiseExplorer.vue @@ -16,7 +16,6 @@ + +
{{ currentFormatDescription }}
+ +
+ +
+ + +
+
+
+ + +
+ -
- ❌ {{ bitratesError }} + +
+
+

🔧 Stream Diagnostics

+ +
+
+
+
Current Stream URL
+
+ {{ currentStreamUrl }} +
+
+
+
Format
+
{{ selectedFormat.toUpperCase() }}
+
+
+
Mode
+
{{ playbackMode === 'live' ? 'Live Stream' : 'Historical Playback' }}
+
+
+
Channel Slug
+
{{ currentChannelSlug }}
+
+
+
+
+ {{ format.name }}: + + {{ getStreamUrlFor(format.id) }} + +
+
- -
- -

{{ audioError }}

- + +
+
+

📻 Live Stream Metadata

+ +
+
+ ❌ {{ streamMetadataError }} +
+
+ 
{{ JSON.stringify(streamMetadata, null, 2) }}
+
+
+ Click "Refresh Metadata" to fetch current stream metadata +
+
+ Last updated: {{ formatTimestamp(streamMetadataLastUpdated) }} +
+
+ + +
+

🧪 Quick Test Commands

+

Copy and paste these commands to test streams with external players

+
+
+
+ ffplay (FLAC) + +
+ {{ ffplayFlacCommand }} +
+
+
+ ffplay (OGG) + +
+ {{ ffplayOggCommand }} +
+
+
+ VLC + +
+ {{ vlcCommand }} +
+
+
+ curl (Download to file) + +
+ {{ curlCommand }} +
+
+
+ + +
+
+

🎵 Now Playing on Radio Paradise

+
+ + +
+
+ +
+ ❌ {{ audioError }} +
+ +
+ +
+
+ +
+
+ 🎵 +
+ +
+
+
{{ playerMetadata.title || 'Unknown Title' }}
+
{{ playerMetadata.artist || 'Unknown Artist' }}
+
{{ playerMetadata.album }}
+
{{ playerMetadata.year }}
+
+
+ Loading metadata... + No metadata available +
+
+
+ + +
+ +
+ + +
+
+ Channel: + {{ channels.find(c => c.id === selectedChannel)?.name || 'Unknown' }} +
+
+ Format: + {{ selectedFormat.toUpperCase() }} +
+
+ Mode: + {{ playbackMode === 'live' ? 'Live' : 'Historic' }} +
+
+ Updated: + {{ formatTimestamp(playerMetadataLastUpdated) }} +
+
+
@@ -166,225 +347,6 @@
-
-
-

📡 Channel Status

- -
-
- ❌ {{ channelStatusError }} -
-
-
-
-
Channel
-
{{ channelStatus.slug }}
-
-
-
Active Clients
-
{{ channelStatus.active_clients }}
-
-
-
Queue Length
-
{{ channelStatus.queue_length }}
-
-
-
Update ID
-
{{ channelStatus.update_id }}
-
-
-
Last Change
-
- {{ channelStatus.last_change ? formatTimestamp(new Date(channelStatus.last_change)) : '—' }} -
-
-
-
History Entries
-
- {{ channelStatus.history_entries }} / {{ channelStatus.history_max_tracks }} -
-
-
-
Cache Collection
-
{{ channelStatus.cache_collection_id }}
-
-
-
Cache Tracks
-
- {{ channelStatus.cache_cached_tracks }} / {{ channelStatus.cache_total_tracks }} -
-
-
-
- Last refresh: {{ formatTimestamp(channelStatusLastUpdated) }} -
-
-
- -
-
-

🎧 Live Playlist

-
- - -
-
-
- Queue length: - {{ channelPlaylist.queue_length }} - Update ID: {{ channelPlaylist.update_id }} - - Last refresh: {{ formatTimestamp(channelPlaylistLastUpdated) }} - -
-
- ❌ {{ channelPlaylistError }} -
-
- ⏳ Loading playlist… -
-
-
-
-
- - {{ cacheStatusLabel(item.cache_status) }} - - Pending listeners: {{ item.pending_clients }} -
-
-
{{ item.title }}
-
{{ item.artist || 'Unknown artist' }}
-
-
- {{ item.album }} - ⏱ {{ formatDuration(item.duration_ms) }} - ▶️ @{{ formatDuration(item.elapsed_ms) }} - 🕒 {{ formatTimestamp(new Date(item.started_at)) }} - 💾 {{ formatBytes(item.cache_status.size_bytes) }} -
-
- - - - - - Open resolved URI - -
-
- ✅ {{ trackExtrasFor(item).cacheRequestMessage }} -
-
- ❌ {{ trackExtrasFor(item).cacheError }} -
-
- ❌ Resolve error: {{ trackExtrasFor(item).resolveError }} -
-
- ❌ Formats error: {{ trackExtrasFor(item).formatsError }} -
-
-
- {{ format.format_id }} - {{ format.mime_type }} - {{ format.sample_rate }} Hz - {{ format.bit_depth }} bit - {{ format.bitrate }} kbps - {{ format.channels }} ch -
-
-
-
-
- No tracks currently queued. Try refreshing after playback starts. -
-
-
- -
-
-

🕰️ Recent History

- -
-
- Last refresh: {{ formatTimestamp(channelHistoryLastUpdated) }} -
-
- ❌ {{ channelHistoryError }} -
-
- ⏳ Loading history… -
-
-
- No history entries yet. -
-
-
{{ entry.title }}
-
- {{ entry.artist }} - • {{ entry.album }} - • {{ formatDuration(entry.duration_ms) }} - • {{ formatTimestamp(new Date(entry.started_at)) }} -
-
-
-
-

⏭️ Next Block Preview

@@ -500,24 +462,99 @@ @@ -1951,31 +1695,540 @@ onUnmounted(() => { flex-wrap: wrap; } -.audio-player-container { - margin: 12px 0 24px; - padding: 16px; - border-radius: 8px; - background: rgba(0, 0, 0, 0.2); - border: 1px solid #333; +/* Enhanced Media Player Section */ +.enhanced-player-section { + background: linear-gradient(135deg, #1a1a2e 0%, #0f0f1e 100%); + border-radius: 12px; + padding: 24px; + margin: 20px 0 32px; + border: 2px solid rgba(0, 212, 255, 0.4); + box-shadow: 0 8px 24px rgba(0, 0, 0, 0.4); +} + +.player-header { display: flex; - gap: 12px; + justify-content: space-between; + align-items: center; + margin-bottom: 20px; + padding-bottom: 16px; + border-bottom: 1px solid rgba(0, 212, 255, 0.2); +} + +.player-header h3 { + margin: 0; + color: #00d4ff; + font-size: 1.3rem; +} + +.player-controls-header { + display: flex; + gap: 10px; align-items: center; } -.audio-error { - margin: 0; - color: #ff6b6b; - flex: 1; +.btn-refresh-meta { + background: rgba(0, 212, 255, 0.15); + border: 1px solid rgba(0, 212, 255, 0.4); + color: #00d4ff; + padding: 8px 12px; + border-radius: 6px; + cursor: pointer; + font-size: 1.1rem; + transition: all 0.2s; + min-width: 40px; + display: flex; + align-items: center; + justify-content: center; +} + +.btn-refresh-meta:hover:not(:disabled) { + background: rgba(0, 212, 255, 0.25); + border-color: rgba(0, 212, 255, 0.6); +} + +.btn-refresh-meta:disabled { + opacity: 0.5; + cursor: not-allowed; } .btn-stop { padding: 8px 16px; - border-radius: 4px; - border: none; + border-radius: 6px; + border: 1px solid rgba(231, 76, 60, 0.4); cursor: pointer; - background: rgba(231, 76, 60, 0.2); + background: rgba(231, 76, 60, 0.15); color: #e74c3c; font-weight: bold; + transition: all 0.2s; +} + +.btn-stop:hover { + background: rgba(231, 76, 60, 0.25); + border-color: rgba(231, 76, 60, 0.6); +} + +.audio-error-banner { + background: rgba(255, 68, 68, 0.15); + border: 1px solid rgba(255, 68, 68, 0.4); + color: #ff6b6b; + padding: 16px; + border-radius: 8px; + margin-bottom: 16px; + font-weight: 500; +} + +.player-content { + display: flex; + flex-direction: column; + gap: 20px; +} + +.player-info { + display: flex; + gap: 24px; + align-items: flex-start; +} + +.player-cover { + flex-shrink: 0; +} + +.player-cover img { + width: 180px; + height: 180px; + object-fit: cover; + border-radius: 12px; + box-shadow: 0 8px 20px rgba(0, 0, 0, 0.6); + border: 2px solid rgba(0, 212, 255, 0.2); +} + +.player-cover-placeholder { + width: 180px; + height: 180px; + display: flex; + align-items: center; + justify-content: center; + background: linear-gradient(135deg, #2a2a3e 0%, #1a1a2e 100%); + border-radius: 12px; + border: 2px dashed rgba(0, 212, 255, 0.3); + font-size: 4rem; + color: rgba(0, 212, 255, 0.3); +} + +.player-metadata { + flex: 1; + display: flex; + flex-direction: column; + justify-content: center; + gap: 8px; + min-width: 0; +} + +.metadata-display { + display: flex; + flex-direction: column; + gap: 8px; +} + +.player-title { + font-size: 1.8rem; + font-weight: bold; + color: #ffffff; + line-height: 1.2; + word-wrap: break-word; +} + +.player-artist { + font-size: 1.4rem; + color: #00d4ff; + font-weight: 600; + line-height: 1.3; +} + +.player-album { + font-size: 1.1rem; + color: #9aa0a6; + font-style: italic; +} + +.player-year { + font-size: 0.95rem; + color: #666; + background: rgba(255, 255, 255, 0.05); + padding: 4px 10px; + border-radius: 4px; + display: inline-block; + align-self: flex-start; + margin-top: 4px; +} + +.metadata-loading { + color: #9aa0a6; + font-size: 1rem; + font-style: italic; +} + +.player-audio-controls { + width: 100%; +} + +.player-audio-controls audio { + width: 100%; + border-radius: 8px; + background: #0a0a0a; + outline: none; +} + +.player-stream-info { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); + gap: 12px; + padding: 16px; + background: rgba(0, 0, 0, 0.3); + border-radius: 8px; + border: 1px solid rgba(0, 212, 255, 0.15); +} + +.stream-info-item { + display: flex; + flex-direction: column; + gap: 4px; +} + +.info-label { + font-size: 0.75rem; + text-transform: uppercase; + letter-spacing: 0.05em; + color: #9aa0a6; + font-weight: 600; +} + +.info-value { + font-size: 1rem; + color: #f5f5f5; + font-weight: 500; +} + +/* Stream Controls Section */ +.stream-controls-section { + background: linear-gradient(135deg, #1a1a2e 0%, #16213e 100%); + border-radius: 8px; + padding: 20px; + margin-bottom: 20px; + border: 1px solid rgba(0, 212, 255, 0.3); +} + +.stream-controls-section h3 { + margin-top: 0; + color: #00d4ff; + margin-bottom: 16px; +} + +.stream-controls-grid { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(280px, 1fr)); + gap: 20px; +} + +.control-group { + display: flex; + flex-direction: column; + gap: 10px; +} + +.control-group label { + font-weight: 600; + color: #9aa0a6; + font-size: 0.9rem; + text-transform: uppercase; + letter-spacing: 0.05em; +} + +.format-chips, +.mode-chips { + display: flex; + gap: 10px; + flex-wrap: wrap; +} + +.format-chip, +.mode-chip { + padding: 10px 18px; + border-radius: 6px; + border: 2px solid rgba(0, 212, 255, 0.3); + background: rgba(0, 212, 255, 0.08); + color: #00d4ff; + cursor: pointer; + font-weight: 600; + transition: all 0.2s; + font-size: 0.9rem; +} + +.format-chip:hover, +.mode-chip:hover { + background: rgba(0, 212, 255, 0.15); + border-color: rgba(0, 212, 255, 0.5); + transform: translateY(-2px); +} + +.format-chip.active, +.mode-chip.active { + background: rgba(0, 212, 255, 0.25); + border-color: #00d4ff; + box-shadow: 0 0 12px rgba(0, 212, 255, 0.3); +} + +.format-description { + padding: 8px 12px; + background: rgba(0, 0, 0, 0.3); + border-radius: 4px; + font-size: 0.85rem; + color: #9aa0a6; + font-style: italic; +} + +.client-id-input { + padding: 10px 12px; + border-radius: 6px; + border: 1px solid #333; + background: #0f0f0f; + color: #fff; + font-family: 'Courier New', monospace; + font-size: 0.9rem; +} + +.client-id-input:focus { + outline: none; + border-color: #00d4ff; + box-shadow: 0 0 8px rgba(0, 212, 255, 0.2); +} + +/* Stream Diagnostics Section */ +.stream-diagnostics-section { + background: #1a1a1a; + border-radius: 8px; + padding: 20px; + margin-bottom: 20px; + border: 1px solid #333; +} + +.stream-diagnostics-section h3 { + margin-top: 0; + color: #00d4ff; +} + +/* Stream Metadata Section */ +.stream-metadata-section { + background: #1a1a1a; + border-radius: 8px; + padding: 20px; + margin-bottom: 20px; + border: 1px solid #333; +} + +.stream-metadata-section h3 { + margin-top: 0; + color: #00d4ff; +} + +.metadata-content { + margin-top: 12px; +} + +.metadata-json { + background: #0f0f0f; + border: 1px solid #333; + border-radius: 6px; + padding: 16px; + overflow-x: auto; + font-family: 'Courier New', monospace; + font-size: 0.85rem; + color: #00d4ff; + line-height: 1.5; + max-height: 400px; + overflow-y: auto; +} + +.metadata-json::-webkit-scrollbar { + width: 8px; + height: 8px; +} + +.metadata-json::-webkit-scrollbar-track { + background: #1a1a1a; + border-radius: 4px; +} + +.metadata-json::-webkit-scrollbar-thumb { + background: #333; + border-radius: 4px; +} + +.metadata-json::-webkit-scrollbar-thumb:hover { + background: #555; +} + +/* Quick Test Section */ +.quick-test-section { + background: linear-gradient(135deg, #2a1a1a 0%, #1a1a1a 100%); + border-radius: 8px; + padding: 20px; + margin-bottom: 20px; + border: 1px solid rgba(255, 193, 7, 0.3); +} + +.quick-test-section h3 { + margin-top: 0; + color: #ffc107; + margin-bottom: 8px; +} + +.section-description { + color: #9aa0a6; + font-size: 0.9rem; + margin-bottom: 16px; +} + +.test-commands { + display: flex; + flex-direction: column; + gap: 16px; +} + +.command-group { + background: rgba(0, 0, 0, 0.3); + border: 1px solid rgba(255, 193, 7, 0.2); + border-radius: 6px; + padding: 12px; +} + +.command-label { + display: flex; + justify-content: space-between; + align-items: center; + margin-bottom: 8px; +} + +.command-label strong { + color: #ffc107; + font-size: 0.9rem; +} + +.btn-copy-small { + background: rgba(255, 193, 7, 0.15); + border: 1px solid rgba(255, 193, 7, 0.3); + color: #ffc107; + padding: 4px 10px; + border-radius: 4px; + cursor: pointer; + font-size: 0.85rem; + transition: all 0.2s; +} + +.btn-copy-small:hover { + background: rgba(255, 193, 7, 0.25); + border-color: rgba(255, 193, 7, 0.5); +} + +.command-text { + display: block; + background: #0a0a0a; + padding: 10px 12px; + border-radius: 4px; + font-family: 'Courier New', monospace; + font-size: 0.85rem; + color: #00d4ff; + overflow-x: auto; + white-space: nowrap; + border: 1px solid #222; +} + +.command-text::-webkit-scrollbar { + height: 6px; +} + +.command-text::-webkit-scrollbar-track { + background: #111; + border-radius: 3px; +} + +.command-text::-webkit-scrollbar-thumb { + background: #333; + border-radius: 3px; +} + +.command-text::-webkit-scrollbar-thumb:hover { + background: #555; +} + +.diagnostics-grid { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); + gap: 12px; + margin-bottom: 20px; +} + +.diagnostic-card { + background: rgba(0, 212, 255, 0.05); + border: 1px solid rgba(0, 212, 255, 0.15); + border-radius: 6px; + padding: 12px; + display: flex; + flex-direction: column; + gap: 6px; +} + +.diagnostic-label { + font-size: 0.75rem; + text-transform: uppercase; + letter-spacing: 0.05em; + color: #9aa0a6; + font-weight: 600; +} + +.diagnostic-value { + font-size: 1rem; + font-weight: 600; + color: #f5f5f5; +} + +.diagnostic-value code { + background: rgba(0, 0, 0, 0.4); + padding: 4px 8px; + border-radius: 4px; + font-family: 'Courier New', monospace; + font-size: 0.85rem; + color: #00d4ff; + word-break: break-all; + display: block; + margin-top: 4px; +} + +.stream-urls-list { + display: flex; + flex-direction: column; + gap: 10px; + background: rgba(0, 0, 0, 0.3); + padding: 16px; + border-radius: 6px; +} + +.url-item { + display: flex; + gap: 10px; + align-items: center; + flex-wrap: wrap; + padding: 8px; + border-bottom: 1px solid rgba(255, 255, 255, 0.05); +} + +.url-item:last-child { + border-bottom: none; +} + +.url-item strong { + color: #00d4ff; + min-width: 120px; } @media (max-width: 768px) { @@ -2005,5 +2258,53 @@ onUnmounted(() => { flex-direction: column; align-items: stretch; } + + .stream-controls-grid { + grid-template-columns: 1fr; + } + + .diagnostics-grid { + grid-template-columns: 1fr; + } + + /* Enhanced Player Responsive */ + .player-header { + flex-direction: column; + align-items: flex-start; + gap: 12px; + } + + .player-controls-header { + width: 100%; + justify-content: flex-end; + } + + .player-info { + flex-direction: column; + align-items: center; + text-align: center; + } + + .player-cover img, + .player-cover-placeholder { + width: 150px; + height: 150px; + } + + .player-title { + font-size: 1.4rem; + } + + .player-artist { + font-size: 1.1rem; + } + + .player-album { + font-size: 1rem; + } + + .player-stream-info { + grid-template-columns: 1fr; + } } diff --git a/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs b/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs index c0303389..b157101f 100644 --- a/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs +++ b/pmoaudio-ext/src/nodes/track_boundary_cover_node.rs @@ -35,6 +35,7 @@ impl TrackBoundaryCoverNode { } } +#[async_trait::async_trait] impl AudioPipelineNode for TrackBoundaryCoverNode { fn get_tx(&self) -> Option>> { self.inner.get_tx() diff --git a/pmoaudio-ext/src/sinks/broadcast_pacing.rs b/pmoaudio-ext/src/sinks/broadcast_pacing.rs index 685f15f7..4d8f9883 100644 --- a/pmoaudio-ext/src/sinks/broadcast_pacing.rs +++ b/pmoaudio-ext/src/sinks/broadcast_pacing.rs @@ -6,7 +6,7 @@ //! - Paces broadcast to match audio playback rate use std::time::Instant; -use tracing::{debug, info, warn}; +use tracing::{trace, warn}; /// Error returned when a frame should be skipped (too late) #[derive(Debug)] @@ -56,7 +56,7 @@ impl BroadcastPacer { let elapsed_since_start = self.start_time.elapsed().as_secs_f64(); if audio_timestamp < 0.1 && elapsed_since_start > 1.0 { self.start_time = Instant::now(); - info!( + trace!( "{} broadcaster: TopZeroSync detected, resetting timer", self.label ); @@ -94,7 +94,7 @@ impl BroadcastPacer { // Log pour info si on est très en avance, mais on ne dort PAS if self.max_lead_time > 0.0 && lead_time > self.max_lead_time { - debug!( + trace!( "{} broadcaster: lead_time={:.3}s > max={:.3}s (audio_ts={:.3}s, elapsed={:.3}s) - relying on natural backpressure", self.label, lead_time, self.max_lead_time, audio_timestamp, elapsed ); diff --git a/pmoaudio-ext/src/sinks/flac_cache_sink.rs b/pmoaudio-ext/src/sinks/flac_cache_sink.rs index dba70812..43f0cf2f 100755 --- a/pmoaudio-ext/src/sinks/flac_cache_sink.rs +++ b/pmoaudio-ext/src/sinks/flac_cache_sink.rs @@ -200,6 +200,7 @@ impl NodeLogic for FlacCacheSinkLogic { // Phase 1: Dispatcher jusqu'à ce que le prebuffer soit terminé let mut end_of_stream_received = false; + let mut early_track_boundary_received = false; let mut track_tx_opt = Some(track_tx); let pk = loop { tokio::select! { @@ -221,9 +222,9 @@ impl NodeLogic for FlacCacheSinkLogic { result = rx.recv() => { match result { Some(segment) => { - // Si EndOfStream a été reçu, ignorer tous les segments suivants + // Si EndOfStream ou TrackBoundary a été reçu, ignorer tous les segments suivants // et continuer à attendre cache_future - if end_of_stream_received { + if end_of_stream_received || early_track_boundary_received { continue; } @@ -239,10 +240,18 @@ impl NodeLogic for FlacCacheSinkLogic { } } _AudioSegment::Sync(marker) => match &**marker { - SyncMarker::TrackBoundary { .. } => { - // TrackBoundary avant fin du prebuffer - track trop courte - tracing::error!("FlacCacheSink: TrackBoundary received before prebuffer complete - track too short"); - return Err(AudioError::ProcessingError("Track too short for prebuffer".to_string())); + SyncMarker::TrackBoundary { metadata } => { + // TrackBoundary pendant le prebuffer - track courte (< 512KB) + tracing::warn!( + "FlacCacheSink: TrackBoundary received before prebuffer complete - track shorter than 512KB, closing pump and waiting for ingestion" + ); + // Stocker les métadonnées pour la prochaine track + next_track_metadata = Some(metadata.clone()); + // Fermer le track_tx pour que le pump se termine proprement + track_tx_opt = None; + // Marquer qu'on a reçu un TrackBoundary précoce + early_track_boundary_received = true; + // Continuer à attendre cache_future } SyncMarker::EndOfStream => { tracing::debug!("FlacCacheSink: EndOfStream during prebuffer - closing pump and waiting for ingestion to complete"); @@ -262,7 +271,7 @@ impl NodeLogic for FlacCacheSinkLogic { } None => { // EOF sur rx pendant le prebuffer - attendre que cache_future se termine - if !end_of_stream_received { + if !end_of_stream_received && !early_track_boundary_received { tracing::debug!("FlacCacheSink: EOF on rx during prebuffer, waiting for ingestion to complete"); track_tx_opt = None; end_of_stream_received = true; @@ -296,26 +305,62 @@ impl NodeLogic for FlacCacheSinkLogic { )) })?; - let url = match dest_metadata.read().await.get_cover_url().await { - Ok(url) => { - tracing::debug!("FlacCacheSink: Got cover URL for pk {}: {:?}", pk, url); - url + let cover_pk_present = match dest_metadata.read().await.get_cover_pk().await { + Ok(Some(existing_pk)) => { + tracing::debug!( + "FlacCacheSink: cover_pk already set for audio asset {} ({})", + pk, + existing_pk + ); + true } + Ok(None) => false, Err(e) if e.is_transient() => { tracing::debug!( - "FlacCacheSink: Transient error getting cover URL for pk {}: {}", + "FlacCacheSink: Transient error getting cover_pk for pk {}: {}", pk, e ); - None + false } Err(e) => { tracing::warn!( - "FlacCacheSink: Cannot obtain cover URL for audio asset {}: {}", + "FlacCacheSink: Cannot obtain cover_pk for audio asset {}: {}", pk, e ); - None + false + } + }; + + let url = if cover_pk_present { + None + } else { + match dest_metadata.read().await.get_cover_url().await { + Ok(url) => { + tracing::debug!( + "FlacCacheSink: Got cover URL for pk {}: {:?}", + pk, + url + ); + url + } + Err(e) if e.is_transient() => { + tracing::debug!( + "FlacCacheSink: Transient error getting cover URL for pk {}: {}", + pk, + e + ); + None + } + Err(e) => { + tracing::warn!( + "FlacCacheSink: Cannot obtain cover URL for audio asset {}: {}", + pk, + e + ); + None + } } }; @@ -378,6 +423,16 @@ impl NodeLogic for FlacCacheSinkLogic { continue; // Passer à la track suivante (qui n'arrivera pas car EndOfStream) } + // Si TrackBoundary précoce a été reçu pendant le prebuffer, passer à la track suivante + if early_track_boundary_received { + tracing::debug!( + "FlacCacheSink: TrackBoundary was received during prebuffer, track complete, moving to next track" + ); + drop(pump_handle); + track_number += 1; + continue; // Passer à la track suivante (métadonnées déjà stockées dans next_track_metadata) + } + // Phase 3: Continuer à dispatcher jusqu'au TrackBoundary tracing::debug!( "FlacCacheSink: Continuing dispatch until TrackBoundary (pump runs in background)" diff --git a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs index aa81021c..25c66364 100644 --- a/pmoaudio-ext/src/sinks/streaming_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_flac_sink.rs @@ -79,7 +79,7 @@ use pmometadata::TrackMetadata; use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf}; use tokio::sync::{mpsc, RwLock}; use tokio_util::sync::CancellationToken; -use tracing::{debug, error, info, trace, warn}; +use tracing::{debug, error, trace, warn}; /// Default ICY metadata interval (bytes of audio between metadata blocks). /// Standard value used by most streaming servers. @@ -291,7 +291,7 @@ impl AsyncRead for FlacClientStream { if let Some(header) = header_opt { self.buffer.extend(header.iter()); - info!( + debug!( "Sending cached FLAC header to new client ({} bytes)", header.len() ); @@ -356,10 +356,10 @@ impl Drop for FlacClientStream { if count == 1 { if self.handle.auto_stop.load(Ordering::SeqCst) { - info!("Last client disconnected, signaling pipeline stop"); + debug!("Last client disconnected, signaling pipeline stop"); self.handle.stop_token.cancel(); } else { - info!("Last client disconnected, keeping pipeline alive"); + debug!("Last client disconnected, keeping pipeline alive"); } } } @@ -465,7 +465,7 @@ impl AsyncRead for IcyClientStream { if let Some(header) = header_opt { self.buffer.extend(header.iter()); - info!( + debug!( "Sending cached FLAC header to new ICY client ({} bytes)", header.len() ); @@ -568,10 +568,10 @@ impl Drop for IcyClientStream { if count == 1 { if self.handle.auto_stop.load(Ordering::SeqCst) { - info!("Last client disconnected, signaling pipeline stop"); + debug!("Last client disconnected, signaling pipeline stop"); self.handle.stop_token.cancel(); } else { - info!("Last client disconnected, keeping pipeline alive"); + debug!("Last client disconnected, keeping pipeline alive"); } } } @@ -603,7 +603,7 @@ impl StreamingFlacSinkLogic { return Ok(()); // Already initialized } - info!( + debug!( "Initializing FLAC encoder with sample rate: {} Hz", sample_rate ); @@ -634,7 +634,7 @@ impl StreamingFlacSinkLogic { AudioError::ProcessingError(format!("Failed to start FLAC encoder: {}", e)) })?; - info!("FLAC encoder initialized successfully"); + debug!("FLAC encoder initialized successfully"); // Spawn broadcaster task with timestamp for pacing let flac_broadcast = self.flac_broadcast.clone(); @@ -656,7 +656,7 @@ impl StreamingFlacSinkLogic { self.encoder_state = Some(EncoderState { broadcaster_task }); - info!("Broadcaster task spawned"); + debug!("Broadcaster task spawned"); Ok(()) } @@ -716,7 +716,7 @@ impl NodeLogic for StreamingFlacSinkLogic { AudioError::ProcessingError("StreamingFlacSink requires an input".into()) })?; - info!("StreamingFlacSink started"); + debug!("StreamingFlacSink started"); // We'll initialize the encoder lazily when we get the first chunk // For now, just process segments @@ -724,7 +724,7 @@ impl NodeLogic for StreamingFlacSinkLogic { loop { tokio::select! { _ = stop_token.cancelled() => { - info!("StreamingFlacSink stopped by cancellation"); + debug!("StreamingFlacSink stopped by cancellation"); break; } @@ -737,7 +737,7 @@ impl NodeLogic for StreamingFlacSinkLogic { if self.sample_rate.is_none() { let sample_rate = chunk.sample_rate(); self.sample_rate = Some(sample_rate); - info!("Detected sample rate: {} Hz", sample_rate); + debug!("Detected sample rate: {} Hz", sample_rate); // Initialize the FLAC encoder now self.initialize_encoder(sample_rate).await?; @@ -765,11 +765,11 @@ impl NodeLogic for StreamingFlacSinkLogic { } let send_duration = send_start.elapsed(); if send_duration.as_millis() >= 50 { - debug!( - "StreamingFlacSink: pcm_tx send blocked for {:.3}s (ts={:.3}s)", - send_duration.as_secs_f64(), - seg.timestamp_sec - ); + trace!( + "StreamingFlacSink: pcm_tx send blocked for {:.3}s (ts={:.3}s)", + send_duration.as_secs_f64(), + seg.timestamp_sec + ); } } @@ -782,7 +782,7 @@ impl NodeLogic for StreamingFlacSinkLogic { } SyncMarker::EndOfStream => { - info!("End of stream marker received"); + debug!("End of stream marker received"); break; } @@ -800,7 +800,7 @@ impl NodeLogic for StreamingFlacSinkLogic { } None => { - info!("Input channel closed"); + debug!("Input channel closed"); break; } } @@ -808,12 +808,12 @@ impl NodeLogic for StreamingFlacSinkLogic { } } - info!("StreamingFlacSink processing complete"); + debug!("StreamingFlacSink processing complete"); Ok(()) } async fn cleanup(&mut self, reason: StopReason) -> Result<(), AudioError> { - info!("StreamingFlacSink cleanup: {:?}", reason); + debug!("StreamingFlacSink cleanup: {:?}", reason); Ok(()) } } @@ -828,7 +828,7 @@ async fn broadcast_flac_stream( current_timestamp: Arc>, broadcast_max_lead_time: f64, ) -> Result<(), AudioError> { - info!( + trace!( "Broadcaster task started with FLAC frame boundary detection (max_lead={:.3}s)", broadcast_max_lead_time ); @@ -861,7 +861,7 @@ async fn broadcast_flac_stream( break; } } - info!("FLAC encoder stream ended, total bytes: {}", total_bytes); + trace!("FLAC encoder stream ended, total bytes: {}", total_bytes); break; } Ok(n) => { @@ -870,7 +870,7 @@ async fn broadcast_flac_stream( total_read_time += read_duration; if read_duration > 0.01 { - debug!( + trace!( "FLAC: flac_stream.read() took {:.3}s for {} bytes (avg: {:.3}s over {} reads)", read_duration, n, @@ -924,7 +924,7 @@ async fn broadcast_flac_stream( let audio_timestamp = *current_timestamp.read().await; if stats_last_log.elapsed() >= Duration::from_secs(1) { - debug!( + trace!( "Broadcaster pacing snapshot: audio_ts={:.3}s buffer_bytes={}", audio_timestamp, accumulator.len() @@ -951,7 +951,7 @@ async fn broadcast_flac_stream( // Log if interval is unusual (too short = burst, too long = stall) if broadcast_interval < 0.01 || broadcast_interval > 0.1 { - debug!( + trace!( "FLAC: broadcast interval {:.3}s ({}ms) - size={} bytes (count={})", broadcast_interval, (broadcast_interval * 1000.0) as u32, @@ -962,7 +962,7 @@ async fn broadcast_flac_stream( // Periodic stats if broadcast_count % 100 == 0 { - debug!( + trace!( "FLAC: {} broadcasts sent, accumulator={} bytes remaining", broadcast_count, accumulator.len() @@ -973,7 +973,7 @@ async fn broadcast_flac_stream( if !header_captured && bytes.len() >= 4 && &bytes[0..4] == b"fLaC" { *header_cache.write().await = Some(bytes.clone()); header_captured = true; - info!("FLAC header captured ({} bytes)", bytes.len()); + trace!("FLAC header captured ({} bytes)", bytes.len()); } let num_receivers = broadcast_tx.receiver_count(); @@ -1013,7 +1013,7 @@ async fn broadcast_flac_stream( ))); } - info!("Broadcaster task completed successfully"); + trace!("Broadcaster task completed successfully"); Ok(()) } @@ -1062,7 +1062,7 @@ impl StreamingFlacSink { // Calculate broadcast capacity based on max_lead_time let broadcast_capacity = calculate_broadcast_capacity(broadcast_max_lead_time); - info!( + debug!( "StreamingFlacSink: using broadcast capacity of {} items (max_lead_time={:.1}s)", broadcast_capacity, broadcast_max_lead_time ); diff --git a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs index 106d6939..3710a799 100644 --- a/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs +++ b/pmoaudio-ext/src/sinks/streaming_ogg_flac_sink.rs @@ -191,7 +191,7 @@ impl AsyncRead for OggFlacClientStream { if let Some(header) = header_opt { self.buffer.extend(header.iter()); - info!( + debug!( "Sending cached OGG-FLAC header to new client ({} bytes)", header.len() ); @@ -255,10 +255,10 @@ impl Drop for OggFlacClientStream { if count == 1 { if self.handle.auto_stop.load(Ordering::SeqCst) { - info!("Last OGG-FLAC client disconnected, signaling pipeline stop"); + debug!("Last OGG-FLAC client disconnected, signaling pipeline stop"); self.handle.stop_token.cancel(); } else { - info!("Last OGG-FLAC client disconnected, keeping pipeline alive"); + debug!("Last OGG-FLAC client disconnected, keeping pipeline alive"); } } } @@ -290,7 +290,7 @@ impl StreamingOggFlacSinkLogic { return Ok(()); // Already initialized } - info!( + debug!( "Initializing OGG-FLAC encoder with sample rate: {} Hz", sample_rate ); @@ -321,7 +321,7 @@ impl StreamingOggFlacSinkLogic { AudioError::ProcessingError(format!("Failed to start FLAC encoder: {}", e)) })?; - info!("OGG-FLAC encoder initialized successfully"); + debug!("OGG-FLAC encoder initialized successfully"); // Spawn OGG wrapper + broadcaster task with timestamp for pacing let ogg_broadcast = self.ogg_broadcast.clone(); @@ -343,7 +343,7 @@ impl StreamingOggFlacSinkLogic { self.encoder_state = Some(EncoderState { broadcaster_task }); - info!("OGG broadcaster task spawned"); + debug!("OGG broadcaster task spawned"); Ok(()) } @@ -402,7 +402,7 @@ impl NodeLogic for StreamingOggFlacSinkLogic { AudioError::ProcessingError("StreamingOggFlacSink requires an input".into()) })?; - info!("StreamingOggFlacSink started"); + debug!("StreamingOggFlacSink started"); // TODO: Implement OGG-FLAC encoding logic // For now, just process segments without encoding @@ -410,7 +410,7 @@ impl NodeLogic for StreamingOggFlacSinkLogic { loop { tokio::select! { _ = stop_token.cancelled() => { - info!("StreamingOggFlacSink stopped by cancellation"); + debug!("StreamingOggFlacSink stopped by cancellation"); break; } @@ -423,7 +423,7 @@ impl NodeLogic for StreamingOggFlacSinkLogic { if self.sample_rate.is_none() { let sample_rate = chunk.sample_rate(); self.sample_rate = Some(sample_rate); - info!("Detected sample rate: {} Hz", sample_rate); + debug!("Detected sample rate: {} Hz", sample_rate); // Initialize the FLAC encoder now self.initialize_encoder(sample_rate).await?; @@ -460,7 +460,7 @@ impl NodeLogic for StreamingOggFlacSinkLogic { } SyncMarker::EndOfStream => { - info!("End of stream marker received"); + debug!("End of stream marker received"); break; } @@ -478,7 +478,7 @@ impl NodeLogic for StreamingOggFlacSinkLogic { } None => { - info!("Input channel closed"); + debug!("Input channel closed"); break; } } @@ -486,12 +486,12 @@ impl NodeLogic for StreamingOggFlacSinkLogic { } } - info!("StreamingOggFlacSink processing complete"); + debug!("StreamingOggFlacSink processing complete"); Ok(()) } async fn cleanup(&mut self, reason: StopReason) -> Result<(), AudioError> { - info!("StreamingOggFlacSink cleanup: {:?}", reason); + debug!("StreamingOggFlacSink cleanup: {:?}", reason); Ok(()) } } @@ -545,7 +545,7 @@ impl StreamingOggFlacSink { // Broadcast channel for OGG-FLAC bytes // Capacity calculated from max_lead_time to ensure enough buffering let broadcast_capacity = calculate_broadcast_capacity(broadcast_max_lead_time); - tracing::debug!( + tracing::trace!( "OGG-FLAC broadcast capacity: {} items (for {:.1}s max lead time)", broadcast_capacity, broadcast_max_lead_time @@ -786,7 +786,7 @@ async fn broadcast_ogg_flac_stream( current_timestamp: Arc>, broadcast_max_lead_time: f64, ) -> Result<(), AudioError> { - info!( + trace!( "OGG-FLAC broadcaster task started with FLAC frame boundary detection (max_lead={:.3}s)", broadcast_max_lead_time ); @@ -807,16 +807,16 @@ async fn broadcast_ogg_flac_stream( // Step 1: Read FLAC header (fLaC + metadata blocks) let flac_header = read_flac_header(&mut flac_stream).await?; - info!("Read FLAC header: {} bytes", flac_header.len()); + trace!("Read FLAC header: {} bytes", flac_header.len()); // Extract sample rate from STREAMINFO for granule position calculation let sample_rate = extract_sample_rate_from_streaminfo(&flac_header)?; - info!("Extracted sample rate from STREAMINFO: {} Hz", sample_rate); + trace!("Extracted sample rate from STREAMINFO: {} Hz", sample_rate); // Step 2: Create OGG-FLAC identification packet (BOS) // Format according to https://xiph.org/flac/ogg_mapping.html let ogg_flac_id = create_ogg_flac_identification(&flac_header)?; - info!( + trace!( "Created OGG-FLAC identification packet: {} bytes", ogg_flac_id.len() ); @@ -835,7 +835,7 @@ async fn broadcast_ogg_flac_stream( cached_header.extend_from_slice(&comment_bytes); *header_cache.write().await = Some(Bytes::from(cached_header)); header_captured = true; - info!( + trace!( "OGG-FLAC header cached ({} bytes: BOS + Vorbis Comment)", bos_bytes.len() + comment_bytes.len() ); @@ -871,7 +871,7 @@ async fn broadcast_ogg_flac_stream( trace!("Broadcast closed before sending final EOS page"); break; } - info!( + trace!( "Sent final EOS page with {} bytes of data", flac_accumulator.len() ); @@ -885,10 +885,10 @@ async fn broadcast_ogg_flac_stream( trace!("Broadcast closed before sending empty EOS page"); break; } - info!("Sent empty EOS page"); + trace!("Sent empty EOS page"); } - info!( + trace!( "OGG-FLAC stream ended, total OGG bytes: {}", total_ogg_bytes ); @@ -900,7 +900,7 @@ async fn broadcast_ogg_flac_stream( total_read_time += read_duration; if read_duration > 0.01 { - debug!( + trace!( "OGG: flac_stream.read() took {:.3}s for {} bytes (avg: {:.3}s over {} reads)", read_duration, n, @@ -1002,7 +1002,7 @@ async fn broadcast_ogg_flac_stream( // Log if interval is unusual (too short = burst, too long = stall) if broadcast_interval < 0.01 || broadcast_interval > 0.1 { - debug!( + trace!( "OGG: broadcast interval {:.3}s ({}ms) - frame_size={} bytes, samples={} (count={})", broadcast_interval, (broadcast_interval * 1000.0) as u32, @@ -1014,7 +1014,7 @@ async fn broadcast_ogg_flac_stream( // Periodic stats if broadcast_count % 100 == 0 { - debug!( + trace!( "OGG: {} broadcasts sent, avg_interval={:.3}s, accumulator={} bytes", broadcast_count, last_broadcast_time.elapsed().as_secs_f64() / broadcast_count as f64, @@ -1053,7 +1053,7 @@ async fn broadcast_ogg_flac_stream( ))); } - info!("OGG-FLAC broadcaster task completed successfully"); + trace!("OGG-FLAC broadcaster task completed successfully"); Ok(()) } @@ -1177,7 +1177,7 @@ fn create_ogg_flac_identification(flac_header: &[u8]) -> Result, AudioEr let block_length = u32::from_be_bytes([0, flac_header[5], flac_header[6], flac_header[7]]) as usize; - info!("STREAMINFO block_length = {} bytes", block_length); + trace!("STREAMINFO block_length = {} bytes", block_length); // STREAMINFO should be exactly 34 bytes of data if block_length != 34 { @@ -1194,7 +1194,7 @@ fn create_ogg_flac_identification(flac_header: &[u8]) -> Result, AudioEr // Extract just the STREAMINFO block (type + length + data) let streaminfo = &flac_header[4..4 + streaminfo_size]; - info!( + trace!( "Extracted STREAMINFO: {} bytes (type+length+data)", streaminfo.len() ); diff --git a/pmoaudio-ext/src/sinks/timed_broadcast.rs b/pmoaudio-ext/src/sinks/timed_broadcast.rs index e73b5f8d..2f06f25a 100644 --- a/pmoaudio-ext/src/sinks/timed_broadcast.rs +++ b/pmoaudio-ext/src/sinks/timed_broadcast.rs @@ -15,7 +15,7 @@ use std::{ }; use tokio::sync::Notify; -use tracing::warn; +use tracing::{trace, warn}; /// Paquet diffusé contenant la charge utile + méta timing. #[derive(Clone)] @@ -99,7 +99,7 @@ impl State { } } if purged > 0 { - tracing::debug!( + trace!( "TimedBroadcast: purged {} expired packet(s) (head_seq={})", purged, self.head_seq diff --git a/pmomediaserver/Cargo.toml b/pmomediaserver/Cargo.toml index 64327c8a..88d7af76 100644 --- a/pmomediaserver/Cargo.toml +++ b/pmomediaserver/Cargo.toml @@ -25,6 +25,10 @@ utoipa = { version = "5.3", optional = true } pmoqobuz = { path = "../pmoqobuz", optional = true } pmoparadise = { path = "../pmoparadise", optional = true } pmoconfig = { path = "../pmoconfig", optional = true } +anyhow = { version = "1.0", optional = true } +pmoaudiocache = { path = "../pmoaudiocache", optional = true } +pmocovers = { path = "../pmocovers", optional = true } +tokio-util = { version = "0.7", features = ["io"], optional = true } [features] default = ["pmosource/server"] @@ -33,6 +37,14 @@ api = ["dep:axum", "dep:utoipa", "pmosource/server"] # Feature pour activer le support Qobuz configuré qobuz = ["api", "dep:pmoqobuz", "dep:pmoconfig", "pmoqobuz/server"] # Feature pour activer le support Radio Paradise -paradise = ["api", "dep:pmoparadise", "pmoparadise/server"] +paradise = [ + "api", + "dep:pmoparadise", + "pmoparadise/full", + "dep:anyhow", + "dep:pmoaudiocache", + "dep:pmocovers", + "dep:tokio-util" +] # Feature pour activer l'API REST de Radio Paradise (en plus de la source UPnP) paradise-api = ["paradise", "pmoparadise/pmoserver"] diff --git a/pmomediaserver/src/lib.rs b/pmomediaserver/src/lib.rs index 796e8ee2..dd63ef93 100644 --- a/pmomediaserver/src/lib.rs +++ b/pmomediaserver/src/lib.rs @@ -75,12 +75,19 @@ pub mod sources; #[cfg(any(feature = "qobuz", feature = "paradise"))] pub mod sources_api; +// Extension pour le streaming Paradise (requires feature paradise) +#[cfg(feature = "paradise")] +pub mod paradise_streaming; + pub use content_handler::ContentHandler; pub use device::MEDIA_SERVER; pub use server_ext::{MediaServerExt, MusicSourceExt, get_source_registry}; pub use source_registry::SourceRegistry; pub use sources::{SourceInitError, SourcesExt}; +#[cfg(feature = "paradise")] +pub use paradise_streaming::ParadiseStreamingExt; + // Re-export sources when features are enabled #[cfg(feature = "qobuz")] pub use pmoqobuz; diff --git a/pmomediaserver/src/paradise_streaming.rs b/pmomediaserver/src/paradise_streaming.rs new file mode 100644 index 00000000..6ef2a28e --- /dev/null +++ b/pmomediaserver/src/paradise_streaming.rs @@ -0,0 +1,280 @@ +//! Extension pour l'initialisation des canaux de streaming Radio Paradise +//! +//! Ce module fournit un trait d'extension pour démarrer les pipelines de streaming +//! Radio Paradise avec caching audio/covers et historique. + +use anyhow::{Context, Result}; +use async_trait::async_trait; +use axum::{ + body::Body, + extract::{Path, State}, + http::StatusCode, + response::{IntoResponse, Response}, + routing::get, + Json, Router, +}; +use pmoaudiocache::{get_audio_cache, register_audio_cache, AudioCacheExt, Cache as AudioCache}; +use pmocovers::{get_cover_cache, register_cover_cache, Cache as CoverCache, CoverCacheExt}; +use pmoparadise::{channels::ALL_CHANNELS, ParadiseChannelManager, ParadiseHistoryBuilder}; +use std::sync::Arc; +use tokio_util::io::ReaderStream; +use tracing::{error, info}; + +/// État partagé pour les routes de streaming Paradise +#[derive(Clone)] +pub struct ParadiseStreamingState { + pub manager: Arc, +} + +/// Extension trait pour initialiser les canaux de streaming Radio Paradise +#[async_trait] +pub trait ParadiseStreamingExt { + /// Initialise les canaux de streaming Radio Paradise avec caching + /// + /// Cette méthode : + /// - Crée les caches audio et covers + /// - Initialise le ParadiseChannelManager avec historique + /// - Ajoute les routes de streaming HTTP (flac, ogg, history, metadata) + /// + /// # Routes créées + /// + /// Pour chaque canal (main, mellow, rock, eclectic) : + /// - `/radioparadise/stream/{slug}/flac` - Stream FLAC live + /// - `/radioparadise/stream/{slug}/ogg` - Stream OGG live + /// - `/radioparadise/stream/{slug}/historic/{client_id}/flac` - Historique FLAC + /// - `/radioparadise/stream/{slug}/historic/{client_id}/ogg` - Historique OGG + /// - `/radioparadise/metadata/{slug}` - Métadonnées en temps réel + /// + /// # Exemples + /// + /// ```ignore + /// use pmomediaserver::ParadiseStreamingExt; + /// + /// server.init_paradise_streaming().await?; + /// ``` + async fn init_paradise_streaming(&mut self) -> Result>; +} + +#[async_trait] +impl ParadiseStreamingExt for pmoserver::Server { + async fn init_paradise_streaming(&mut self) -> Result> { + info!("🎵 Initializing Radio Paradise streaming channels..."); + + // Récupérer ou initialiser les caches singletons + info!("📦 Getting cache singletons..."); + let cover_cache = match get_cover_cache() { + Some(cache) => { + info!(" ✅ Using existing cover cache singleton"); + cache + } + None => { + info!(" 📦 Initializing new cover cache singleton"); + let cache = self + .init_cover_cache_configured() + .await + .context("Failed to initialize cover cache")?; + register_cover_cache(cache.clone()); + cache + } + }; + + let audio_cache = match get_audio_cache() { + Some(cache) => { + info!(" ✅ Using existing audio cache singleton"); + cache + } + None => { + info!(" 📦 Initializing new audio cache singleton"); + let cache = self + .init_audio_cache_configured() + .await + .context("Failed to initialize audio cache")?; + register_audio_cache(cache.clone()); + cache + } + }; + + // Créer le builder d'historique + let history_builder = ParadiseHistoryBuilder { + audio_cache: audio_cache.clone(), + cover_cache: cover_cache.clone(), + playlist_prefix: "radioparadise-history".into(), + playlist_title_prefix: Some("Radio Paradise History".into()), + max_history_tracks: Some(500), + collection_prefix: Some("radioparadise".into()), + replay_max_lead_seconds: 1.0, + }; + + // Créer le manager de canaux + info!("📡 Creating ParadiseChannelManager..."); + let manager = Arc::new( + ParadiseChannelManager::with_defaults_with_cover_cache( + Some(cover_cache.clone()), + Some(history_builder), + ) + .await + .context("Failed to create ParadiseChannelManager")?, + ); + + let state = Arc::new(ParadiseStreamingState { + manager: manager.clone(), + }); + + // Ajouter les routes pour chaque canal + info!("🌐 Registering streaming routes..."); + for descriptor in ALL_CHANNELS.iter() { + let slug = descriptor.slug; + let channel_id = descriptor.id; + + // Route FLAC live + let flac_path = format!("/radioparadise/stream/{}/flac", slug); + self.add_handler_with_state( + &flac_path, + move |State(state): State>| { + let manager = state.manager.clone(); + async move { stream_flac(manager, channel_id).await } + }, + state.clone(), + ) + .await; + + // Route OGG live + let ogg_path = format!("/radioparadise/stream/{}/ogg", slug); + self.add_handler_with_state( + &ogg_path, + move |State(state): State>| { + let manager = state.manager.clone(); + async move { stream_ogg(manager, channel_id).await } + }, + state.clone(), + ) + .await; + + // Routes historique + let history_path = format!("/radioparadise/stream/{}/historic", slug); + let history_router = Router::new() + .route( + "/{client_id}/flac", + get({ + let manager = manager.clone(); + move |Path(client_id): Path| { + let manager = manager.clone(); + async move { stream_history_flac(manager, channel_id, client_id).await } + } + }), + ) + .route( + "/{client_id}/ogg", + get({ + let manager = manager.clone(); + move |Path(client_id): Path| { + let manager = manager.clone(); + async move { stream_history_ogg(manager, channel_id, client_id).await } + } + }), + ); + + self.add_router(&history_path, history_router).await; + + // Route métadonnées + let meta_path = format!("/radioparadise/metadata/{}", slug); + self.add_handler_with_state( + &meta_path, + move |State(state): State>| { + let manager = state.manager.clone(); + async move { get_metadata(manager, channel_id).await } + }, + state.clone(), + ) + .await; + + info!( + " ✅ {} - /radioparadise/stream/{}/{{flac,ogg}}", + descriptor.display_name, slug + ); + } + + info!("✅ Radio Paradise streaming channels initialized"); + + Ok(manager) + } +} + +// ============================================================================ +// Handlers de streaming +// ============================================================================ + +async fn stream_flac( + manager: Arc, + channel_id: u8, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let stream = channel.subscribe_flac(); + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/flac") + .body(Body::from_stream(ReaderStream::new(stream))) + .unwrap()) +} + +async fn stream_ogg( + manager: Arc, + channel_id: u8, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let stream = channel.subscribe_ogg(); + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/ogg") + .body(Body::from_stream(ReaderStream::new(stream))) + .unwrap()) +} + +async fn get_metadata( + manager: Arc, + channel_id: u8, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let metadata = channel.metadata().await; + Ok(Json(metadata)) +} + +async fn stream_history_flac( + manager: Arc, + channel_id: u8, + client_id: String, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let stream = channel.stream_history_flac(&client_id).await.map_err(|e| { + error!( + "Failed to start historical FLAC stream for channel {} (client_id={}): {}", + channel_id, client_id, e + ); + StatusCode::INTERNAL_SERVER_ERROR + })?; + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/flac") + .body(Body::from_stream(ReaderStream::new(stream))) + .unwrap()) +} + +async fn stream_history_ogg( + manager: Arc, + channel_id: u8, + client_id: String, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let stream = channel.stream_history_ogg(&client_id).await.map_err(|e| { + error!( + "Failed to start historical OGG stream for channel {} (client_id={}): {}", + channel_id, client_id, e + ); + StatusCode::INTERNAL_SERVER_ERROR + })?; + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/ogg") + .body(Body::from_stream(ReaderStream::new(stream))) + .unwrap()) +} diff --git a/pmomediaserver/src/sources.rs b/pmomediaserver/src/sources.rs index aa8895b0..370809ba 100644 --- a/pmomediaserver/src/sources.rs +++ b/pmomediaserver/src/sources.rs @@ -177,18 +177,13 @@ impl SourcesExt for Server { tracing::info!("Initializing Radio Paradise source..."); - // Créer le client (Radio Paradise ne nécessite pas d'authentification) - let client = RadioParadiseClient::new().await.map_err(|e| { - SourceInitError::ParadiseError(format!("Failed to create client: {}", e)) - })?; + // Obtenir l'URL de base du serveur + let base_url = self.base_url(); - // Créer la source depuis le registry avec capacité FIFO par défaut - let source = RadioParadiseSource::from_registry_default(client).map_err(|e| { - SourceInitError::ParadiseError(format!("Failed to create source: {}", e)) - })?; + // Créer la source Radio Paradise (utilise le singleton PlaylistManager) + let source = RadioParadiseSource::new(base_url.to_string()); // Enregistrer la source - // Note: La FIFO sera peuplée automatiquement lors du premier browse self.register_music_source(Arc::new(source)).await; tracing::info!("✅ Radio Paradise source registered successfully"); diff --git a/pmomediaserver/src/sources_api.rs b/pmomediaserver/src/sources_api.rs index ee4b8193..6b4d56e1 100644 --- a/pmomediaserver/src/sources_api.rs +++ b/pmomediaserver/src/sources_api.rs @@ -35,6 +35,10 @@ pub struct ParadiseParams { /// Capacité FIFO (optionnelle, 50 par défaut) #[serde(default)] pub fifo_capacity: Option, + + /// URL de base du serveur (optionnelle, "http://localhost:8080" par défaut) + #[serde(default)] + pub base_url: Option, } /// Réponse d'enregistrement de source @@ -136,34 +140,11 @@ async fn register_paradise(Json(params): Json) -> impl IntoRespo use pmoparadise::{RadioParadiseClient, RadioParadiseSource}; use pmosource::api::register_source; - // Créer le client (Radio Paradise ne nécessite pas d'auth) - let client = match RadioParadiseClient::new().await { - Ok(c) => c, - Err(e) => { - return ( - StatusCode::BAD_REQUEST, - Json(ErrorResponse { - error: format!("Failed to create Radio Paradise client: {}", e), - }), - ) - .into_response(); - } - }; + // Utiliser l'URL de base depuis les params ou une valeur par défaut + let base_url = params.base_url.unwrap_or_else(|| "http://localhost:8080".to_string()); - // Créer et enregistrer la source depuis le registry - // Note: params.fifo_capacity is currently not used by from_registry - let source = match RadioParadiseSource::from_registry(client) { - Ok(s) => Arc::new(s), - Err(e) => { - return ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(ErrorResponse { - error: format!("Failed to create source: {}", e), - }), - ) - .into_response(); - } - }; + // Créer la source Radio Paradise (utilise le singleton PlaylistManager) + let source = Arc::new(RadioParadiseSource::new(base_url)); let source_id = source.as_ref().id().to_string(); diff --git a/pmoparadise/Cargo.toml b/pmoparadise/Cargo.toml index 6267f96b..0e747ff9 100644 --- a/pmoparadise/Cargo.toml +++ b/pmoparadise/Cargo.toml @@ -52,7 +52,7 @@ symphonia = { version = "0.5", features = ["all"] } claxon = "0.4" # pmoaudio-ext with playlist support (optional for examples) -pmoaudio-ext = { path = "../pmoaudio-ext", optional = true, features = ["playlist", "http-stream"] } +pmoaudio-ext = { path = "../pmoaudio-ext", optional = true, features = ["playlist", "http-stream", "cache-sink"] } # Common music source traits pmosource = { path = "../pmosource" } diff --git a/pmoparadise/examples/serve_channels.rs b/pmoparadise/examples/serve_channels.rs index 19c4aef9..9081c1c5 100644 --- a/pmoparadise/examples/serve_channels.rs +++ b/pmoparadise/examples/serve_channels.rs @@ -4,21 +4,27 @@ //! - `/radioparadise/stream//flac` //! - `/radioparadise/stream//ogg` //! - `/radioparadise/stream//icy` +//! - `/radioparadise/stream//historic//flac` +//! - `/radioparadise/stream//historic//ogg` //! - `/radioparadise/metadata/` -use std::sync::Arc; +use std::{fs, sync::Arc}; use axum::{ body::Body, - extract::State, + extract::{Path, State}, http::StatusCode, response::{IntoResponse, Response}, - Json, + routing::get, + Json, Router, }; -use pmoparadise::{channels::ALL_CHANNELS, stream_channel::ParadiseChannelManager}; +use pmoaudiocache::new_cache as new_audio_cache; +use pmocovers::new_cache as new_cover_cache; +use pmoparadise::{channels::ALL_CHANNELS, ParadiseChannelManager, ParadiseHistoryBuilder}; +use pmoplaylist::register_audio_cache; use pmoserver::{init_logging, ServerBuilder}; use tokio_util::io::ReaderStream; -use tracing::info; +use tracing::{error, info}; #[derive(Clone)] struct AppState { @@ -29,8 +35,35 @@ struct AppState { async fn main() -> anyhow::Result<()> { let _ = init_logging(); + // Préparer les caches partagés + let cover_cache_dir = "./cache/rp_covers"; + let audio_cache_dir = "./cache/rp_audio"; + fs::create_dir_all(cover_cache_dir)?; + fs::create_dir_all(audio_cache_dir)?; + + let cover_cache = Arc::new(new_cover_cache(cover_cache_dir, 500)?); + let audio_cache = Arc::new(new_audio_cache(audio_cache_dir, 1000)?); + register_audio_cache(audio_cache.clone()); + let _playlist_manager = pmoplaylist::PlaylistManager(); + + let history_builder = ParadiseHistoryBuilder { + audio_cache: audio_cache.clone(), + cover_cache: cover_cache.clone(), + playlist_prefix: "radioparadise-history".into(), + playlist_title_prefix: Some("Radio Paradise History".into()), + max_history_tracks: Some(500), + collection_prefix: Some("radioparadise".into()), + replay_max_lead_seconds: 1.0, + }; + info!("Initializing Radio Paradise channels..."); - let manager = Arc::new(ParadiseChannelManager::with_defaults().await?); + let manager = Arc::new( + ParadiseChannelManager::with_defaults_with_cover_cache( + Some(cover_cache), + Some(history_builder), + ) + .await?, + ); let app_state = Arc::new(AppState { manager: manager.clone(), }); @@ -42,6 +75,7 @@ async fn main() -> anyhow::Result<()> { let flac_path = format!("/radioparadise/stream/{}/flac", slug); let ogg_path = format!("/radioparadise/stream/{}/ogg", slug); let icy_path = format!("/radioparadise/stream/{}/icy", slug); + let history_path = format!("/radioparadise/stream/{}/historic", slug); let meta_path = format!("/radioparadise/metadata/{}", slug); let channel_id = descriptor.id; @@ -78,6 +112,30 @@ async fn main() -> anyhow::Result<()> { ) .await; + let history_router = Router::new() + .route( + "/{client_id}/flac", + get({ + let manager = manager.clone(); + move |Path(client_id): Path| { + let manager = manager.clone(); + async move { stream_history_flac(manager, channel_id, client_id).await } + } + }), + ) + .route( + "/{client_id}/ogg", + get({ + let manager = manager.clone(); + move |Path(client_id): Path| { + let manager = manager.clone(); + async move { stream_history_ogg(manager, channel_id, client_id).await } + } + }), + ); + + server.add_router(&history_path, history_router).await; + server .add_handler_with_state( &meta_path, @@ -95,7 +153,7 @@ async fn main() -> anyhow::Result<()> { info!("Available channels:"); for descriptor in ALL_CHANNELS.iter() { info!( - " {}: /radioparadise/stream/{}/flac (also /ogg, /icy, metadata)", + " {}: /radioparadise/stream/{}/flac (also /ogg, /icy, metadata, /historic//(flac|ogg))", descriptor.display_name, descriptor.slug ); } @@ -155,3 +213,43 @@ async fn get_metadata( let metadata = channel.metadata().await; Ok(Json(metadata)) } + +async fn stream_history_flac( + manager: Arc, + channel_id: u8, + client_id: String, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let stream = channel.stream_history_flac(&client_id).await.map_err(|e| { + error!( + "Failed to start historical FLAC stream for channel {} (client_id={}): {}", + channel_id, client_id, e + ); + StatusCode::INTERNAL_SERVER_ERROR + })?; + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/flac") + .body(Body::from_stream(ReaderStream::new(stream))) + .unwrap()) +} + +async fn stream_history_ogg( + manager: Arc, + channel_id: u8, + client_id: String, +) -> Result { + let channel = manager.get(channel_id).ok_or(StatusCode::NOT_FOUND)?; + let stream = channel.stream_history_ogg(&client_id).await.map_err(|e| { + error!( + "Failed to start historical OGG stream for channel {} (client_id={}): {}", + channel_id, client_id, e + ); + StatusCode::INTERNAL_SERVER_ERROR + })?; + Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "audio/ogg") + .body(Body::from_stream(ReaderStream::new(stream))) + .unwrap()) +} diff --git a/pmoparadise/src/lib.rs b/pmoparadise/src/lib.rs index b0b7d0b5..f7896c1f 100644 --- a/pmoparadise/src/lib.rs +++ b/pmoparadise/src/lib.rs @@ -203,7 +203,7 @@ //! - `pmoserver`: Enable REST API extension for pmoserver integration (adds `utoipa`, `axum`) //! - `pmoaudio`: Enable RadioParadiseStreamSource for pmoaudio integration //! - `pmoconfig`: Enable configuration integration with pmoconfig -//! - `server`: Enable RadioParadiseSource stub for backward compatibility (deprecated) +//! - `server`: Enable RadioParadiseSource for UPnP ContentDirectory integration //! //! ## See Also //! @@ -242,7 +242,9 @@ pub use radio_paradise_stream_source::{RadioParadiseStreamSource, END_OF_BLOCKS_ #[cfg(feature = "pmoaudio")] pub use stream_channel::{ - ParadiseChannelManager, ParadiseStreamChannel, ParadiseStreamChannelConfig, + HistoryFlacStream, HistoryOggStream, HistoryStreamError, ParadiseChannelManager, + ParadiseHistoryBuilder, ParadiseHistoryOptions, ParadiseStreamChannel, + ParadiseStreamChannelConfig, }; #[cfg(feature = "pmoserver")] diff --git a/pmoparadise/src/models.rs b/pmoparadise/src/models.rs index cfad4bc8..aa76ca0d 100644 --- a/pmoparadise/src/models.rs +++ b/pmoparadise/src/models.rs @@ -161,6 +161,27 @@ pub struct Song { #[serde(default, deserialize_with = "deserialize_optional_string_or_f32")] pub rating: Option, + /// Gapless URL for individual song FLAC + /// This URL points to a FLAC file containing only this song + #[serde(default)] + pub gapless_url: Option, + + /// Scheduled playback time on Radio Paradise (Unix timestamp in milliseconds, UTC) + #[serde(default)] + pub sched_time_millis: Option, + + /// Radio Paradise song ID (unique identifier) + #[serde(default)] + pub song_id: Option, + + /// Radio Paradise artist ID (for building artist URLs) + #[serde(default)] + pub artist_id: Option, + + /// Large cover image path (best quality) + #[serde(default)] + pub cover_large: Option, + /// Additional metadata #[serde(flatten)] pub extra: HashMap, diff --git a/pmoparadise/src/pmoserver_ext.rs b/pmoparadise/src/pmoserver_ext.rs index 0f4867da..5c19d5ad 100644 --- a/pmoparadise/src/pmoserver_ext.rs +++ b/pmoparadise/src/pmoserver_ext.rs @@ -341,14 +341,12 @@ pub struct RadioParadiseApiDoc; /// Crée le router pour l'API Radio Paradise pub fn create_api_router(state: RadioParadiseState) -> Router { - let api = Router::new() + Router::new() .route("/now-playing", get(get_now_playing)) .route("/block/current", get(get_current_block)) .route("/block/{event_id}", get(get_block_by_id)) .route("/channels", get(get_channels)) - .with_state(state); - - Router::new().nest("/radioparadise", api) + .with_state(state) } /// Trait d'extension pour pmoserver::Server diff --git a/pmoparadise/src/radio_paradise_stream_source.rs b/pmoparadise/src/radio_paradise_stream_source.rs index 57afbc92..337e2ab8 100644 --- a/pmoparadise/src/radio_paradise_stream_source.rs +++ b/pmoparadise/src/radio_paradise_stream_source.rs @@ -415,7 +415,7 @@ impl RadioParadiseStreamSourceLogic { if send_duration.as_millis() > 10 { let duration_ms = send_duration.as_millis() as u64; self.stats.record_backpressure(duration_ms); - tracing::debug!( + tracing::trace!( "send_to_children: Send to child {} BLOCKED for {:.3}s (channel capacity before send={}, timestamp={:.3}s)", i, send_duration.as_secs_f64(), diff --git a/pmoparadise/src/source.rs b/pmoparadise/src/source.rs index e9e91328..40cfd434 100644 --- a/pmoparadise/src/source.rs +++ b/pmoparadise/src/source.rs @@ -1,113 +1,264 @@ -//! DEPRECATED: Stub implementation of RadioParadiseSource +//! RadioParadiseSource - Implementation of MusicSource for Radio Paradise //! -//! **⚠️ 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. +//! This module provides a UPnP ContentDirectory source for Radio Paradise, +//! exposing live streams and historical playlists for all 4 channels. -use crate::client::RadioParadiseClient; -use pmosource::pmodidl::{Container, Item}; +use crate::channels::{ChannelDescriptor, ALL_CHANNELS}; +use pmosource::pmodidl::{Container, Item, Resource}; use pmosource::{async_trait, BrowseResult, MusicSource, MusicSourceError, Result}; +use std::sync::Arc; use std::time::SystemTime; +use tokio::sync::RwLock; + +#[cfg(feature = "playlist")] +use pmoplaylist::PlaylistManager; /// Default Radio Paradise image (embedded in binary) const DEFAULT_IMAGE: &[u8] = include_bytes!("../assets/default.webp"); -/// DEPRECATED: Stub implementation of RadioParadiseSource +/// RadioParadiseSource - UPnP ContentDirectory source for Radio Paradise /// -/// 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. +/// Provides access to: +/// - Live OGG streams for all 4 channels (Main, Mellow, Rock, Eclectic) +/// - Historical playlists (FIFO) for each channel /// -/// **Do not use this in new code.** Use `RadioParadiseStreamSource` instead. -#[derive(Clone, Debug)] +/// # Object ID Schema +/// +/// - Root: `radio-paradise` +/// - Channel container: `radio-paradise:channel:{slug}` +/// - Live stream item: `radio-paradise:channel:{slug}:live` +/// - History container: `radio-paradise:channel:{slug}:history` +/// - History track: `radio-paradise:channel:{slug}:history:track:{pk}` +#[derive(Debug, Clone)] pub struct RadioParadiseSource { - _client: RadioParadiseClient, + /// Base URL for streaming server (e.g., "http://localhost:8080") + base_url: String, + /// Update counter for change notifications + update_counter: Arc>, + /// Last change timestamp + last_change: Arc>, } impl RadioParadiseSource { - /// DEPRECATED: Create a new RadioParadiseSource from registry + /// Create a new RadioParadiseSource /// - /// This method is deprecated and will always return an error indicating - /// that the orchestration-based source is no longer supported. + /// # Arguments /// - /// Use `RadioParadiseStreamSource` instead for audio streaming. - #[cfg(feature = "server")] - pub fn from_registry(_client: RadioParadiseClient) -> Result { - Err(MusicSourceError::SourceUnavailable( - "RadioParadiseSource is deprecated. Use RadioParadiseStreamSource instead.".to_string(), - )) + /// * `base_url` - Base URL for streaming server (e.g., "http://localhost:8080") + /// + /// # Note + /// + /// With the "playlist" feature enabled, this source will use the global PlaylistManager + /// singleton to access history playlists. + pub fn new(base_url: impl Into) -> Self { + Self { + base_url: base_url.into(), + update_counter: Arc::new(RwLock::new(0)), + last_change: Arc::new(RwLock::new(SystemTime::now())), + } } - /// 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) -> Self { - tracing::warn!( - "RadioParadiseSource::from_registry_default is deprecated. \ - Use RadioParadiseStreamSource for audio streaming." - ); - Self { _client: client } + /// Build a live stream URL for a channel + fn build_live_url(&self, slug: &str) -> String { + format!("{}/radioparadise/stream/{}/ogg", self.base_url, slug) } - /// 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 } + /// Get the playlist ID for a channel's history + #[cfg(feature = "playlist")] + fn history_playlist_id(slug: &str) -> String { + format!("radioparadise-history-{}", slug) } - /// 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 } + /// Get channel descriptor by slug + fn get_channel_by_slug(slug: &str) -> Option<&'static ChannelDescriptor> { + ALL_CHANNELS.iter().find(|ch| ch.slug == slug) } + + /// Parse an object ID into its components + fn parse_object_id(id: &str) -> ObjectIdType { + let parts: Vec<&str> = id.split(':').collect(); + match parts.as_slice() { + ["radio-paradise"] => ObjectIdType::Root, + ["radio-paradise", "channel", slug] => ObjectIdType::Channel { + slug: (*slug).to_string(), + }, + ["radio-paradise", "channel", slug, "live"] => ObjectIdType::LiveStream { + slug: (*slug).to_string(), + }, + ["radio-paradise", "channel", slug, "history"] => ObjectIdType::History { + slug: (*slug).to_string(), + }, + ["radio-paradise", "channel", slug, "history", "track", pk] => { + ObjectIdType::HistoryTrack { + slug: (*slug).to_string(), + pk: (*pk).to_string(), + } + } + _ => ObjectIdType::Unknown, + } + } + + /// Build a channel container + fn build_channel_container(&self, descriptor: &ChannelDescriptor) -> Container { + Container { + id: format!("radio-paradise:channel:{}", descriptor.slug), + parent_id: "radio-paradise".to_string(), + restricted: Some("1".to_string()), + child_count: Some("2".to_string()), // Live + History + searchable: Some("0".to_string()), + title: descriptor.display_name.to_string(), + class: "object.container".to_string(), + containers: vec![], + items: vec![], + } + } + + /// Build a live stream item for a channel + fn build_live_stream_item(&self, descriptor: &ChannelDescriptor) -> Item { + let stream_url = self.build_live_url(descriptor.slug); + + Item { + id: format!("radio-paradise:channel:{}:live", descriptor.slug), + parent_id: format!("radio-paradise:channel:{}", descriptor.slug), + restricted: Some("1".to_string()), + title: format!("{} - Live Stream", descriptor.display_name), + creator: Some("Radio Paradise".to_string()), + class: "object.item.audioItem.audioBroadcast".to_string(), + artist: Some("Radio Paradise".to_string()), + album: Some(descriptor.display_name.to_string()), + genre: Some("Radio".to_string()), + album_art: None, + album_art_pk: None, + date: None, + original_track_number: None, + resources: vec![Resource { + protocol_info: "http-get:*:audio/ogg:*".to_string(), + bits_per_sample: None, + sample_frequency: None, + nr_audio_channels: Some("2".to_string()), + duration: None, + url: stream_url, + }], + descriptions: vec![], + } + } + + /// Build a history container for a channel + fn build_history_container(&self, descriptor: &ChannelDescriptor) -> Container { + Container { + id: format!("radio-paradise:channel:{}:history", descriptor.slug), + parent_id: format!("radio-paradise:channel:{}", descriptor.slug), + restricted: Some("1".to_string()), + child_count: None, // Will be determined by playlist + searchable: Some("1".to_string()), + title: format!("{} - History", descriptor.display_name), + class: "object.container.playlistContainer".to_string(), + containers: vec![], + items: vec![], + } + } + + /// Get items from history playlist + #[cfg(feature = "playlist")] + async fn get_history_items( + &self, + slug: &str, + offset: usize, + count: usize, + ) -> Result> { + let playlist_id = Self::history_playlist_id(slug); + + // Get read handle for the playlist from the singleton + let manager = pmoplaylist::PlaylistManager(); + let reader = manager + .get_read_handle(&playlist_id) + .await + .map_err(|e| { + MusicSourceError::BrowseError(format!( + "Failed to get playlist {}: {}", + playlist_id, e + )) + })?; + + // Get entries from playlist + let entries = reader.get_entries(offset, count).await.map_err(|e| { + MusicSourceError::BrowseError(format!("Failed to read playlist entries: {}", e)) + })?; + + // Convert entries to Items + let mut items = Vec::new(); + for entry in entries { + if let Ok(item) = self.playlist_entry_to_item(slug, &entry).await { + items.push(item); + } + } + + Ok(items) + } + + /// Convert a playlist entry to a DIDL Item + #[cfg(feature = "playlist")] + async fn playlist_entry_to_item( + &self, + slug: &str, + entry: &pmoplaylist::PlaylistEntry, + ) -> Result { + let metadata = &entry.metadata; + + // Build audio URL from cache + let audio_url = format!( + "{}/cache/audio/{}", + self.base_url, + entry.pk + ); + + // Build item + Ok(Item { + id: format!("radio-paradise:channel:{}:history:track:{}", slug, entry.pk), + parent_id: format!("radio-paradise:channel:{}:history", slug), + restricted: Some("1".to_string()), + title: metadata.title.clone().unwrap_or_else(|| "Unknown Title".to_string()), + creator: metadata.artist.clone(), + class: "object.item.audioItem.musicTrack".to_string(), + artist: metadata.artist.clone(), + album: metadata.album.clone(), + genre: metadata.genre.clone(), + album_art: None, + album_art_pk: metadata.cover_pk.clone(), + date: metadata.year.map(|y| y.to_string()), + original_track_number: metadata.track_number.map(|n| n.to_string()), + resources: vec![Resource { + protocol_info: "http-get:*:audio/flac:*".to_string(), + bits_per_sample: metadata.bits_per_sample.map(|b| b.to_string()), + sample_frequency: metadata.sample_rate.map(|s| s.to_string()), + nr_audio_channels: Some("2".to_string()), + duration: metadata.duration.map(|d| format!("{}:{:02}:{:02}", d / 3600, (d % 3600) / 60, d % 60)), + url: audio_url, + }], + descriptions: vec![], + }) + } +} + +/// Types of object IDs in the Radio Paradise source +#[derive(Debug, Clone, PartialEq)] +enum ObjectIdType { + Root, + Channel { slug: String }, + LiveStream { slug: String }, + History { slug: String }, + HistoryTrack { slug: String, pk: String }, + Unknown, } #[async_trait] impl MusicSource for RadioParadiseSource { fn name(&self) -> &str { - "Radio Paradise (DEPRECATED)" + "Radio Paradise" } fn id(&self) -> &str { - "radio-paradise-deprecated" + "radio-paradise" } fn default_image(&self) -> &[u8] { @@ -116,55 +267,124 @@ impl MusicSource for RadioParadiseSource { async fn root_container(&self) -> Result { Ok(Container { - id: "radio-paradise-deprecated".to_string(), + id: "radio-paradise".to_string(), parent_id: "0".to_string(), restricted: Some("1".to_string()), - child_count: Some("0".to_string()), + child_count: Some("4".to_string()), // 4 channels searchable: Some("0".to_string()), - title: "Radio Paradise (DEPRECATED)".to_string(), + title: "Radio Paradise".to_string(), class: "object.container".to_string(), containers: vec![], items: vec![], }) } - async fn browse(&self, _object_id: &str) -> Result { - tracing::warn!("RadioParadiseSource::browse called but source is deprecated"); - Ok(BrowseResult::Mixed { - containers: vec![], - items: vec![], - }) + async fn browse(&self, object_id: &str) -> Result { + match Self::parse_object_id(object_id) { + ObjectIdType::Root => { + // Return the 4 channel containers + let containers: Vec = ALL_CHANNELS + .iter() + .map(|ch| self.build_channel_container(ch)) + .collect(); + + Ok(BrowseResult::Containers(containers)) + } + + ObjectIdType::Channel { slug } => { + // Return live stream item + history container + let descriptor = Self::get_channel_by_slug(&slug).ok_or_else(|| { + MusicSourceError::ObjectNotFound(format!("Unknown channel: {}", slug)) + })?; + + let live_item = self.build_live_stream_item(descriptor); + let history_container = self.build_history_container(descriptor); + + Ok(BrowseResult::Mixed { + containers: vec![history_container], + items: vec![live_item], + }) + } + + ObjectIdType::History { slug } => { + // Return items from history playlist + #[cfg(feature = "playlist")] + { + let items = self.get_history_items(&slug, 0, 100).await?; + Ok(BrowseResult::Items(items)) + } + + #[cfg(not(feature = "playlist"))] + { + let _ = slug; + Ok(BrowseResult::Items(vec![])) + } + } + + ObjectIdType::LiveStream { .. } | ObjectIdType::HistoryTrack { .. } => { + // These are leaf nodes, cannot be browsed + Err(MusicSourceError::ObjectNotFound(format!( + "Object {} is not a container", + object_id + ))) + } + + ObjectIdType::Unknown => Err(MusicSourceError::ObjectNotFound(format!( + "Unknown object ID: {}", + object_id + ))), + } } - async fn resolve_uri(&self, _object_id: &str) -> Result { - Err(MusicSourceError::SourceUnavailable( - "RadioParadiseSource is deprecated. Use RadioParadiseStreamSource instead.".to_string(), - )) + async fn resolve_uri(&self, object_id: &str) -> Result { + match Self::parse_object_id(object_id) { + ObjectIdType::LiveStream { slug } => { + // Return live stream URL + Ok(self.build_live_url(&slug)) + } + + ObjectIdType::HistoryTrack { pk, .. } => { + // Return cached audio URL + Ok(format!("{}/cache/audio/{}", self.base_url, pk)) + } + + _ => Err(MusicSourceError::ObjectNotFound(format!( + "Cannot resolve URI for object: {}", + object_id + ))), + } } fn supports_fifo(&self) -> bool { - false + // History playlists are FIFO + cfg!(feature = "playlist") } async fn append_track(&self, _track: Item) -> Result<()> { - Err(MusicSourceError::SourceUnavailable( - "RadioParadiseSource is deprecated and does not support FIFO operations.".to_string(), + // Tracks are added automatically by FlacCacheSink + Err(MusicSourceError::NotSupported( + "Tracks are automatically added to history by the streaming system".to_string(), )) } async fn remove_oldest(&self) -> Result> { + // Managed automatically by playlist FIFO Ok(None) } async fn update_id(&self) -> u32 { - 0 + *self.update_counter.read().await } async fn last_change(&self) -> Option { - None + Some(*self.last_change.read().await) } - async fn get_items(&self, _offset: usize, _count: usize) -> Result> { + async fn get_items(&self, offset: usize, count: usize) -> Result> { + // For Radio Paradise, we don't have a global FIFO + // Each channel has its own history + // Return empty for now - clients should browse specific channel histories + let _ = (offset, count); Ok(vec![]) } } diff --git a/pmoparadise/src/stream_channel.rs b/pmoparadise/src/stream_channel.rs index c5ad6f31..7e7b49b0 100644 --- a/pmoparadise/src/stream_channel.rs +++ b/pmoparadise/src/stream_channel.rs @@ -14,14 +14,18 @@ use crate::{ client::RadioParadiseClient, radio_paradise_stream_source::RadioParadiseStreamSource, }; -use anyhow::Result; -use pmoaudio::AudioPipelineNode; +use anyhow::{anyhow, Result}; +use pmoaudio::{nodes::DEFAULT_CHANNEL_SIZE, AudioPipelineNode}; use pmoaudio_ext::{ - FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, OggFlacStreamHandle, - StreamHandle, StreamingFlacSink, StreamingOggFlacSink, TrackBoundaryCoverNode, + FlacCacheSink, FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, + OggFlacStreamHandle, PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, + TrackBoundaryCoverNode, }; +use pmoaudiocache::Cache as AudioCache; use pmocovers::Cache as CoverCache; use pmoflac::EncoderOptions; +use pmoplaylist::WriteHandle; +use thiserror::Error; use tokio::io::{AsyncRead, ReadBuf}; use tokio::sync::Notify; use tokio::task::JoinHandle; @@ -43,6 +47,82 @@ impl Default for ParadiseStreamChannelConfig { } } +/// Options pour activer l'archivage/historique d'un canal. +pub struct ParadiseHistoryOptions { + pub audio_cache: Arc, + pub cover_cache: Arc, + pub playlist_id: String, + pub playlist_writer: WriteHandle, + pub collection: Option, + pub replay_max_lead_seconds: f64, +} + +/// Builder pratique pour configurer automatiquement les playlists historiques. +#[derive(Clone)] +pub struct ParadiseHistoryBuilder { + pub audio_cache: Arc, + pub cover_cache: Arc, + pub playlist_prefix: String, + pub playlist_title_prefix: Option, + pub max_history_tracks: Option, + pub collection_prefix: Option, + pub replay_max_lead_seconds: f64, +} + +impl ParadiseHistoryBuilder { + pub fn new(audio_cache: Arc, cover_cache: Arc) -> Self { + Self { + audio_cache, + cover_cache, + playlist_prefix: "radio-paradise-history".into(), + playlist_title_prefix: Some("Radio Paradise History".into()), + max_history_tracks: Some(500), + collection_prefix: Some("radio-paradise".into()), + replay_max_lead_seconds: 1.0, + } + } + + pub async fn build_for_channel( + &self, + descriptor: &ChannelDescriptor, + ) -> Result { + let playlist_id = format!("{}-{}", self.playlist_prefix, descriptor.slug); + let manager = pmoplaylist::PlaylistManager(); + let writer = manager + .get_persistent_write_handle(playlist_id.clone()) + .await?; + + if let Some(prefix) = &self.playlist_title_prefix { + let title = format!("{} - {}", prefix, descriptor.display_name); + writer.set_title(title).await?; + } + + if let Some(capacity) = self.max_history_tracks { + writer.set_capacity(Some(capacity)).await?; + } + + let collection = self + .collection_prefix + .as_ref() + .map(|prefix| format!("{}-{}", prefix, descriptor.slug)); + + Ok(ParadiseHistoryOptions { + audio_cache: self.audio_cache.clone(), + cover_cache: self.cover_cache.clone(), + playlist_id, + playlist_writer: writer, + collection, + replay_max_lead_seconds: self.replay_max_lead_seconds, + }) + } +} + +struct HistoryState { + playlist_id: String, + audio_cache: Arc, + replay_max_lead_seconds: f64, +} + #[cfg(feature = "pmoconfig")] impl ParadiseStreamChannelConfig { pub fn from_config(cfg: &pmoconfig::Config, channel: ParadiseChannelKind) -> Self { @@ -94,6 +174,7 @@ pub struct ParadiseStreamChannel { state: Arc, pipeline_handle: JoinHandle<()>, feeder_handle: JoinHandle<()>, + history: Option, } impl ParadiseStreamChannel { @@ -103,6 +184,7 @@ impl ParadiseStreamChannel { client: RadioParadiseClient, config: ParadiseStreamChannelConfig, cover_cache: Option>, + history: Option, ) -> Self { let mut source = RadioParadiseStreamSource::new(client.clone()); let block_handle = source.block_handle(); @@ -118,14 +200,47 @@ impl ParadiseStreamChannel { config.max_lead_seconds, ); + let mut downstream_children: Vec> = Vec::new(); + downstream_children.push(Box::new(flac_sink)); + downstream_children.push(Box::new(ogg_sink)); + + let mut history_state = None; + + if let Some(history_opts) = history { + let ParadiseHistoryOptions { + audio_cache, + cover_cache, + playlist_id, + playlist_writer, + collection, + replay_max_lead_seconds, + } = history_opts; + let mut cache_sink = FlacCacheSink::with_config( + audio_cache.clone(), + cover_cache, + DEFAULT_CHANNEL_SIZE, + EncoderOptions::default(), + collection, + ); + cache_sink.register_playlist(playlist_writer); + downstream_children.push(Box::new(cache_sink)); + history_state = Some(HistoryState { + playlist_id, + audio_cache, + replay_max_lead_seconds, + }); + } + if let Some(cache) = cover_cache { let mut cover_node = TrackBoundaryCoverNode::new(cache); - cover_node.register(Box::new(flac_sink)); - cover_node.register(Box::new(ogg_sink)); + for child in downstream_children { + cover_node.register(child); + } source.register(Box::new(cover_node)); } else { - source.register(Box::new(flac_sink)); - source.register(Box::new(ogg_sink)); + for child in downstream_children { + source.register(child); + } } stream_handle.set_auto_stop(false); ogg_handle.set_auto_stop(false); @@ -167,6 +282,7 @@ impl ParadiseStreamChannel { state, pipeline_handle, feeder_handle, + history: history_state, } } @@ -175,12 +291,19 @@ impl ParadiseStreamChannel { descriptor: ChannelDescriptor, config: ParadiseStreamChannelConfig, cover_cache: Option>, + history: Option, ) -> Result { let client = RadioParadiseClient::builder() .channel(descriptor.id) .build() .await?; - Ok(Self::with_client(descriptor, client, config, cover_cache)) + Ok(Self::with_client( + descriptor, + client, + config, + cover_cache, + history, + )) } /// S'abonne au flux FLAC pur. @@ -217,6 +340,78 @@ impl ParadiseStreamChannel { pub fn descriptor(&self) -> ChannelDescriptor { self.descriptor } + + /// Lance un pipeline dédié pour rejouer l'historique (FLAC pur) pour un client. + pub async fn stream_history_flac( + &self, + client_id: &str, + ) -> Result { + let history = self + .history + .as_ref() + .ok_or(HistoryStreamError::HistoryDisabled)?; + tracing::info!( + "Starting historical FLAC replay for channel {} (client_id={})", + self.descriptor.display_name, + client_id + ); + + let reader = pmoplaylist::PlaylistManager() + .get_read_handle(&history.playlist_id) + .await + .map_err(|e| HistoryStreamError::Playlist(e.to_string()))?; + let mut source = PlaylistSource::new(reader, history.audio_cache.clone()); + let (flac_sink, handle) = StreamingFlacSink::with_max_broadcast_lead( + EncoderOptions::default(), + 16, + history.replay_max_lead_seconds, + ); + source.register(Box::new(flac_sink)); + let stop_token = CancellationToken::new(); + let mut pipeline_source = source; + let stop_clone = stop_token.clone(); + let pipeline = tokio::spawn(async move { + let _ = Box::new(pipeline_source).run(stop_clone).await; + }); + let stream = handle.subscribe_flac(); + Ok(HistoryFlacStream::new(stream, stop_token, pipeline)) + } + + /// Lance un pipeline dédié pour rejouer l'historique (OGG-FLAC) pour un client. + pub async fn stream_history_ogg( + &self, + client_id: &str, + ) -> Result { + let history = self + .history + .as_ref() + .ok_or(HistoryStreamError::HistoryDisabled)?; + tracing::info!( + "Starting historical OGG replay for channel {} (client_id={})", + self.descriptor.display_name, + client_id + ); + + let reader = pmoplaylist::PlaylistManager() + .get_read_handle(&history.playlist_id) + .await + .map_err(|e| HistoryStreamError::Playlist(e.to_string()))?; + let mut source = PlaylistSource::new(reader, history.audio_cache.clone()); + let (ogg_sink, handle) = StreamingOggFlacSink::with_max_broadcast_lead( + EncoderOptions::default(), + 16, + history.replay_max_lead_seconds, + ); + source.register(Box::new(ogg_sink)); + let stop_token = CancellationToken::new(); + let mut pipeline_source = source; + let stop_clone = stop_token.clone(); + let pipeline = tokio::spawn(async move { + let _ = Box::new(pipeline_source).run(stop_clone).await; + }); + let stream = handle.subscribe(); + Ok(HistoryOggStream::new(stream, stop_token, pipeline)) + } } impl Drop for ParadiseStreamChannel { @@ -360,6 +555,96 @@ wrap_stream!(ChannelFlacStream, FlacClientStream); wrap_stream!(ChannelIcyStream, IcyClientStream); wrap_stream!(ChannelOggStream, OggFlacClientStream); +#[derive(Debug, Error)] +pub enum HistoryStreamError { + #[error("history replay not enabled for this channel")] + HistoryDisabled, + #[error("playlist error: {0}")] + Playlist(String), +} + +pub struct HistoryFlacStream { + inner: FlacClientStream, + stop_token: CancellationToken, + pipeline: Option>, +} + +impl HistoryFlacStream { + fn new( + inner: FlacClientStream, + stop_token: CancellationToken, + pipeline: JoinHandle<()>, + ) -> Self { + Self { + inner, + stop_token, + pipeline: Some(pipeline), + } + } +} + +impl AsyncRead for HistoryFlacStream { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +impl Unpin for HistoryFlacStream {} + +impl Drop for HistoryFlacStream { + fn drop(&mut self) { + self.stop_token.cancel(); + if let Some(handle) = self.pipeline.take() { + handle.abort(); + } + } +} + +pub struct HistoryOggStream { + inner: OggFlacClientStream, + stop_token: CancellationToken, + pipeline: Option>, +} + +impl HistoryOggStream { + fn new( + inner: OggFlacClientStream, + stop_token: CancellationToken, + pipeline: JoinHandle<()>, + ) -> Self { + Self { + inner, + stop_token, + pipeline: Some(pipeline), + } + } +} + +impl AsyncRead for HistoryOggStream { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.inner).poll_read(cx, buf) + } +} + +impl Unpin for HistoryOggStream {} + +impl Drop for HistoryOggStream { + fn drop(&mut self) { + self.stop_token.cancel(); + if let Some(handle) = self.pipeline.take() { + handle.abort(); + } + } +} + /// Gestionnaire multi-canaux. pub struct ParadiseChannelManager { channels: HashMap>, @@ -372,13 +657,25 @@ impl ParadiseChannelManager { pub async fn with_defaults_with_cover_cache( cover_cache: Option>, + history_builder: Option, ) -> Result { let mut map = HashMap::new(); for descriptor in ALL_CHANNELS.iter().copied() { + let history_opts = if let Some(builder) = &history_builder { + Some( + builder + .build_for_channel(&descriptor) + .await + .map_err(|e| anyhow!("Failed to init history playlist: {}", e))?, + ) + } else { + None + }; let channel = ParadiseStreamChannel::new( descriptor, ParadiseStreamChannelConfig::default(), cover_cache.clone(), + history_opts, ) .await?; map.insert(descriptor.id, Arc::new(channel)); @@ -387,7 +684,7 @@ impl ParadiseChannelManager { } pub async fn with_defaults() -> Result { - Self::with_defaults_with_cover_cache(None).await + Self::with_defaults_with_cover_cache(None, None).await } pub fn get(&self, id: u8) -> Option> {