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

Nouvelle tentative de broadcast avec considération d'un temps d'expiration.

Browse Source
This commit is contained in:
Eric Coissac
2025-11-15 12:21:46 +01:00
parent 1c2d30cbe9
commit 4e27255305
5 changed files with 628 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -16,6 +16,9 @@ mod broadcast_pacing;
#[cfg(feature = "http-stream")]
mod flac_frame_utils;
#[cfg(feature = "http-stream")]
mod timed_broadcast;
#[cfg(feature = "http-stream")]
mod streaming_flac_sink;

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

@@ -19,7 +19,7 @@
//! ↓
//! [Broadcaster Task]
//! ↓
//! broadcast::channel<Bytes> (FLAC bytes)
//! timed_broadcast::channel<Bytes> (FLAC bytes)
//! ↓
//! Multiple clients via StreamHandle::subscribe()
//! ├─ FLAC pure (for standard renderers)
@@ -62,7 +62,11 @@ use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
use super::{broadcast_pacing::BroadcastPacer, flac_frame_utils};
use super::{
broadcast_pacing::BroadcastPacer,
flac_frame_utils,
timed_broadcast::{self, TimedPacket, TryRecvError},
};
use async_trait::async_trait;
use bytes::Bytes;
use pmoaudio::{
@@ -73,7 +77,7 @@ use pmoaudio::{
use pmoflac::{encode_flac_stream, EncoderOptions, FlacEncodedStream, PcmFormat};
use pmometadata::TrackMetadata;
use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
use tokio::sync::{broadcast, mpsc, RwLock};
use tokio::sync::{mpsc, RwLock};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, trace, warn};
@@ -160,7 +164,7 @@ pub struct MetadataSnapshot {
#[derive(Clone)]
pub struct StreamHandle {
/// Broadcast sender for FLAC bytes (pure mode)
flac_broadcast: broadcast::Sender<Bytes>,
flac_broadcast: timed_broadcast::Sender<Bytes>,
/// Current track metadata (read-only for consumers)
metadata: Arc<RwLock<MetadataSnapshot>>,
@@ -189,6 +193,7 @@ impl StreamHandle {
finished: false,
handle: self.clone(),
state: FlacStreamState::SendingHeader,
current_epoch: 0,
}
}
@@ -220,6 +225,7 @@ impl StreamHandle {
finished: false,
handle: self.clone(),
state: FlacStreamState::SendingHeader,
current_epoch: 0,
}
}
@@ -247,11 +253,18 @@ enum FlacStreamState {
/// Pure FLAC client stream (implements AsyncRead).
pub struct FlacClientStream {
rx: broadcast::Receiver<Bytes>,
rx: timed_broadcast::Receiver<Bytes>,
buffer: VecDeque<u8>,
finished: bool,
handle: StreamHandle,
state: FlacStreamState,
current_epoch: u64,
}
impl FlacClientStream {
pub fn current_epoch(&self) -> u64 {
self.current_epoch
}
}
impl AsyncRead for FlacClientStream {
@@ -302,10 +315,11 @@ impl AsyncRead for FlacClientStream {
// Try to receive more data
match self.rx.try_recv() {
Ok(bytes) => {
self.buffer.extend(bytes.iter());
Ok(packet) => {
self.current_epoch = packet.epoch;
self.buffer.extend(packet.payload.iter());
}
Err(broadcast::error::TryRecvError::Empty) => {
Err(TryRecvError::Empty) => {
// No data available right now.
// Schedule a wakeup after a small delay to avoid busy-loop polling.
let waker = cx.waker().clone();
@@ -315,11 +329,11 @@ impl AsyncRead for FlacClientStream {
});
return Poll::Pending;
}
Err(broadcast::error::TryRecvError::Lagged(skipped)) => {
Err(TryRecvError::Lagged(skipped)) => {
warn!("FLAC client lagged, skipped {} messages", skipped);
// Continue to try receiving again
}
Err(broadcast::error::TryRecvError::Closed) => {
Err(TryRecvError::Closed) => {
self.finished = true;
return Poll::Ready(Ok(()));
}
@@ -345,7 +359,7 @@ impl Drop for FlacClientStream {
/// This stream injects ICY metadata blocks at regular intervals,
/// allowing clients to display "Now Playing" information.
pub struct IcyClientStream {
rx: broadcast::Receiver<Bytes>,
rx: timed_broadcast::Receiver<Bytes>,
metadata: Arc<RwLock<MetadataSnapshot>>,
metaint: usize,
byte_count: usize,
@@ -355,6 +369,13 @@ pub struct IcyClientStream {
finished: bool,
handle: StreamHandle,
state: FlacStreamState,
current_epoch: u64,
}
impl IcyClientStream {
pub fn current_epoch(&self) -> u64 {
self.current_epoch
}
}
impl IcyClientStream {
@@ -491,22 +512,23 @@ impl AsyncRead for IcyClientStream {
// Try to receive audio data
match self.rx.try_recv() {
Ok(bytes) => {
Ok(packet) => {
self.current_epoch = packet.epoch;
// Calculate how many bytes until next metadata block
let until_metadata = self.metaint - (self.byte_count % self.metaint);
let to_buffer = bytes.len().min(until_metadata);
let to_buffer = packet.payload.len().min(until_metadata);
self.buffer.extend(bytes[..to_buffer].iter());
self.buffer.extend(packet.payload[..to_buffer].iter());
self.byte_count += to_buffer;
// If we have more data, we'll process it in the next iteration
if to_buffer < bytes.len() {
if to_buffer < packet.payload.len() {
// Save remaining for next iteration
// For now, we'll just drop it and get it again
// TODO: Improve this
}
}
Err(broadcast::error::TryRecvError::Empty) => {
Err(TryRecvError::Empty) => {
// No data available right now.
// Schedule a wakeup after a small delay to avoid busy-loop polling.
let waker = cx.waker().clone();
@@ -516,10 +538,10 @@ impl AsyncRead for IcyClientStream {
});
return Poll::Pending;
}
Err(broadcast::error::TryRecvError::Lagged(skipped)) => {
Err(TryRecvError::Lagged(skipped)) => {
warn!("ICY client lagged, skipped {} messages", skipped);
}
Err(broadcast::error::TryRecvError::Closed) => {
Err(TryRecvError::Closed) => {
self.finished = true;
return Poll::Ready(Ok(()));
}
@@ -552,7 +574,7 @@ struct StreamingFlacSinkLogic {
pcm_tx: mpsc::Sender<PcmChunk>,
pcm_rx: Option<mpsc::Receiver<PcmChunk>>,
metadata: Arc<RwLock<MetadataSnapshot>>,
flac_broadcast: broadcast::Sender<Bytes>,
flac_broadcast: timed_broadcast::Sender<Bytes>,
flac_header: Arc<RwLock<Option<Bytes>>>,
encoder_state: Option<EncoderState>,
sample_rate: Option<u32>,
@@ -749,6 +771,11 @@ impl NodeLogic for StreamingFlacSinkLogic {
break;
}
SyncMarker::TopZeroSync => {
self.flac_broadcast.mark_top_zero();
trace!("TopZeroSync propagated to FLAC broadcast");
}
_ => {
trace!("Received other sync marker");
}
@@ -781,7 +808,7 @@ impl NodeLogic for StreamingFlacSinkLogic {
/// Ensures data is sent at FLAC frame boundaries to prevent sync errors in strict decoders like FFPlay.
async fn broadcast_flac_stream(
mut flac_stream: FlacEncodedStream,
broadcast_tx: broadcast::Sender<Bytes>,
broadcast_tx: timed_broadcast::Sender<Bytes>,
header_cache: Arc<RwLock<Option<Bytes>>>,
current_timestamp: Arc<RwLock<f64>>,
broadcast_max_lead_time: f64,
@@ -813,7 +840,11 @@ async fn broadcast_flac_stream(
// EOF - send any remaining data
if !accumulator.is_empty() {
let bytes = Bytes::from(std::mem::take(&mut accumulator));
let _ = broadcast_tx.send(bytes);
let audio_ts = *current_timestamp.read().await;
if broadcast_tx.send(bytes.clone(), audio_ts).await.is_err() {
trace!("Broadcast closed before sending final FLAC data");
break;
}
}
info!("FLAC encoder stream ended, total bytes: {}", total_bytes);
break;
@@ -931,15 +962,20 @@ async fn broadcast_flac_stream(
}
let num_receivers = broadcast_tx.receiver_count();
if let Err(e) = broadcast_tx.send(bytes.clone()) {
// No receivers, but that's okay - clients may not be connected yet
trace!("No active receivers for FLAC broadcast: {}", e);
} else if num_receivers > 0 {
trace!(
"Broadcasted {} bytes to {} receivers",
bytes.len(),
num_receivers
);
match broadcast_tx.send(bytes.clone(), audio_timestamp).await {
Ok(_) => {
if num_receivers > 0 {
trace!(
"Broadcasted {} bytes to {} receivers",
bytes.len(),
num_receivers
);
}
}
Err(_) => {
trace!("No active receivers for FLAC broadcast, terminating");
return Ok(());
}
}
}
}
@@ -1017,7 +1053,7 @@ impl StreamingFlacSink {
);
// Broadcast channel for FLAC bytes
let (flac_broadcast, _) = broadcast::channel(broadcast_capacity);
let (flac_broadcast, _) = timed_broadcast::channel(broadcast_capacity);
// FLAC header cache
let flac_header = Arc::new(RwLock::new(None));

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

@@ -21,7 +21,7 @@
//! ↓
//! [OGG Wrapper Task] - wraps FLAC frames in OGG pages
//! ↓
//! broadcast::channel<Bytes> (OGG-FLAC bytes)
//! timed_broadcast::channel<Bytes> (OGG-FLAC bytes)
//! ↓
//! Multiple HTTP clients
//! ```
@@ -53,7 +53,11 @@ use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};
use super::{broadcast_pacing::BroadcastPacer, flac_frame_utils};
use super::{
broadcast_pacing::BroadcastPacer,
flac_frame_utils,
timed_broadcast::{self, TimedPacket, TryRecvError},
};
use async_trait::async_trait;
use bytes::Bytes;
use pmoaudio::{
@@ -64,7 +68,7 @@ use pmoaudio::{
use pmoflac::{encode_flac_stream, EncoderOptions, FlacEncodedStream, PcmFormat};
use pmometadata::TrackMetadata;
use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
use tokio::sync::{broadcast, mpsc, RwLock};
use tokio::sync::{mpsc, RwLock};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, trace, warn};
@@ -97,7 +101,7 @@ pub use super::streaming_flac_sink::MetadataSnapshot;
#[derive(Clone)]
pub struct OggFlacStreamHandle {
/// Broadcast sender for OGG-FLAC bytes
ogg_broadcast: broadcast::Sender<Bytes>,
ogg_broadcast: timed_broadcast::Sender<Bytes>,
/// Current track metadata
metadata: Arc<RwLock<MetadataSnapshot>>,
@@ -126,6 +130,7 @@ impl OggFlacStreamHandle {
finished: false,
handle: self.clone(),
state: OggFlacStreamState::SendingHeader,
current_epoch: 0,
}
}
@@ -148,11 +153,19 @@ enum OggFlacStreamState {
/// OGG-FLAC client stream (implements AsyncRead).
pub struct OggFlacClientStream {
rx: broadcast::Receiver<Bytes>,
rx: timed_broadcast::Receiver<Bytes>,
buffer: VecDeque<u8>,
finished: bool,
handle: OggFlacStreamHandle,
state: OggFlacStreamState,
current_epoch: u64,
}
impl OggFlacClientStream {
/// Dernier epoch observé (incrémenté à chaque TopZeroSync).
pub fn current_epoch(&self) -> u64 {
self.current_epoch
}
}
impl AsyncRead for OggFlacClientStream {
@@ -203,10 +216,11 @@ impl AsyncRead for OggFlacClientStream {
// Try to receive more data
match self.rx.try_recv() {
Ok(bytes) => {
self.buffer.extend(bytes.iter());
Ok(packet) => {
self.current_epoch = packet.epoch;
self.buffer.extend(packet.payload.iter());
}
Err(broadcast::error::TryRecvError::Empty) => {
Err(TryRecvError::Empty) => {
// No data available right now.
// Schedule a wakeup after a small delay to avoid busy-loop polling.
let waker = cx.waker().clone();
@@ -216,10 +230,10 @@ impl AsyncRead for OggFlacClientStream {
});
return Poll::Pending;
}
Err(broadcast::error::TryRecvError::Lagged(skipped)) => {
Err(TryRecvError::Lagged(skipped)) => {
warn!("OGG-FLAC client lagged, skipped {} messages", skipped);
}
Err(broadcast::error::TryRecvError::Closed) => {
Err(TryRecvError::Closed) => {
self.finished = true;
return Poll::Ready(Ok(()));
}
@@ -252,7 +266,7 @@ struct StreamingOggFlacSinkLogic {
pcm_tx: mpsc::Sender<PcmChunk>,
pcm_rx: Option<mpsc::Receiver<PcmChunk>>,
metadata: Arc<RwLock<MetadataSnapshot>>,
ogg_broadcast: broadcast::Sender<Bytes>,
ogg_broadcast: timed_broadcast::Sender<Bytes>,
ogg_header: Arc<RwLock<Option<Bytes>>>,
encoder_state: Option<EncoderState>,
sample_rate: Option<u32>,
@@ -440,6 +454,11 @@ impl NodeLogic for StreamingOggFlacSinkLogic {
break;
}
SyncMarker::TopZeroSync => {
self.ogg_broadcast.mark_top_zero();
trace!("TopZeroSync propagated to timed broadcast");
}
_ => {
trace!("Received other sync marker");
}
@@ -521,7 +540,7 @@ impl StreamingOggFlacSink {
broadcast_capacity,
broadcast_max_lead_time
);
let (ogg_broadcast, _) = broadcast::channel(broadcast_capacity);
let (ogg_broadcast, _) = timed_broadcast::channel(broadcast_capacity);
// OGG-FLAC header cache
let ogg_header = Arc::new(RwLock::new(None));
@@ -751,7 +770,7 @@ fn chunk_to_pcm_bytes(chunk: &AudioChunk, bits_per_sample: u8) -> Result<Vec<u8>
/// Ensures FLAC frames are only sent at frame boundaries to prevent sync errors in strict decoders like FFPlay.
async fn broadcast_ogg_flac_stream(
mut flac_stream: FlacEncodedStream,
broadcast_tx: broadcast::Sender<Bytes>,
broadcast_tx: timed_broadcast::Sender<Bytes>,
header_cache: Arc<RwLock<Option<Bytes>>>,
current_timestamp: Arc<RwLock<f64>>,
broadcast_max_lead_time: f64,
@@ -811,9 +830,15 @@ async fn broadcast_ogg_flac_stream(
);
// Broadcast header
let _ = broadcast_tx.send(bos_bytes);
if broadcast_tx.send(bos_bytes.clone(), 0.0).await.is_err() {
trace!("No receivers for BOS page, terminating broadcast");
return Ok(());
}
total_ogg_bytes += comment_bytes.len() as u64;
let _ = broadcast_tx.send(comment_bytes);
if broadcast_tx.send(comment_bytes.clone(), 0.0).await.is_err() {
trace!("No receivers for comment page, terminating broadcast");
return Ok(());
}
// Step 4: Read FLAC stream and create OGG packets
// Use larger read buffer (16KB) to reduce syscalls and accumulator for frame boundary detection
@@ -830,7 +855,11 @@ async fn broadcast_ogg_flac_stream(
let eos_page = ogg_writer.create_page(&flac_accumulator, false, true, false);
let eos_bytes = Bytes::from(eos_page);
total_ogg_bytes += eos_bytes.len() as u64;
let _ = broadcast_tx.send(eos_bytes);
let eos_ts = *current_timestamp.read().await;
if broadcast_tx.send(eos_bytes.clone(), eos_ts).await.is_err() {
trace!("Broadcast closed before sending final EOS page");
break;
}
info!(
"Sent final EOS page with {} bytes of data",
flac_accumulator.len()
@@ -840,7 +869,11 @@ async fn broadcast_ogg_flac_stream(
let eos_page = ogg_writer.create_page(&[], false, true, false);
let eos_bytes = Bytes::from(eos_page);
total_ogg_bytes += eos_bytes.len() as u64;
let _ = broadcast_tx.send(eos_bytes);
let eos_ts = *current_timestamp.read().await;
if broadcast_tx.send(eos_bytes.clone(), eos_ts).await.is_err() {
trace!("Broadcast closed before sending empty EOS page");
break;
}
info!("Sent empty EOS page");
}
@@ -979,12 +1012,13 @@ async fn broadcast_ogg_flac_stream(
}
// Envoyer au broadcast
match broadcast_tx.send(ogg_bytes.clone()) {
match broadcast_tx.send(ogg_bytes.clone(), audio_timestamp).await {
Ok(n) => {
trace!("Broadcasted OGG page with 1 FLAC frame ({} bytes), {} samples ({} bytes total with OGG overhead) to {} receivers", first_frame.len(), first_frame_samples, ogg_bytes.len(), n);
}
Err(e) => {
trace!("No active receivers for OGG-FLAC broadcast: {}", e);
Err(_) => {
trace!("No active receivers for OGG-FLAC broadcast, terminating loop");
return Ok(());
}
}
}

475
pmoaudio-ext/src/sinks/timed_broadcast.rs Normal file
View File

@@ -0,0 +1,475 @@
//! Broadcast channel avec TTL et propagation de TopZero.
//! Inspiré de `tokio::sync::broadcast` mais ajoute :
//! - Capacité bornée avec blocage des producteurs quand aucun slot n’est libre.
//! - Expiration automatique des messages (TTL) pour libérer les slots.
//! - Propagation d’un compteur `epoch` incrémenté sur chaque TopZeroSync.
use std::{
collections::VecDeque,
fmt,
sync::{
atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
Arc, Mutex, Weak,
},
time::{Duration, Instant},
};
use tokio::sync::Notify;
use tracing::warn;
/// Paquet diffusé contenant la charge utile + méta timing.
#[derive(Clone)]
pub struct TimedPacket<T> {
/// Charge utile diffusée aux clients.
pub payload: T,
/// Timestamp audio relatif (en secondes) pour pacing côté client.
pub audio_timestamp: f64,
/// Compteur incrémenté lorsqu'un TopZeroSync est reçu.
pub epoch: u64,
}
impl<T> fmt::Debug for TimedPacket<T> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("TimedPacket")
.field("audio_timestamp", &self.audio_timestamp)
.field("epoch", &self.epoch)
.finish_non_exhaustive()
}
}
/// Erreur remontée par `Receiver::try_recv`.
#[derive(Debug)]
pub enum TryRecvError {
Empty,
Lagged(u64),
Closed,
}
/// Erreur remontée par `Receiver::recv`.
#[derive(Debug)]
pub enum RecvError {
Lagged(u64),
Closed,
}
/// Erreur remontée par `Sender::send`.
#[derive(Debug)]
pub struct SendError<T>(pub T);
struct Entry<T> {
seq: u64,
expires_at: Instant,
payload: T,
audio_timestamp: f64,
epoch: u64,
}
struct State<T> {
buffer: VecDeque<Entry<T>>,
head_seq: u64,
next_seq: u64,
closed: bool,
epoch: u64,
epoch_start: Instant,
cursors: Vec<Weak<ReceiverCursor>>,
}
impl<T> State<T> {
fn new(capacity: usize, epoch_start: Instant) -> Self {
Self {
buffer: VecDeque::with_capacity(capacity),
head_seq: 0,
next_seq: 0,
closed: false,
epoch: 0,
epoch_start,
cursors: Vec::new(),
}
}
fn purge_expired(&mut self, now: Instant) -> bool {
let mut purged = 0u64;
while let Some(entry) = self.buffer.front() {
if entry.expires_at <= now {
self.buffer.pop_front();
self.head_seq += 1;
purged += 1;
} else {
break;
}
}
if purged > 0 {
warn!(
"TimedBroadcast: purged {} expired packet(s) (head_seq={})",
purged,
self.head_seq
);
return true;
}
false
}
fn prune_consumed(&mut self) -> bool {
let mut min_next = self.next_seq;
let mut has_cursor = false;
self.cursors.retain(|weak| {
if let Some(cursor) = weak.upgrade() {
let pos = cursor.next_seq.load(Ordering::SeqCst);
if pos < min_next {
min_next = pos;
}
has_cursor = true;
true
} else {
false
}
});
if !has_cursor {
return false;
}
let removable = min_next.saturating_sub(self.head_seq) as usize;
if removable == 0 {
return false;
}
for _ in 0..removable {
if self.buffer.pop_front().is_some() {
self.head_seq += 1;
}
}
true
}
}
struct Inner<T> {
state: Mutex<State<T>>,
data_notify: Notify,
space_notify: Notify,
capacity: usize,
sender_count: AtomicUsize,
receiver_count: AtomicUsize,
is_closed: AtomicBool,
}
impl<T> Inner<T> {
fn new(capacity: usize) -> Self {
Self {
state: Mutex::new(State::new(capacity, Instant::now())),
data_notify: Notify::new(),
space_notify: Notify::new(),
capacity,
sender_count: AtomicUsize::new(1),
receiver_count: AtomicUsize::new(0),
is_closed: AtomicBool::new(false),
}
}
fn close(&self) {
if !self
.is_closed
.swap(true, Ordering::SeqCst)
{
if let Ok(mut state) = self.state.lock() {
state.closed = true;
}
self.data_notify.notify_waiters();
self.space_notify.notify_waiters();
}
}
}
/// Créé un channel broadcast temporisé.
pub fn channel<T>(capacity: usize) -> (Sender<T>, Receiver<T>) {
assert!(capacity > 0, "capacity must be > 0");
let inner = Arc::new(Inner::new(capacity));
let next_seq = {
let state = inner
.state
.lock()
.expect("timed broadcast mutex poisoned");
state.next_seq
};
let sender = Sender {
inner: inner.clone(),
};
let cursor = Arc::new(ReceiverCursor {
next_seq: AtomicU64::new(next_seq),
});
{
let mut state = inner
.state
.lock()
.expect("timed broadcast mutex poisoned");
state.cursors.push(Arc::downgrade(&cursor));
}
inner.receiver_count.store(1, Ordering::SeqCst);
let receiver = Receiver {
inner,
next_seq,
cursor,
};
(sender, receiver)
}
/// Sender côté producteur.
pub struct Sender<T> {
inner: Arc<Inner<T>>,
}
impl<T> Clone for Sender<T> {
fn clone(&self) -> Self {
self.inner.sender_count.fetch_add(1, Ordering::SeqCst);
Self {
inner: self.inner.clone(),
}
}
}
impl<T> Sender<T> {
/// Diffuse un paquet. Bloque si la capacité est atteinte avec des paquets non périmés.
pub async fn send(&self, payload: T, audio_timestamp: f64) -> Result<usize, SendError<T>>
where
T: Clone,
{
let mut payload = Some(payload);
loop {
let mut wait_deadline = None;
{
let now = Instant::now();
let mut state = self
.inner
.state
.lock()
.expect("timed broadcast mutex poisoned");
if state.closed {
return Err(SendError(payload.expect("payload already consumed")));
}
if state.purge_expired(now) {
self.inner.space_notify.notify_waiters();
}
if state.prune_consumed() {
self.inner.space_notify.notify_waiters();
}
if state.buffer.len() < self.inner.capacity {
let audio_offset =
Duration::from_secs_f64(audio_timestamp.max(0.0));
let expires_at = state.epoch_start + audio_offset;
let entry = Entry {
seq: state.next_seq,
expires_at,
payload: payload
.take()
.expect("payload already consumed"),
audio_timestamp,
epoch: state.epoch,
};
state.next_seq += 1;
state.buffer.push_back(entry);
let receivers = self.inner.receiver_count.load(Ordering::SeqCst);
drop(state);
self.inner.data_notify.notify_waiters();
return Ok(receivers);
}
wait_deadline = state.buffer.front().map(|entry| entry.expires_at);
}
if let Some(deadline) = wait_deadline {
let deadline = tokio::time::Instant::from_std(deadline);
tokio::select! {
_ = self.inner.space_notify.notified() => {},
_ = tokio::time::sleep_until(deadline) => {},
}
} else {
self.inner.space_notify.notified().await;
}
}
}
/// Crée un nouveau receiver abonné au flux.
pub fn subscribe(&self) -> Receiver<T> {
let mut state = self
.inner
.state
.lock()
.expect("timed broadcast mutex poisoned");
let next_seq = state.next_seq;
let cursor = Arc::new(ReceiverCursor {
next_seq: AtomicU64::new(next_seq),
});
state.cursors.push(Arc::downgrade(&cursor));
state.prune_consumed();
drop(state);
self.inner.receiver_count.fetch_add(1, Ordering::SeqCst);
Receiver {
inner: self.inner.clone(),
next_seq,
cursor,
}
}
/// Marque un TopZero : incrémente l'epoch pour les paquets suivants.
pub fn mark_top_zero(&self) {
let mut state = self
.inner
.state
.lock()
.expect("timed broadcast mutex poisoned");
state.epoch = state.epoch.wrapping_add(1);
state.epoch_start = Instant::now();
if !state.buffer.is_empty() {
state.head_seq = state.next_seq;
state.buffer.clear();
self.inner.space_notify.notify_waiters();
}
}
/// Nombre actuel de receivers abonnés.
pub fn receiver_count(&self) -> usize {
self.inner.receiver_count.load(Ordering::SeqCst)
}
/// Ferme explicitement le channel.
pub fn close(&self) {
self.inner.close();
}
}
impl<T> Drop for Sender<T> {
fn drop(&mut self) {
if self.inner.sender_count.fetch_sub(1, Ordering::SeqCst) == 1 {
self.inner.close();
}
}
}
/// Receiver côté consommateur.
pub struct Receiver<T> {
inner: Arc<Inner<T>>,
next_seq: u64,
cursor: Arc<ReceiverCursor>,
}
struct ReceiverCursor {
next_seq: AtomicU64,
}
impl<T> Receiver<T>
where
T: Clone,
{
fn poll_entry(&mut self) -> Result<TimedPacket<T>, TryRecvError> {
let mut state = self
.inner
.state
.lock()
.expect("timed broadcast mutex poisoned");
if state.closed && state.buffer.is_empty() {
return Err(TryRecvError::Closed);
}
let now = Instant::now();
if state.purge_expired(now) {
self.inner.space_notify.notify_waiters();
}
if self.next_seq < state.head_seq {
let skipped = state.head_seq - self.next_seq;
self.next_seq = state.head_seq;
return Err(TryRecvError::Lagged(skipped));
}
let offset = (self.next_seq - state.head_seq) as usize;
if offset < state.buffer.len() {
let entry = state
.buffer
.get(offset)
.expect("invalid buffer offset");
let packet = TimedPacket {
payload: entry.payload.clone(),
audio_timestamp: entry.audio_timestamp,
epoch: entry.epoch,
};
self.next_seq += 1;
self.cursor
.next_seq
.store(self.next_seq, Ordering::SeqCst);
if state.prune_consumed() {
self.inner.space_notify.notify_waiters();
}
return Ok(packet);
}
if state.closed {
Err(TryRecvError::Closed)
} else {
Err(TryRecvError::Empty)
}
}
/// Version synchrone utilisée dans `poll_read`.
pub fn try_recv(&mut self) -> Result<TimedPacket<T>, TryRecvError> {
self.poll_entry()
}
/// Attends qu'un paquet soit disponible.
pub async fn recv(&mut self) -> Result<TimedPacket<T>, RecvError> {
loop {
match self.try_recv() {
Ok(packet) => return Ok(packet),
Err(TryRecvError::Empty) => {
self.inner.data_notify.notified().await;
}
Err(TryRecvError::Lagged(skipped)) => return Err(RecvError::Lagged(skipped)),
Err(TryRecvError::Closed) => return Err(RecvError::Closed),
}
}
}
}
impl<T> Clone for Receiver<T> {
fn clone(&self) -> Self {
self.inner.receiver_count.fetch_add(1, Ordering::SeqCst);
let cursor = Arc::new(ReceiverCursor {
next_seq: AtomicU64::new(self.next_seq),
});
{
let mut state = self
.inner
.state
.lock()
.expect("timed broadcast mutex poisoned");
state.cursors.push(Arc::downgrade(&cursor));
}
Self {
inner: self.inner.clone(),
next_seq: self.next_seq,
cursor,
}
}
}
impl<T> Drop for Receiver<T> {
fn drop(&mut self) {
self.cursor
.next_seq
.store(self.next_seq, Ordering::SeqCst);
if let Ok(mut state) = self.inner.state.lock() {
if state.prune_consumed() {
self.inner.space_notify.notify_waiters();
}
}
if self.inner.receiver_count.fetch_sub(1, Ordering::SeqCst) == 1 {
self.inner.space_notify.notify_waiters();
}
}
}