`chunk`, `chunkOn` & `chunkEvery`

이벤트를 리스트로 모읍니다 — 개수로, 트리거로, 또는 시계로 — 수다스러운 스트림이 몇 번의 묶인 호출이 되도록.

FxEvents<List<T>> FxEvents<T>.chunk(int count) // chain (events) FxEvents<List<T>> FxEvents<T>.chunkOn(Stream<void> trigger) FxEvents<List<T>> FxEvents<T>.chunkEvery(Duration window)

강의

묶기는 수다스러운 스트림이 얻을 수 있는 가장 값싼 성능 개선입니다. 분석 이벤트 백 개는 하나씩 보내면 왕복 백 번이고, 오십 개씩 묶으면 왕복 두 번입니다. 일의 양은 같고 비용은 다릅니다. 풀 레이어는 chunk로 개수로 묶고, 할 수 있는 건 그것뿐입니다 — 풀 파이프라인에는 시계가 없으므로 "지난 2초 동안 일어난 모든 것"은 물을 수 없는 질문입니다.

푸시 쪽은 물을 수 있습니다. chunk(count)는 같은 고정 크기 묶기이고, 소스가 닫힐 때 짧은 마지막 묶음을 흘려보내 아무것도 남겨지지 않게 합니다. chunkEvery(window)는 대신 시간으로 묶습니다: 지난 창 동안 도착한 것이 하나의 리스트로 나옵니다. 그리고 chunkOn(trigger)은 그 결정을 두 번째 스트림에 넘깁니다 — 사용자가 스크롤할 때, 프레임이 끝날 때, 연결이 돌아올 때 묶으세요.

시간 기반 두 형태는 모두 빈 창에서 침묵합니다. 버퍼가 빈 채로 발화한 틱은 빈 리스트 대신 아무것도 내보내지 않으므로, 하류 코드가 "아무 일도 없었다"는 뜻의 묶음을 걸러낼 일이 없습니다 — sampleOn이 지키는 것과 같은 정직함의 규칙입니다. 소스가 닫힐 때 아직 버퍼에 있는 것은 닫히기 전에 흘려보냅니다.

fxdart 이벤트 레이어, Rx의 bufferCount, buffer, bufferTime을 따랐습니다. 이 가족은 풀 레이어의 chunk를 어근으로 유지합니다 — 라이브러리의 양쪽 절반에서 하나의 개념에 하나의 이름 — 트리거에는 …On, 시계에는 …Every 접미사를 붙여서요.

데모 1 · 시계로 묶기

데모 2 · 개수로, 그리고 트리거로

import 'dart:async';

import 'package:fxdart/fxdart.dart';

/// Emits each (offsetMs, value) pair at its offset, closing at [closeMs].
Stream<T> timed<T>(List<(int, T)> events, int closeMs) {
  final c = StreamController<T>();
  for (final (ms, v) in events) {
    Timer(Duration(milliseconds: ms), () => c.add(v));
  }
  Timer(Duration(milliseconds: closeMs), c.close);
  return c.stream;
}

Future<void> main() async {
  // By count: fixed-size batches, with a short final one.
  final byCount =
      await fxEvents(Stream.fromIterable([1, 2, 3, 4, 5])).chunk(2).toList();
  print(byCount); // [[1, 2], [3, 4], [5]]

// By trigger: someone else decides when a batch is due.
  final rows = timed([(0, 'a'), (40, 'b'), (200, 'c')], 400);
  final flush = timed([(100, null), (150, null), (300, null)], 420);

final byTrigger = await fxEvents(rows).chunkOn(flush).toList();
  print(byTrigger); // [[a, b], [c]]
  // The 100ms flush took [a, b]; the 150ms flush found an empty buffer
  // and stayed silent; the 300ms flush took [c].
}

직접 해 보기

연습: 클릭마다 리포트 하나 대신 창마다 요약 하나.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise: turn a noisy click stream into a per-window summary
  // instead of one report per click.
  final clicks = timed([
    (0, 'home'),
    (40, 'home'),
    (80, 'search'),
    (250, 'cart'),
    (290, 'cart'),
    (330, 'checkout'),
  ], 620);

final report = await fxEvents(clicks)
      .chunkEvery(const Duration(milliseconds: 200))
      .map((batch) => '${batch.length} clicks: ${batch.toSet().join(", ")}')
      .toList();

report.forEach(print);
  // 3 clicks: home, search
  // 3 clicks: cart, checkout
  //
  // chunk* is the push-side counterpart of the pull layer's chunk: same
  // idea, but a window can be closed by a clock or a trigger, which a
  // pull pipeline has no way to express.
}

관련 항목: chunk — Iterable을 개수로 묶는 풀 레이어의 원본 · throttle — 창마다 전부가 아니라 하나만 원할 때 · spaceBy — 버스트를 늦추는 다른 방법: 묶는 대신 늘이기

chunk, chunkOn & chunkEvery

강의

데모 1 · 시계로 묶기

데모 2 · 개수로, 그리고 트리거로

직접 해 보기

`chunk`, `chunkOn` & `chunkEvery`