Alertas por ventanas sobre un stream de sensores

Gana FxDart async

Requisito

Un sensor de temperatura de caldera entrega lecturas como un Stream real de Dart — una cada 10 ms, doce en total (datos fijos, en el código de abajo). Agrupa el stream en ventanas de cuatro lecturas, informa de la media y el pico de cada ventana, y emite una línea ALERT para toda ventana cuya media sea igual o superior a 75.00.

La respuesta de FxDart es su puente con los streams: fxStream eleva el Stream al pipeline basado en pull, y a partir de ahí las ventanas son simplemente chunk(4) — el mismo operador que usan los ejemplos síncronos — seguido de un map que resume cada ventana con averageBy y maxBy.

Salida esperada

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

Lado a lado

Dart nativo

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');
  }
}

Por qué difieren

La API de Stream de Dart no tiene ningún operador de ventanas. Las opciones idiomáticas son un bucle await for con un búfer mutable — acumular cuatro, volcar, reiniciar, como se muestra — o empaquetar esa misma contabilidad en un StreamTransformer propio, que es más código, no menos. En cualquier caso el búfer, la condición de volcado y el reinicio son cosa tuya, y el caso límite de la ventana incompleta también te toca razonarlo a ti. En FxDart, chunk(4) es una sola palabra sobre un stream exactamente igual que sobre una lista — cruzar de Stream al pipeline cuesta una llamada a fxStream, y con ella viene todo el vocabulario de operadores.

Benchmark

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

Caso async: la escala principal es N = 100,000, no 1,000,000. Cada elemento cuesta una vuelta del event loop en ambos lados, así que un millón de awaits reales mediría el event loop de Dart durante minutos — no el pipeline. Los retardos son de longitud cero y se conserva el límite de concurrencia del ejemplo; lo que comparan las barras es la maquinaria del pipeline.

N = 100

Tiempo Empate

Dart nativo 37 µs

FxDart 42 µs

Memoria pico Empate

Dart nativo 16.5 MB

FxDart 17.2 MB

N = 10,000

Tiempo Empate

Dart nativo 3.30 ms

FxDart 3.47 ms

Memoria pico Gana nativo

Dart nativo 24.0 MB

FxDart 25.2 MB

N = 100,000

Tiempo Gana nativo

Dart nativo 32.8 ms

FxDart 36.6 ms

Memoria pico Empate

Dart nativo 75.0 MB

FxDart 74.8 MB

Las barras son medianas de iteraciones cronometradas repetidas en procesos nuevos por lado (los N pequeños se agrupan por resolución del temporizador). Dos lados a menos del 5% entre sí — o a menos de 0.6 ms, una diferencia que nadie puede percibir — cuentan como empate; las carreras relativas ajustadas se vuelven a medir hasta 5 veces. En una app, cualquier cosa por debajo de unos pocos milisegundos es invisible para el usuario, gane la barra que gane. La memoria es el RSS pico del proceso. La VM de Dart y el dataset son idénticos en ambos lados, así que la diferencia entre las dos barras es lo que retiene el pipeline en sí.