sim_lib_audio_graph_live/
queue.rs

1use std::collections::VecDeque;
2
3use sim_kernel::{Error, Result};
4use sim_lib_stream_core::StreamStats;
5
6use crate::{LiveAudioEvent, LiveControlEvent, LiveQueuePush};
7
8/// Bounded queue carrying control events into the audio callback.
9pub type ControlToAudioQueue = BoundedLiveQueue<LiveControlEvent>;
10/// Bounded queue carrying audio-thread events back to the control thread.
11pub type AudioToControlQueue = BoundedLiveQueue<LiveAudioEvent>;
12
13/// Bounded queue used at live audio/control boundaries.
14#[derive(Clone, Debug)]
15pub struct BoundedLiveQueue<T> {
16    entries: VecDeque<T>,
17    bound: usize,
18    pending_dropped_newest: u64,
19    stats: StreamStats,
20}
21
22impl<T> BoundedLiveQueue<T> {
23    /// Creates a queue bounded to `bound` items, rejecting a zero bound.
24    pub fn with_capacity(bound: usize) -> Result<Self> {
25        if bound == 0 {
26            return Err(Error::Eval(
27                "live queue capacity must be greater than zero".to_owned(),
28            ));
29        }
30        Ok(Self {
31            entries: VecDeque::with_capacity(bound),
32            bound,
33            pending_dropped_newest: 0,
34            stats: StreamStats::default(),
35        })
36    }
37
38    /// Pushes an item, dropping the newest under backpressure when full.
39    pub fn push(&mut self, item: T) -> LiveQueuePush {
40        self.stats.pushed = self.stats.pushed.saturating_add(1);
41        if self.entries.len() >= self.bound {
42            self.pending_dropped_newest = self.pending_dropped_newest.saturating_add(1);
43            self.stats.dropped_newest = self.stats.dropped_newest.saturating_add(1);
44            LiveQueuePush::DroppedNewest
45        } else {
46            self.entries.push_back(item);
47            self.stats.accepted = self.stats.accepted.saturating_add(1);
48            LiveQueuePush::Accepted
49        }
50    }
51
52    /// Removes and returns the oldest item, if any.
53    pub fn pop(&mut self) -> Option<T> {
54        let item = self.entries.pop_front();
55        if item.is_some() {
56            self.stats.yielded = self.stats.yielded.saturating_add(1);
57        }
58        item
59    }
60
61    /// Returns the number of queued items.
62    pub fn len(&self) -> usize {
63        self.entries.len()
64    }
65
66    /// Returns whether the queue is empty.
67    pub fn is_empty(&self) -> bool {
68        self.entries.is_empty()
69    }
70
71    /// Returns the configured capacity bound.
72    pub fn capacity(&self) -> usize {
73        self.bound
74    }
75
76    /// Returns the backing buffer's allocated capacity.
77    pub fn allocated_capacity(&self) -> usize {
78        self.entries.capacity()
79    }
80
81    /// Returns the count of items dropped since the last [`take_dropped`](Self::take_dropped).
82    pub fn dropped(&self) -> u64 {
83        self.pending_dropped_newest
84    }
85
86    /// Returns and clears the pending dropped-item count.
87    pub fn take_dropped(&mut self) -> u64 {
88        let dropped = self.pending_dropped_newest;
89        self.pending_dropped_newest = 0;
90        dropped
91    }
92
93    /// Returns a snapshot of the queue's stream statistics.
94    pub fn stats(&self) -> StreamStats {
95        self.stats.clone()
96    }
97}
sim_lib_audio_graph_live/queue.rs

sim_lib_audio_graph_live/
queue.rs
