File size: 2,307 Bytes
50fe3c9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | use std::collections::VecDeque;
/// Byte-oriented bounded ring buffer with monotonic offsets.
///
/// Callers tail the buffer using `since_offset`: each `push` advances
/// `next_offset` by the number of bytes appended, even when older bytes are
/// dropped to fit the cap. `read_from(since)` returns the slice of bytes from
/// the requested offset (clamped to whatever is still resident) plus the new
/// offset for the next call.
pub struct BoundedRingBuffer {
buf: VecDeque<u8>,
cap: usize,
next_offset: u64,
/// Bytes that were dropped to keep the buffer ≤ cap. Helps the caller
/// detect overflow ("you missed N bytes").
dropped: u64,
}
impl BoundedRingBuffer {
pub fn new(cap: usize) -> Self {
Self {
buf: VecDeque::with_capacity(cap.min(64 * 1024)),
cap,
next_offset: 0,
dropped: 0,
}
}
pub fn push(&mut self, data: &[u8]) {
self.next_offset = self.next_offset.saturating_add(data.len() as u64);
if data.len() >= self.cap {
// Incoming chunk alone exceeds cap: keep only its tail.
let keep_from = data.len() - self.cap;
self.dropped = self
.dropped
.saturating_add((self.buf.len() + keep_from) as u64);
self.buf.clear();
self.buf.extend(&data[keep_from..]);
return;
}
let overflow = (self.buf.len() + data.len()).saturating_sub(self.cap);
if overflow > 0 {
for _ in 0..overflow {
self.buf.pop_front();
}
self.dropped = self.dropped.saturating_add(overflow as u64);
}
self.buf.extend(data);
}
/// Return bytes available since `since`, plus the new offset.
/// If `since` is older than the resident window, the returned bytes start
/// from the oldest available offset (which is `next_offset - buf.len()`).
pub fn read_from(&self, since: u64) -> (Vec<u8>, u64, u64) {
let oldest = self.next_offset.saturating_sub(self.buf.len() as u64);
let start = since.max(oldest);
let skip = (start - oldest) as usize;
let bytes: Vec<u8> = self.buf.iter().copied().skip(skip).collect();
(bytes, self.next_offset, self.dropped)
}
}
|