oms_modbus/monitor/
ring_buffer.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2//!
3//! In-memory ring buffer for Modbus traffic capture.
4//!
5//! Default mode: bounded ring buffer (10,000 records).
6//! Optional: unlimited mode (never drops).
7//!
8//! Records are always returned in chronological order by [`RingBufferCapture::drain`] and [`RingBufferCapture::snapshot`].
9
10use std::collections::VecDeque;
11use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
12use std::sync::Mutex;
13
14use crate::intercept::{PacketData, PacketRecord};
15use crate::wire_tap::WireTap;
16
17/// In-memory packet capture with optional ring-buffer eviction.
18///
19/// Prefer [`crate::BusCapture`] for most use cases — it includes
20/// statistics counters and supports stats-only, bounded, and unbounded
21/// modes. Use `RingBufferCapture` only when you need a standalone
22/// ring buffer without statistics tracking.
23///
24/// # Example
25///
26/// ```no_run
27/// use oms_modbus::monitor::RingBufferCapture;
28/// let capture = RingBufferCapture::new(10_000); // bounded, 10k records
29/// ```
30pub struct RingBufferCapture {
31    records: Mutex<VecDeque<PacketRecord>>,
32    capacity: usize,
33    unbounded: AtomicBool,
34    dropped: AtomicU64,
35}
36
37impl RingBufferCapture {
38    /// Create a bounded ring buffer with the given capacity.
39    /// When full, oldest records are evicted.
40    ///
41    /// `capacity` is clamped to a minimum of 1. Passing 0 is equivalent to
42    /// passing 1 — a single-record buffer.
43    pub fn new(capacity: usize) -> Self {
44        let cap = capacity.max(1);
45        Self {
46            records: Mutex::new(VecDeque::with_capacity(cap)),
47            capacity: cap,
48            unbounded: AtomicBool::new(false),
49            dropped: AtomicU64::new(0),
50        }
51    }
52
53    /// Create an unbounded capture that never drops records.
54    /// Use with caution — memory grows indefinitely.
55    pub fn unbounded() -> Self {
56        Self {
57            records: Mutex::new(VecDeque::with_capacity(1024)),
58            capacity: usize::MAX,
59            unbounded: AtomicBool::new(true),
60            dropped: AtomicU64::new(0),
61        }
62    }
63
64    /// Number of records currently buffered.
65    pub fn len(&self) -> usize {
66        self.records.lock().unwrap_or_else(|e| e.into_inner()).len()
67    }
68
69    /// Check if the buffer is empty.
70    pub fn is_empty(&self) -> bool {
71        self.records
72            .lock()
73            .unwrap_or_else(|e| e.into_inner())
74            .is_empty()
75    }
76
77    /// Number of records dropped due to buffer overflow (bounded mode only).
78    pub fn dropped(&self) -> u64 {
79        self.dropped.load(Ordering::Relaxed)
80    }
81
82    /// Drain all records in chronological order. Clears the buffer.
83    pub fn drain(&self) -> Vec<PacketRecord> {
84        let mut r = self.records.lock().unwrap_or_else(|e| e.into_inner());
85        // Swap out the VecDeque under the lock, drain outside.
86        let capacity = r.capacity();
87        let old = std::mem::replace(&mut *r, VecDeque::with_capacity(capacity));
88        old.into_iter().collect()
89    }
90
91    /// Take a snapshot without clearing. Records are in chronological order.
92    ///
93    /// Clones all records under the lock. For large buffers in hot paths,
94    /// prefer [`Self::drain`] which swaps the buffer and releases the lock immediately.
95    pub fn snapshot(&self) -> Vec<PacketRecord> {
96        self.records
97            .lock()
98            .unwrap_or_else(|e| e.into_inner())
99            .iter()
100            .cloned()
101            .collect()
102    }
103
104    /// Switch to unbounded mode (existing records preserved).
105    pub fn set_unbounded(&self) {
106        self.unbounded.store(true, Ordering::Relaxed);
107    }
108
109    /// Switch to bounded mode, stopping unbounded growth.
110    ///
111    /// Capacity is fixed at construction time; this only prevents the ring
112    /// buffer from growing without bound. Use [`RingBufferCapture::new`]
113    /// to create a capture with a specific capacity.
114    pub fn set_bounded(&self) {
115        self.unbounded.store(false, Ordering::Relaxed);
116    }
117
118    fn record(&self, timestamp_us: u64, data: PacketData) {
119        let record = PacketRecord { timestamp_us, data };
120        // Read unbounded flag before lock — TOCTOU is harmless here:
121        // worst case one record uses the old mode during a mode switch.
122        let is_unbounded = self.unbounded.load(Ordering::Relaxed);
123        let mut r = self.records.lock().unwrap_or_else(|e| e.into_inner());
124        if is_unbounded || r.len() < self.capacity {
125            r.push_back(record);
126        } else {
127            // Evict oldest — VecDeque pop_front is O(1).
128            r.pop_front();
129            r.push_back(record);
130            self.dropped.fetch_add(1, Ordering::Relaxed);
131        }
132    }
133}
134
135impl Default for RingBufferCapture {
136    fn default() -> Self {
137        Self::new(10_000)
138    }
139}
140
141impl WireTap for RingBufferCapture {
142    fn on_write(&self, bytes: &[u8], ts: u64) {
143        self.record(ts, PacketData::RawTx(bytes.to_vec()));
144    }
145    fn on_read(&self, bytes: &[u8], ts: u64) {
146        self.record(ts, PacketData::RawRx(bytes.to_vec()));
147    }
148    fn on_error(&self, bytes: &[u8], error: &str, ts: u64) {
149        self.record(ts, PacketData::RawError(bytes.to_vec(), error.to_string()));
150    }
151}
oms_modbus/monitor/ring_buffer.rs

oms_modbus/monitor/
ring_buffer.rs
