센서 스트림에서 윈도우 단위 경고 감지하기

FxDart 승 async

요구사항

보일러 온도 센서가 실제 Dart Stream으로 측정값을 전달합니다 — 10 ms마다 하나씩, 총 12개입니다(아래 코드의 고정 데이터). 스트림을 4개씩 묶은 윈도우로 그룹화하고, 각 윈도우의 평균과 최댓값을 보고하며, 평균이 75.00 이상인 윈도우에는 ALERT 줄을 출력하세요.

FxDart의 답은 스트림 브리지입니다: fxStream이 Stream을 풀 기반(pull-based) 파이프라인으로 끌어올리고, 그 다음부터는 윈도우 나누기가 그저 chunk(4)일 뿐입니다 — 동기 예제에서 쓰인 것과 같은 연산자입니다 — 이어서 map이 각 윈도우를 averageBy와 maxBy로 요약합니다.

예상 출력

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

나란히 보기

네이티브 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');
  }
}

차이가 나는 이유

Dart의 Stream API에는 윈도우 나누기 연산자가 없습니다. 관용적인 선택지는 두 가지입니다: 가변 버퍼를 두는 await for 루프 — 예시처럼 4개를 모으고, 내보내고, 초기화하는 방식 — 아니면 그 똑같은 관리 작업을 커스텀 StreamTransformer로 패키징하는 것인데, 이는 코드가 줄어들기는커녕 오히려 늘어납니다. 어느 쪽이든 버퍼, 플러시 조건, 초기화는 직접 유지보수해야 하고, 마지막 윈도우가 가득 차지 않는 경계 사례도 직접 따져야 합니다. FxDart에서는 chunk(4)가 리스트에서와 똑같이 스트림에서도 한 단어로 끝납니다 — Stream에서 파이프라인으로 넘어가는 데 드는 비용은 fxStream 호출 한 번뿐이고, 연산자 어휘 전체가 함께 딸려 옵니다.

벤치마크

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

비동기 예제입니다. 대표 스케일이 1,000,000이 아니라 N = 100,000인 이유 — 원소 하나마다 양쪽 모두 이벤트 루프를 한 바퀴 돌아야 하므로, 실제 await 백만 번은 파이프라인이 아니라 Dart 이벤트 루프를 몇 분씩 재게 됩니다. 지연은 0으로 두고 예제의 동시성 제한은 그대로 유지합니다. 막대가 비교하는 것은 파이프라인 기계 장치 자체입니다.

N = 100

시간 무승부

네이티브 Dart 37 µs

FxDart 42 µs

최대 메모리 무승부

네이티브 Dart 16.5 MB

FxDart 17.2 MB

N = 10,000

시간 무승부

네이티브 Dart 3.30 ms

FxDart 3.47 ms

최대 메모리 네이티브 승

네이티브 Dart 24.0 MB

FxDart 25.2 MB

N = 100,000

시간 네이티브 승

네이티브 Dart 32.8 ms

FxDart 36.6 ms

최대 메모리 무승부

네이티브 Dart 75.0 MB

FxDart 74.8 MB

막대는 사이드별로 새 프로세스에서 반복 측정한 중앙값입니다(작은 N은 타이머 해상도를 위해 배치 처리). 두 사이드가 서로 5% 이내이거나 — 사람이 지각할 수 없는 차이인 0.6ms 이내이면 — 무승부로 칩니다. 상대 차이가 근소한 경우는 최대 5회까지 다시 측정합니다. 앱에서는 어느 막대가 짧든 몇 밀리초 이하의 차이는 사용자에게 보이지 않습니다. 메모리는 프로세스 최대 RSS입니다. Dart VM과 데이터셋은 양쪽이 동일하므로, 두 막대의 차이가 곧 파이프라인 자체가 붙들고 있는 양입니다.