rill_telemetry/
collector.rs

1use std::sync::Arc;
2
3use rill_core::math::Transcendental;
4use rill_core::queues::spsc::SpscQueue;
5use rill_core::queues::TelemetryBlock;
6
7/// Non-real-time consumer of telemetry frames from a shared `SpscQueue`.
8///
9/// Runs on the control thread (or any non-RT context). Drains available
10/// frames from the ring buffer and delivers them to a user-supplied callback.
11///
12/// # Example
13/// ```ignore
14/// use std::sync::Arc;
15/// use rill_core::queues::spsc::SpscQueue;
16/// use rill_core::queues::TelemetryBlock;
17/// use rill_telemetry::collector::TelemetryCollector;
18///
19/// let queue = Arc::new(SpscQueue::<TelemetryBlock<f32, 64>, 8>::new());
20/// let mut collector = TelemetryCollector::new(queue, |frame| {
21///     println!("peak={} rms={} dc={}", frame.peak, frame.rms, frame.dc_offset);
22/// });
23/// collector.poll();  // drain available frames
24/// ```
25pub struct TelemetryCollector<T: Transcendental, const BUF_SIZE: usize, const QUEUE_CAP: usize> {
26    queue: Arc<SpscQueue<TelemetryBlock<T, BUF_SIZE>, QUEUE_CAP>>,
27    callback: Box<dyn FnMut(TelemetryBlock<T, BUF_SIZE>)>,
28}
29
30impl<T: Transcendental, const BUF_SIZE: usize, const QUEUE_CAP: usize>
31    TelemetryCollector<T, BUF_SIZE, QUEUE_CAP>
32{
33    /// Create a new collector backed by the shared ring buffer.
34    ///
35    /// The `callback` is invoked for every frame drained from the queue.
36    pub fn new(
37        queue: Arc<SpscQueue<TelemetryBlock<T, BUF_SIZE>, QUEUE_CAP>>,
38        callback: impl FnMut(TelemetryBlock<T, BUF_SIZE>) + 'static,
39    ) -> Self {
40        Self {
41            queue,
42            callback: Box::new(callback),
43        }
44    }
45
46    /// Drain all available telemetry frames from the ring buffer.
47    ///
48    /// Returns the number of frames processed.
49    pub fn poll(&mut self) -> usize {
50        let mut count = 0;
51        while let Some(frame) = self.queue.pop() {
52            (self.callback)(frame);
53            count += 1;
54        }
55        count
56    }
57
58    /// Drain at most `max` frames from the ring buffer.
59    ///
60    /// Useful when you want to bound processing time per tick.
61    pub fn poll_n(&mut self, max: usize) -> usize {
62        let mut count = 0;
63        while count < max {
64            match self.queue.pop() {
65                Some(frame) => {
66                    (self.callback)(frame);
67                    count += 1;
68                }
69                None => break,
70            }
71        }
72        count
73    }
74
75    /// Number of frames currently available in the ring buffer (approximate).
76    pub fn available(&self) -> usize {
77        self.queue.len()
78    }
79
80    /// Consume the collector and return the underlying queue Arc.
81    pub fn into_queue(self) -> Arc<SpscQueue<TelemetryBlock<T, BUF_SIZE>, QUEUE_CAP>> {
82        self.queue
83    }
84}
rill_telemetry/collector.rs

rill_telemetry/
collector.rs
