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

Fin de la correction de l'implémentation par ChatGPT.

Browse Source
This commit is contained in:
Eric Coissac
2025-11-16 08:34:33 +01:00
parent c81a4651d6
commit 66416dafa8
20 changed files with 567 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
pmoparadise/Cargo.toml
View File

@@ -111,3 +111,8 @@ path = "examples/now_playing.rs"
name = "stream_block"
path = "examples/stream_block.rs"
required-features = ["full"]
[[example]]
name = "single_channel_server"
path = "examples/single_channel_server.rs"
required-features = ["full"]

22
pmoparadise/examples/play_and_cache.rs
View File

@@ -28,9 +28,13 @@
use pmoaudio::{AudioPipelineNode, AudioSink, TimerNode};
use pmoaudio_ext::{FlacCacheSink, PlaylistSource};
use pmoaudiocache::Cache as AudioCache;
use pmocovers::Cache as CoverCache;
use pmoaudiocache::{
new_cache_with_consolidation as new_audio_cache,
register_audio_cache as register_global_audio_cache,
};
use pmocovers::{new_cache_with_consolidation as new_cover_cache, register_cover_cache};
use pmoparadise::{RadioParadiseClient, RadioParadiseStreamSource};
use pmoplaylist::register_audio_cache as register_playlist_audio_cache;
use std::env;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
@@ -98,24 +102,20 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Créer le cache audio
let audio_cache_dir = format!("{}/audio_cache", base_dir);
std::fs::create_dir_all(&audio_cache_dir)?;
let audio_cache = Arc::new(AudioCache::new(
&audio_cache_dir,
1000, // 1000 MB limit
)?);
let audio_cache = new_audio_cache(&audio_cache_dir, 1000).await?;
tracing::debug!("Audio cache initialized at: {}", audio_cache_dir);
// Créer le cache de covers
let cover_cache_dir = format!("{}/cover_cache", base_dir);
std::fs::create_dir_all(&cover_cache_dir)?;
let cover_cache = Arc::new(CoverCache::new(
&cover_cache_dir,
100, // 100 MB limit
)?);
let cover_cache = new_cover_cache(&cover_cache_dir, 100).await?;
tracing::debug!("Cover cache initialized at: {}", cover_cache_dir);
// Enregistrer le cache audio dans pmoplaylist
// (requis par pmoplaylist pour valider les pks)
pmoplaylist::register_audio_cache(audio_cache.clone());
register_global_audio_cache(audio_cache.clone());
register_playlist_audio_cache(audio_cache.clone());
register_cover_cache(cover_cache.clone());
tracing::debug!("Audio cache registered in pmoplaylist");
// Utiliser le gestionnaire de playlist singleton

17
pmoparadise/examples/serve_channels.rs
View File

@@ -18,10 +18,13 @@ use axum::{
routing::get,
Json, Router,
};
use pmoaudiocache::new_cache as new_audio_cache;
use pmocovers::new_cache as new_cover_cache;
use pmoaudiocache::{
new_cache_with_consolidation as new_audio_cache,
register_audio_cache as register_global_audio_cache,
};
use pmocovers::{new_cache_with_consolidation as new_cover_cache, register_cover_cache};
use pmoparadise::{channels::ALL_CHANNELS, ParadiseChannelManager, ParadiseHistoryBuilder};
use pmoplaylist::register_audio_cache;
use pmoplaylist::register_audio_cache as register_playlist_audio_cache;
use pmoserver::{init_logging, ServerBuilder};
use tokio_util::io::ReaderStream;
use tracing::{error, info};
@@ -41,9 +44,11 @@ async fn main() -> anyhow::Result<()> {
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 cover_cache = new_cover_cache(cover_cache_dir, 500).await?;
let audio_cache = new_audio_cache(audio_cache_dir, 1000).await?;
register_global_audio_cache(audio_cache.clone());
register_playlist_audio_cache(audio_cache.clone());
register_cover_cache(cover_cache.clone());
let _playlist_manager = pmoplaylist::PlaylistManager();
let history_builder = ParadiseHistoryBuilder {

122
pmoparadise/examples/single_channel_server.rs Normal file
View File

@@ -0,0 +1,122 @@
//! Simple web server that exposes one Radio Paradise channel over HTTP.
//!
//! Usage:
//! ```bash
//! cargo run --example single_channel_server --features full -- main
//! ```
//! Valid arguments are either the slug (`main`, `mellow`, `rock`, `eclectic`) or
//! the numeric channel id (`0`..`3`). When no argument is provided, the example
//! defaults to the “main” mix.
use axum::{
body::Body, extract::State, http::StatusCode, response::Response, routing::get, Router,
};
use pmoaudiocache::{
new_cache_with_consolidation as new_audio_cache,
register_audio_cache as register_global_audio_cache,
};
use pmocovers::{new_cache_with_consolidation as new_cover_cache, register_cover_cache};
use pmoparadise::{
channels::{ChannelDescriptor, ALL_CHANNELS},
ParadiseHistoryBuilder, ParadiseStreamChannel, ParadiseStreamChannelConfig,
};
use pmoplaylist::register_audio_cache as register_playlist_audio_cache;
use std::{fs, net::SocketAddr, sync::Arc};
use tokio::net::TcpListener;
use tokio_util::io::ReaderStream;
use tracing::info;
#[derive(Clone)]
struct AppState {
channel: Arc<ParadiseStreamChannel>,
descriptor: ChannelDescriptor,
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
tracing_subscriber::fmt().with_env_filter("info").init();
let descriptor = pick_descriptor(std::env::args().nth(1))?;
info!(
"Selected Radio Paradise channel: {} ({})",
descriptor.display_name, descriptor.slug
);
// Prepare caches under ./cache/single-channel
let cache_root = "./cache/single-channel";
let audio_cache_dir = format!("{}/audio", cache_root);
let cover_cache_dir = format!("{}/covers", cache_root);
fs::create_dir_all(&audio_cache_dir)?;
fs::create_dir_all(&cover_cache_dir)?;
let audio_cache = new_audio_cache(&audio_cache_dir, 1000).await?;
let cover_cache = new_cover_cache(&cover_cache_dir, 200).await?;
register_global_audio_cache(audio_cache.clone());
register_playlist_audio_cache(audio_cache.clone());
register_cover_cache(cover_cache.clone());
let mut history_builder = ParadiseHistoryBuilder::new(audio_cache.clone(), cover_cache.clone());
history_builder.playlist_prefix = format!("single-channel-history-{}", descriptor.slug);
history_builder.collection_prefix = Some(format!("single-channel-{}", descriptor.slug));
let history_opts = history_builder.build_for_channel(&descriptor).await?;
let channel = Arc::new(
ParadiseStreamChannel::new(
descriptor,
ParadiseStreamChannelConfig::default(),
Some(cover_cache),
Some(history_opts),
)
.await?,
);
let state = AppState {
channel,
descriptor,
};
let app = Router::new()
.route("/stream/flac", get(stream_flac))
.with_state(state);
let addr: SocketAddr = ([0, 0, 0, 0], 8080).into();
info!("HTTP server listening on http://{addr}/stream/flac");
info!("Connect with a FLAC player (e.g. ffplay http://localhost:8080/stream/flac)");
let listener = TcpListener::bind(addr).await?;
axum::serve(listener, app.into_make_service()).await?;
Ok(())
}
async fn stream_flac(State(state): State<AppState>) -> Result<Response, StatusCode> {
let stream = state.channel.subscribe_flac();
let body = Body::from_stream(ReaderStream::new(stream));
Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "audio/flac")
.header(
"X-PMO-Channel",
format!(
"{} ({})",
state.descriptor.display_name, state.descriptor.slug
),
)
.body(body)
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
}
fn pick_descriptor(arg: Option<String>) -> anyhow::Result<ChannelDescriptor> {
if let Some(token) = arg {
if let Some(desc) = ALL_CHANNELS.iter().find(|c| c.slug == token) {
return Ok(*desc);
}
if let Ok(id) = token.parse::<u8>() {
if let Some(desc) = ALL_CHANNELS.iter().find(|c| c.id == id) {
return Ok(*desc);
}
}
anyhow::bail!("Unknown channel identifier: {token}");
}
Ok(ALL_CHANNELS[0])
}

14
pmoparadise/src/models.rs
View File

@@ -247,6 +247,10 @@ pub struct Block {
#[serde(default)]
pub image_base: Option<String>,
/// Scheduled start time for this block (Unix timestamp in milliseconds, UTC)
#[serde(default)]
pub sched_time_millis: Option<u64>,
/// Map of song index (as string) to Song metadata
/// Keys are "0", "1", "2", etc.
#[serde(default)]
@@ -258,6 +262,16 @@ pub struct Block {
}
impl Block {
/// Scheduled start time in milliseconds if available.
pub fn start_time_millis(&self) -> Option<u64> {
if let Some(ts) = self.sched_time_millis {
return Some(ts);
}
self.songs_ordered()
.into_iter()
.find_map(|(_, song)| song.sched_time_millis)
}
/// Get songs in order by index
pub fn songs_ordered(&self) -> Vec<(usize, &Song)> {
let mut songs: Vec<_> = self

170
pmoparadise/src/playlist_feeder.rs
View File

@@ -3,19 +3,94 @@
//! Architecture simplifiée utilisant les URLs gapless individuelles au lieu du bloc FLAC entier.
use crate::{client::RadioParadiseClient, models::EventId};
use anyhow::Result;
use pmoaudiocache::Cache as AudioCache;
use pmocovers::Cache as CoversCache;
use pmoplaylist::{PlaylistManager, ReadHandle, WriteHandle};
use std::{
collections::VecDeque,
collections::{HashMap, VecDeque},
sync::Arc,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use tokio::sync::Notify;
use anyhow::Result;
/// Signal de fin de blocs
pub const END_OF_BLOCKS_SIGNAL: EventId = EventId::MAX;
const RECENT_BLOCKS_CACHE_SIZE: usize = 10;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum BlockStatus {
Pending,
InProgress,
Done,
}
struct RecentBlocks {
states: HashMap<EventId, BlockStatus>,
order: VecDeque<EventId>,
capacity: usize,
}
impl RecentBlocks {
fn new(capacity: usize) -> Self {
Self {
states: HashMap::new(),
order: VecDeque::new(),
capacity,
}
}
fn try_enqueue(&mut self, event_id: EventId) -> bool {
match self.states.get(&event_id) {
Some(_) => false,
None => {
self.order.push_back(event_id);
self.states.insert(event_id, BlockStatus::Pending);
self.evict_old_done();
true
}
}
}
fn mark_in_progress(&mut self, event_id: EventId) {
if let Some(state) = self.states.get_mut(&event_id) {
*state = BlockStatus::InProgress;
} else {
self.order.push_back(event_id);
self.states.insert(event_id, BlockStatus::InProgress);
}
self.evict_old_done();
}
fn mark_done(&mut self, event_id: EventId) {
if let Some(state) = self.states.get_mut(&event_id) {
*state = BlockStatus::Done;
} else {
self.order.push_back(event_id);
self.states.insert(event_id, BlockStatus::Done);
}
self.evict_old_done();
}
fn purge(&mut self, event_id: EventId) {
self.states.remove(&event_id);
}
fn evict_old_done(&mut self) {
while self.order.len() > self.capacity {
let Some(front) = self.order.front().copied() else {
break;
};
match self.states.get(&front) {
Some(BlockStatus::Done) | None => {
self.order.pop_front();
self.states.remove(&front);
}
Some(_) => break,
}
}
}
}
/// Feeder qui télécharge les blocs RP et alimente une playlist
pub struct RadioParadisePlaylistFeeder {
@@ -26,6 +101,7 @@ pub struct RadioParadisePlaylistFeeder {
block_queue: Arc<tokio::sync::Mutex<VecDeque<EventId>>>,
notify: Arc<Notify>,
collection: Option<String>,
recent_blocks: tokio::sync::Mutex<RecentBlocks>,
}
impl RadioParadisePlaylistFeeder {
@@ -38,7 +114,9 @@ impl RadioParadisePlaylistFeeder {
collection: Option<String>,
) -> Result<(Self, ReadHandle)> {
let manager = PlaylistManager::get();
let write_handle = manager.create_persistent_playlist(playlist_id.clone()).await?;
let write_handle = manager
.create_persistent_playlist(playlist_id.clone())
.await?;
let read_handle = manager.get_read_handle(&playlist_id).await?;
Ok((
@@ -50,6 +128,7 @@ impl RadioParadisePlaylistFeeder {
block_queue: Arc::new(tokio::sync::Mutex::new(VecDeque::new())),
notify: Arc::new(Notify::new()),
collection,
recent_blocks: tokio::sync::Mutex::new(RecentBlocks::new(RECENT_BLOCKS_CACHE_SIZE)),
},
read_handle,
))
@@ -57,6 +136,17 @@ impl RadioParadisePlaylistFeeder {
/// Enqueue un bloc pour traitement
pub async fn push_block_id(&self, event_id: EventId) {
{
let mut recent = self.recent_blocks.lock().await;
if !recent.try_enqueue(event_id) {
tracing::debug!(
"RadioParadisePlaylistFeeder: Ignoring duplicate enqueue for block {}",
event_id
);
return;
}
}
{
let mut queue = self.block_queue.lock().await;
queue.push_back(event_id);
@@ -64,6 +154,21 @@ impl RadioParadisePlaylistFeeder {
self.notify.notify_one();
}
async fn mark_in_progress(&self, event_id: EventId) {
let mut recent = self.recent_blocks.lock().await;
recent.mark_in_progress(event_id);
}
async fn mark_done(&self, event_id: EventId) {
let mut recent = self.recent_blocks.lock().await;
recent.mark_done(event_id);
}
async fn purge_block_state(&self, event_id: EventId) {
let mut recent = self.recent_blocks.lock().await;
recent.purge(event_id);
}
/// Boucle principale de traitement (à exécuter dans une tâche tokio)
pub async fn run(self: Arc<Self>) -> Result<()> {
loop {
@@ -73,7 +178,9 @@ impl RadioParadisePlaylistFeeder {
let mut queue = self.block_queue.lock().await;
if let Some(id) = queue.pop_front() {
if id == END_OF_BLOCKS_SIGNAL {
tracing::info!("RadioParadisePlaylistFeeder: END_OF_BLOCKS_SIGNAL received");
tracing::info!(
"RadioParadisePlaylistFeeder: END_OF_BLOCKS_SIGNAL received"
);
return Ok(());
}
break id;
@@ -82,9 +189,22 @@ impl RadioParadisePlaylistFeeder {
self.notify.notified().await;
};
self.mark_in_progress(event_id).await;
// Traiter le bloc
if let Err(e) = self.process_block(event_id).await {
tracing::error!("RadioParadisePlaylistFeeder: Failed to process block {}: {}", event_id, e);
tracing::error!(
"RadioParadisePlaylistFeeder: Failed to process block {}: {}",
event_id,
e
);
self.purge_block_state(event_id).await;
tracing::debug!(
"RadioParadisePlaylistFeeder: Cleared block {} state after error",
event_id
);
} else {
self.mark_done(event_id).await;
}
}
}
@@ -97,9 +217,7 @@ impl RadioParadisePlaylistFeeder {
let block = self.client.get_block(Some(event_id)).await?;
// 2. Timestamp actuel
let now_ms = SystemTime::now()
.duration_since(UNIX_EPOCH)?
.as_millis() as u64;
let now_ms = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as u64;
// 3. Filtrer les chansons encore en lecture ou à venir
let songs = block.songs_ordered();
@@ -109,21 +227,28 @@ impl RadioParadisePlaylistFeeder {
if !song.is_still_playing(now_ms) {
tracing::debug!(
"RadioParadisePlaylistFeeder: Skipping finished song {} - {} (ended at {})",
idx, song.title, song.sched_end_time_ms().unwrap_or(0)
idx,
song.title,
song.sched_end_time_ms().unwrap_or(0)
);
continue;
}
// 4. Télécharger la chanson
let gapless_url = song.gapless_url.as_ref()
let gapless_url = song
.gapless_url
.as_ref()
.ok_or_else(|| anyhow::anyhow!("Missing gapless_url for song {}", idx))?;
tracing::info!(
"RadioParadisePlaylistFeeder: Downloading song {} - {} by {}",
idx, song.title, song.artist
idx,
song.title,
song.artist
);
let pk = self.audio_cache
let pk = self
.audio_cache
.add_from_url(gapless_url, self.collection.as_deref())
.await?;
@@ -131,7 +256,8 @@ impl RadioParadisePlaylistFeeder {
self.save_metadata(&pk, song, &block).await?;
// 6. Calculer le TTL
let sched_end = song.sched_end_time_ms()
let sched_end = song
.sched_end_time_ms()
.ok_or_else(|| anyhow::anyhow!("Cannot calculate TTL without sched_time_millis"))?;
let ttl_ms = sched_end.saturating_sub(now_ms);
let ttl = Duration::from_millis(ttl_ms);
@@ -141,7 +267,9 @@ impl RadioParadisePlaylistFeeder {
tracing::info!(
"RadioParadisePlaylistFeeder: Added {} to playlist (pk={}, ttl={}s)",
song.title, pk, ttl.as_secs()
song.title,
pk,
ttl.as_secs()
);
processed += 1;
@@ -149,7 +277,8 @@ impl RadioParadisePlaylistFeeder {
tracing::info!(
"RadioParadisePlaylistFeeder: Processed block {} - added {} songs to playlist",
event_id, processed
event_id,
processed
);
Ok(())
@@ -183,10 +312,17 @@ impl RadioParadisePlaylistFeeder {
meta.set_cover_url(Some(cover_url.clone())).await?;
// Télécharger la cover
match self.covers_cache.add_from_url(&cover_url, self.collection.as_deref()).await {
match self
.covers_cache
.add_from_url(&cover_url, self.collection.as_deref())
.await
{
Ok(cover_pk) => {
meta.set_cover_pk(Some(cover_pk)).await?;
tracing::debug!("RadioParadisePlaylistFeeder: Cached cover for {}", song.title);
tracing::debug!(
"RadioParadisePlaylistFeeder: Cached cover for {}",
song.title
);
}
Err(e) => {
tracing::warn!("RadioParadisePlaylistFeeder: Failed to cache cover: {}", e);

27
pmoparadise/src/source.rs
View File

@@ -170,15 +170,9 @@ impl RadioParadiseSource {
// 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
))
})?;
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| {
@@ -206,18 +200,17 @@ impl RadioParadiseSource {
let metadata = &entry.metadata;
// Build audio URL from cache
let audio_url = format!(
"{}/cache/audio/{}",
self.base_url,
entry.pk
);
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()),
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(),
@@ -232,7 +225,9 @@ impl RadioParadiseSource {
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)),
duration: metadata
.duration
.map(|d| format!("{}:{:02}:{:02}", d / 3600, (d % 3600) / 60, d % 60)),
url: audio_url,
}],
descriptions: vec![],

103
pmoparadise/src/stream_channel.rs
View File

@@ -12,23 +12,23 @@ use std::{
Arc,
},
task::{Context, Poll},
time::Duration,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use crate::{
channels::{ChannelDescriptor, ParadiseChannelKind, ALL_CHANNELS},
client::RadioParadiseClient,
models::Block,
playlist_feeder::RadioParadisePlaylistFeeder,
};
use anyhow::{anyhow, Result};
use pmoaudio::AudioPipelineNode;
use pmoaudio_ext::{
FlacClientStream, IcyClientStream, MetadataSnapshot, OggFlacClientStream, OggFlacStreamHandle,
PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink,
TrackBoundaryCoverNode,
PlaylistSource, StreamHandle, StreamingFlacSink, StreamingOggFlacSink, TrackBoundaryCoverNode,
};
use pmoaudiocache::Cache as AudioCache;
use pmocovers::Cache as CoverCache;
use pmoaudiocache::{get_audio_cache, Cache as AudioCache};
use pmocovers::{get_cover_cache, Cache as CoverCache};
use pmoflac::EncoderOptions;
use pmoplaylist::PlaylistManager;
use thiserror::Error;
@@ -110,6 +110,16 @@ impl ParadiseHistoryBuilder {
}
}
impl Default for ParadiseHistoryBuilder {
fn default() -> Self {
let audio_cache = get_audio_cache()
.expect("pmoaudiocache::register_audio_cache must be called before using ParadiseHistoryBuilder::default()");
let cover_cache = get_cover_cache()
.expect("pmocovers::register_cover_cache must be called before using ParadiseHistoryBuilder::default()");
Self::new(audio_cache, cover_cache)
}
}
#[cfg(feature = "pmoconfig")]
impl ParadiseStreamChannelConfig {
pub fn from_config(cfg: &pmoconfig::Config, channel: ParadiseChannelKind) -> Self {
@@ -174,6 +184,9 @@ impl ParadiseStreamChannel {
cover_cache: Option<Arc<CoverCache>>,
history: Option<ParadiseHistoryOptions>,
) -> Result<Self> {
let cover_cache = cover_cache
.or_else(|| history.as_ref().map(|opts| opts.cover_cache.clone()))
.or_else(|| get_cover_cache());
let manager = PlaylistManager::get();
// 1. Créer la playlist live pour ce canal
@@ -189,7 +202,9 @@ impl ParadiseStreamChannel {
.await?
} else {
// Pas d'historique, on a besoin quand même d'un cache audio basique
return Err(anyhow!("History options required for now (audio cache needed)"));
return Err(anyhow!(
"History options required for now (audio cache needed)"
));
};
let feeder = Arc::new(feeder);
@@ -317,13 +332,7 @@ impl ParadiseStreamChannel {
.channel(descriptor.id)
.build()
.await?;
Self::with_client(
descriptor,
client,
config,
cover_cache,
history,
).await
Self::with_client(descriptor, client, config, cover_cache, history).await
}
/// S'abonne au flux FLAC pur.
@@ -458,6 +467,9 @@ impl Drop for ParadiseStreamChannel {
}
}
const MAX_BLOCK_LEAD: Duration = Duration::from_secs(3600);
const BLOCK_LEAD_CHECK_CHUNK: Duration = Duration::from_secs(300);
struct ChannelState {
descriptor: ChannelDescriptor,
config: ParadiseStreamChannelConfig,
@@ -473,6 +485,24 @@ struct ChannelState {
}
impl ChannelState {
fn current_unix_millis() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0)
}
fn block_lead_delay(&self, block: &Block) -> Option<Duration> {
let start = block.start_time_millis()?;
let now = Self::current_unix_millis();
let max_lead_ms = MAX_BLOCK_LEAD.as_millis() as u64;
if start <= now + max_lead_ms {
None
} else {
Some(Duration::from_millis(start - now - max_lead_ms))
}
}
fn on_client_added(&self) {
if self.active_clients.fetch_add(1, Ordering::SeqCst) == 0 {
self.activity_notify.notify_one();
@@ -493,9 +523,38 @@ impl ChannelState {
true
}
async fn wait_until_block_ready(&self, block: &Block) -> BlockReadiness {
loop {
if self.stop_token.is_cancelled() {
return BlockReadiness::Stopped;
}
if self.active_clients.load(Ordering::SeqCst) == 0 {
return BlockReadiness::NoClients;
}
if let Some(delay) = self.block_lead_delay(block) {
let sleep_for = delay.min(BLOCK_LEAD_CHECK_CHUNK);
let lead_secs = delay.as_secs_f64();
info!(
"Block {} scheduled too far in the future ({:.1} min). Sleeping {:?} before retrying.",
block.event,
lead_secs / 60.0,
sleep_for
);
tokio::select! {
_ = self.stop_token.cancelled() => return BlockReadiness::Stopped,
_ = tokio::time::sleep(sleep_for) => {},
}
continue;
}
return BlockReadiness::Ready;
}
}
async fn run_scheduler(self: Arc<Self>) {
let mut backoff = Duration::from_secs(5);
loop {
'scheduler: loop {
if self.stop_token.is_cancelled() {
break;
}
@@ -506,6 +565,11 @@ impl ChannelState {
match self.client.get_block(None).await {
Ok(block) => {
match self.wait_until_block_ready(&block).await {
BlockReadiness::Ready => {}
BlockReadiness::NoClients => continue,
BlockReadiness::Stopped => break,
}
info!(
"Channel {} streaming block {}",
self.descriptor.display_name, block.event
@@ -524,6 +588,11 @@ impl ChannelState {
match self.client.get_block(Some(next_event)).await {
Ok(next_block) => {
match self.wait_until_block_ready(&next_block).await {
BlockReadiness::Ready => {}
BlockReadiness::NoClients => break,
BlockReadiness::Stopped => break 'scheduler,
}
self.feeder.push_block_id(next_block.event).await;
next_event = next_block.end_event;
backoff = Duration::from_secs(5);
@@ -558,6 +627,12 @@ impl ChannelState {
}
}
enum BlockReadiness {
Ready,
NoClients,
Stopped,
}
macro_rules! wrap_stream {
($name:ident, $inner:ty) => {
pub struct $name {