use std::collections::VecDeque;
pub struct RollingAverage {
buf: VecDeque<f64>,
capacity: usize,
sum: f64,
since_resync: usize,
resync_interval: usize,
}
impl RollingAverage {
pub fn new(capacity: usize) -> Self {
assert!(capacity > 0, "window capacity must be positive");
RollingAverage {
buf: VecDeque::with_capacity(capacity),
capacity,
sum: 0.0,
since_resync: 0,
resync_interval: 4096,
}
}
pub fn push(&mut self, value: f64) {
if self.buf.len() == self.capacity {
if let Some(old) = self.buf.pop_front() {
self.sum -= old;
}
}
self.buf.push_back(value);
self.sum += value;
self.recompute_if_stale();
}
pub fn average(&self) -> Option<f64> {
if self.buf.is_empty() {
return None;
}
Some(self.sum / self.buf.len() as f64)
}
pub fn is_full(&self) -> bool {
self.buf.len() == self.capacity
}
pub fn len(&self) -> usize {
self.buf.len()
}
fn recompute_if_stale(&mut self) {
self.since_resync += 1;
if self.since_resync >= self.resync_interval {
self.sum = self.buf.iter().sum();
self.since_resync = 0;
}
}
}mod rolling_average;
use rolling_average::RollingAverage;
#[derive(Debug)]
pub enum Reading {
Warmup,
Ok(f64),
Alarm(f64),
}
pub struct SensorMonitor {
window: RollingAverage,
alarm_threshold: f64,
}
impl SensorMonitor {
pub fn new(window_size: usize, alarm_threshold: f64) -> Self {
SensorMonitor {
window: RollingAverage::new(window_size),
alarm_threshold,
}
}
pub fn ingest(&mut self, raw: f64) -> Reading {
self.window.push(raw);
if !self.window.is_full() {
return Reading::Warmup;
}
let smoothed = self.window.average().expect("full window has an average");
if smoothed > self.alarm_threshold {
Reading::Alarm(smoothed)
} else {
Reading::Ok(smoothed)
}
}
}
fn main() {
let mut monitor = SensorMonitor::new(3, 70.0);
let samples = [22.0, 21.5, 23.0, 68.0, 95.0, 72.0, 24.0];
for (i, &sample) in samples.iter().enumerate() {
match monitor.ingest(sample) {
Reading::Warmup => println!("t={i}: warming up"),
Reading::Ok(avg) => println!("t={i}: ok, avg={avg:.2}"),
Reading::Alarm(avg) => println!("t={i}: ALARM, avg={avg:.2}"),
}
}
}This snippet shows a small, focused streaming aggregator that maintains a rolling average over the most recent N sensor readings without ever re-scanning the whole window on each update. The design centers on a fixed-capacity ring buffer so that each push is O(1) in both time and memory, which matters when readings arrive at high frequency from an embedded source.
In rolling_average.rs, RollingAverage stores a VecDeque<f64> bounded by capacity plus a running sum. The core idea is incremental maintenance: when a new reading arrives, push adds it to sum, and if the buffer is already full it pops the oldest value and subtracts it. This turns the average into a constant-time operation rather than the naive O(N) recomputation. The average method returns Option<f64> because an empty window has no defined mean — a deliberate choice that forces callers to handle the cold-start case instead of dividing by zero.
A subtle trade-off is numerical drift: continuously adding and subtracting f64 values lets rounding error accumulate over millions of samples. To bound this, push calls recompute_if_stale, which every resync_interval updates rebuilds sum from the buffer contents via iter().sum(). This keeps the fast path cheap while periodically restoring accuracy. The is_full helper and len expose window state for downstream logic.
In sensor_stream.rs, SensorMonitor wires the aggregator to a threshold alarm. ingest feeds each raw reading into the RollingAverage and then compares the smoothed value against alarm_threshold, returning a Reading enum variant. Smoothing before alarming is what prevents a single noisy spike from triggering false positives — the window acts as a low-pass filter. The main function demonstrates the pattern with a short burst of readings, showing how Reading::Warmup covers the period before the window fills.
This approach fits any bounded, latency-sensitive stream — telemetry, financial ticks, or rate limiting — where recomputing over the full history is wasteful and only recent context matters.
Related snips
struct Config<'a> {
name: &'a str,
value: &'a str,
}
fn parse_config(line: &str) -> Config {Lifetime annotations for flexible borrowing in structs
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
use tracing::{info, instrument};
#[instrument]
fn process_request(user_id: u64) {
info!(user_id, "Processing request");
// Work happens heretracing for structured logging and distributed tracing
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
use clap::Parser;
#[derive(Parser, Debug)]
#[command(author, version, about)]
struct Args {
#[arg(short, long)]clap for CLI argument parsing with derive macros
Share this code
Here's the card — post it anywhere.
