Windowed alerts from a sensor stream

FxDart wins async

Requirement

A boiler temperature sensor delivers readings as a real Dart Stream — one every 10 ms, twelve in total (fixed data, in the code below). Group the stream into windows of four readings, report each window's average and peak, and raise an ALERT line for any window whose average is at or above 75.00.

FxDart's answer is its stream bridge: fxStream lifts the Stream into the pull-based pipeline, and from there windowing is just chunk(4) — the same operator the sync examples use — followed by a map that summarizes each window with averageBy and maxBy.

Expected output

boiler sensor, windows of 4 readings:
  0s-3s  avg 69.50  peak 70.5
  4s-7s  avg 75.75  peak 77.0
  8s-11s  avg 69.88  peak 72.0
ALERT 4s-7s: average 75.75 is above the 75.00 limit

Side by side

Native Dart

class Reading {
  final int second; // offset into the run
  final double c; // temperature in celsius
  const Reading(this.second, this.c);
}

const samples = [
  Reading(0, 68.0), Reading(1, 69.5), Reading(2, 70.0), Reading(3, 70.5),
  Reading(4, 74.0), Reading(5, 76.5), Reading(6, 77.0), Reading(7, 75.5),
  Reading(8, 72.0), Reading(9, 70.0), Reading(10, 69.0), Reading(11, 68.5),
];

/// The sensor: a real Dart Stream, one reading every 10 ms.
Stream<Reading> sensor() async* {
  for (final r in samples) {
    await Future.delayed(const Duration(milliseconds: 10));
    yield r;
  }
}

Future<void> main() async {
  // Manual windowing: buffer 4 readings, flush, repeat.
  final windows = <(String, double, double)>[];
  var buf = <Reading>[];
  await for (final r in sensor()) {
    buf.add(r);
    if (buf.length == 4) {
      final avg =
          buf.fold(0.0, (sum, r) => sum + r.c) / buf.length;
      final peak = buf.map((r) => r.c).reduce((a, b) => a > b ? a : b);
      windows.add(('${buf.first.second}s-${buf.last.second}s', avg, peak));
      buf = [];
    }
  }
  print('boiler sensor, windows of 4 readings:');
  for (final (span, avg, peak) in windows) {
    print('  $span  avg ${avg.toStringAsFixed(2)}  '
        'peak ${peak.toStringAsFixed(1)}');
  }
  for (final w in windows.where((w) => w.$2 >= 75.0)) {
    print('ALERT ${w.$1}: average ${w.$2.toStringAsFixed(2)} '
        'is above the 75.00 limit');
  }
}

FxDart

import 'package:fxdart/fxdart.dart';

class Reading {
  final int second; // offset into the run
  final double c; // temperature in celsius
  const Reading(this.second, this.c);
}

Future<void> main() async {
  final windows = await fxStream(sensor())
      .chunk(4)
      .map((w) => (
            '${w.first.second}s-${w.last.second}s',
            fx(w).averageBy((r) => r.c),
            fx(w).maxBy((r) => r.c)!.c,
          ))
      .toList();
  print('boiler sensor, windows of 4 readings:');
  for (final (span, avg, peak) in windows) {
    print('  $span  avg ${avg.toStringAsFixed(2)}  '
        'peak ${peak.toStringAsFixed(1)}');
  }
  for (final w in fx(windows).filter((w) => w.$2 >= 75.0)) {
    print('ALERT ${w.$1}: average ${w.$2.toStringAsFixed(2)} '
        'is above the 75.00 limit');
  }
}

Why they differ

Dart's Stream API has no windowing operator. The idiomatic options are an await for loop with a mutable buffer — accumulate four, flush, reset, as shown — or packaging that same bookkeeping into a custom StreamTransformer, which is more code, not less. Either way the buffer, the flush condition, and the reset are yours to maintain, and the partial-window edge case is yours to reason about. In FxDart, chunk(4) is one word on a stream exactly as it is on a list — crossing from Stream to pipeline costs one fxStream call, and the whole operator vocabulary comes with it.

Benchmark

Apple M1 Max, 32 GB RAM · Dart 3.12.2 (AOT-compiled) · 2026-08-24

Async case: the headline scale is N = 100,000, not 1,000,000. Every element costs an event-loop turn on both sides, so a million real awaits would measure Dart's event loop for minutes — not the pipeline. Delays are zero-length and the example's concurrency limit is kept; what the bars compare is the pipeline machinery.

N = 100

Time Tie

Native Dart 37 µs

FxDart 42 µs

Peak memory Tie

Native Dart 16.5 MB

FxDart 17.2 MB

N = 10,000

Time Tie

Native Dart 3.30 ms

FxDart 3.47 ms

Peak memory Native wins

Native Dart 24.0 MB

FxDart 25.2 MB

N = 100,000

Time Native wins

Native Dart 32.8 ms

FxDart 36.6 ms

Peak memory Tie

Native Dart 75.0 MB

FxDart 74.8 MB

Bars are medians of repeated timed iterations in fresh processes per side (small N is batched for timer resolution). Sides within 5% of each other — or within 0.6 ms, a difference no person can perceive — count as a tie; close relative races are re-measured up to 5 runs. In an app, anything under a few milliseconds is invisible to the user regardless of which bar is shorter. Memory is peak process RSS. The Dart VM and the dataset are identical on both sides, so the difference between the two bars is what the pipeline itself holds onto.