rskit_stateful/
trigger.rs

1//! Flush trigger implementations.
2
3use std::time::Duration;
4
5use rskit_errors::AppResult;
6
7use crate::Accumulator;
8
9/// Decides when an accumulator should flush.
10pub trait Trigger<V>: Send + Sync
11where
12    V: Clone + Send + Sync + 'static,
13{
14    /// Return `true` when the accumulator should flush.
15    fn should_flush(&self, accumulator: &Accumulator<V>) -> AppResult<bool>;
16
17    /// Human-readable trigger name.
18    fn name(&self) -> &'static str;
19}
20
21/// Flush when buffered item count reaches `threshold`.
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23pub struct SizeTrigger {
24    /// Threshold that triggers a flush.
25    pub threshold: usize,
26}
27
28impl SizeTrigger {
29    /// Create a new size trigger.
30    #[must_use]
31    pub fn new(threshold: usize) -> Self {
32        Self { threshold }
33    }
34}
35
36impl<V> Trigger<V> for SizeTrigger
37where
38    V: Clone + Send + Sync + 'static,
39{
40    fn should_flush(&self, accumulator: &Accumulator<V>) -> AppResult<bool> {
41        Ok(accumulator.size()? >= self.threshold)
42    }
43
44    fn name(&self) -> &'static str {
45        "size"
46    }
47}
48
49/// Flush when buffered byte size reaches `threshold`.
50#[derive(Debug, Clone, Copy, PartialEq, Eq)]
51pub struct ByteSizeTrigger {
52    /// Threshold that triggers a flush.
53    pub threshold: usize,
54}
55
56impl ByteSizeTrigger {
57    /// Create a new byte-size trigger.
58    #[must_use]
59    pub fn new(threshold: usize) -> Self {
60        Self { threshold }
61    }
62}
63
64impl<V> Trigger<V> for ByteSizeTrigger
65where
66    V: Clone + Send + Sync + 'static,
67{
68    fn should_flush(&self, accumulator: &Accumulator<V>) -> AppResult<bool> {
69        Ok(accumulator.measure()? >= self.threshold)
70    }
71
72    fn name(&self) -> &'static str {
73        "byte_size"
74    }
75}
76
77/// Flush when the time since the last flush exceeds `interval`.
78#[derive(Debug, Clone, Copy, PartialEq, Eq)]
79pub struct TimeTrigger {
80    /// Interval that triggers a flush.
81    pub interval: Duration,
82}
83
84impl TimeTrigger {
85    /// Create a new time trigger.
86    #[must_use]
87    pub fn new(interval: Duration) -> Self {
88        Self { interval }
89    }
90}
91
92impl<V> Trigger<V> for TimeTrigger
93where
94    V: Clone + Send + Sync + 'static,
95{
96    fn should_flush(&self, accumulator: &Accumulator<V>) -> AppResult<bool> {
97        Ok(accumulator.elapsed_since_flush() >= self.interval)
98    }
99
100    fn name(&self) -> &'static str {
101        "time"
102    }
103}
rskit_stateful/trigger.rs

rskit_stateful/
trigger.rs
