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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
//! `Batch<T>` — accumulate items, emit a full batch when a count threshold is
//! hit.
//!
//! The coalescing half of a buffered sink, factored out of the sink so any
//! caller composes it with its own terminal: recording's `AccumulatingSink`
//! buffers `RecordingEvent`s before a durable write; a telemetry exporter
//! buffers records before an OTLP flush. `Batch` owns only the buffer — it never
//! touches a sink, so it carries no `SendPipe` bound and no error type. `push`
//! hands back the full batch for the caller to flush; `drain` takes the partial
//! remainder on an explicit flush.
//!
//! The buffer is guarded by [`crate::sync::blocking::Mutex`] — the crate's
//! canonical blocking lock. It never poisons, so a panic in one appender
//! cannot wedge the batcher for every other one.
use alloc::vec::Vec;
use crate::sync::blocking::Mutex;
/// A count-thresholded accumulator. `Clone`-free, shareable behind an `Arc`.
pub struct Batch<T> {
buffer: Mutex<Vec<T>>,
batch_size: usize,
}
impl<T> Batch<T> {
/// Coalesce into `batch_size`-item blocks (clamped to at least 1).
#[must_use]
pub fn new(batch_size: usize) -> Self {
let batch_size = batch_size.max(1);
Self {
buffer: Mutex::new(Vec::with_capacity(batch_size)),
batch_size,
}
}
/// Push one item. Returns the full batch to flush once the buffer reaches
/// the threshold (leaving the buffer empty), otherwise `None`.
#[must_use = "the returned batch must be flushed to its sink"]
pub fn push(&self, item: T) -> Option<Vec<T>> {
let mut guard = self.buffer.lock();
guard.push(item);
if guard.len() >= self.batch_size {
Some(core::mem::take(&mut *guard))
} else {
None
}
}
/// Take whatever is buffered (an explicit flush); empties the buffer.
#[must_use = "the drained batch must be flushed to its sink"]
pub fn drain(&self) -> Vec<T> {
core::mem::take(&mut *self.buffer.lock())
}
/// Items currently buffered (not yet flushed).
#[must_use]
pub fn len(&self) -> usize {
self.buffer.lock().len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.buffer.lock().is_empty()
}
/// The flush threshold.
#[must_use]
pub fn batch_size(&self) -> usize {
self.batch_size
}
}
#[cfg(test)]
mod tests {
#![allow(clippy::unwrap_used, clippy::expect_used)]
use super::*;
#[test]
fn push_below_threshold_buffers_and_returns_none() {
let batch = Batch::new(4);
assert!(batch.push(1).is_none());
assert!(batch.push(2).is_none());
assert_eq!(batch.len(), 2);
}
#[test]
fn push_at_threshold_emits_full_batch_and_empties() {
let batch = Batch::new(3);
assert!(batch.push(10).is_none());
assert!(batch.push(20).is_none());
let full = batch.push(30).expect("threshold reached emits the batch");
assert_eq!(full, alloc::vec![10, 20, 30]);
assert!(batch.is_empty(), "buffer emptied after emitting");
}
#[test]
fn drain_takes_the_partial_remainder() {
let batch = Batch::new(8);
let _ = batch.push(1);
let _ = batch.push(2);
assert_eq!(batch.drain(), alloc::vec![1, 2]);
assert!(batch.is_empty());
assert_eq!(
batch.drain(),
Vec::<i32>::new(),
"draining empty yields nothing"
);
}
#[test]
fn zero_batch_size_clamps_to_one() {
let batch = Batch::new(0);
assert_eq!(batch.batch_size(), 1);
assert_eq!(
batch.push(7),
Some(alloc::vec![7]),
"every push flushes at size 1"
);
}
}