`delay`, `spaceBy` & `sample`

이벤트를 시간축에서 옮기는 세 가지 방법: 전부 미루거나, 넓게 펼치거나, 고정된 시계로 최신 것만 읽거나.

FxEvents<T> FxEvents<T>.delay(Duration duration) // chain (events) FxEvents<T> FxEvents<T>.spaceBy(Duration gap) FxEvents<T> FxEvents<T>.sample(Duration period)

강의

속도 제한은 언제나 무언가를 대가로 치르고, 진짜 질문은 무엇을 치르느냐입니다. throttle과 debounce는 이벤트로 치릅니다: 창마다 하나를 남기고 나머지를 버리는데, 이벤트가 연속적인 무언가의 표본이고 오래된 것이 쓸모없을 때 옳습니다. spaceBy(gap)는 대신 시간으로 치릅니다: 모든 이벤트가 살아남아 큐에 들어가고 gap마다 하나씩 풀려납니다. 각 이벤트가 잃어서는 안 되는 개별 명령일 때 — 100ms에 한 번만 허용하는 API로 보낼 메시지 여섯 개 — 옳습니다.

그 거래에는 날카로운 모서리가 있습니다. spaceBy는 버리지 않고 큐에 쌓으므로, gap보다 빠르게 영원히 생산하는 소스는 무한히 자라는 큐를 만듭니다. 이것은 버스트를 위한 것입니다 — 한꺼번에 도착했고 전부 통과해야 하는 묶음 — 진짜로 끝없는 입력을 위한 것이 아닙니다. 그럴 때는 throttle의 손실성이 오히려 기능입니다.

delay(duration)은 셋 중 가장 단순합니다: 스트림 전체가 고정된 양만큼 밀리고, 간격은 그대로이며, 아무것도 버려지지 않습니다. 닫기는 마지막으로 미뤄진 이벤트가 도착하기를 기다리므로 끝에서 잃는 것이 없고, 에러는 즉시 전달됩니다 — 붙들어 둘 가치가 있는 것은 데이터뿐이니까요.

sample(period)는 시계가 내장된 sampleOn입니다 — period마다 최신 값, 새로운 것이 없으면 침묵. 소스가 상태 같은 피드(위치, 온도, 스크롤 오프셋)이고 소비자가 자기 갱신 주기를 가질 때 꺼내 쓰세요. fxdart 이벤트 레이어, Rx의 delay, interval, sampleTime을 따랐습니다.

데모 1 · 버스트를 손실 없이 조절하기

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 {
  // Four requests queued up at once, against an API that allows one
  // call per 150ms.
  final watch = Stopwatch()..start();
  final sent = await fxEvents(Stream.fromIterable(['r1', 'r2', 'r3', 'r4']))
      .spaceBy(const Duration(milliseconds: 150))
      .toList();
  watch.stop();

print(sent); // [r1, r2, r3, r4] — every one of them
  print('took ~${(watch.elapsedMilliseconds / 100).round() * 100}ms');

// throttle, given the same burst, keeps ONE and drops the rest:
  final throttled = await fxEvents(Stream.fromIterable(['r1', 'r2', 'r3', 'r4']))
      .throttle(const Duration(milliseconds: 150))
      .toList();
  print(throttled); // [r1]
  // Same rate limit, opposite trade: spaceBy is lossless and slow,
  // throttle is lossy and immediate.
}

데모 2 · 미루기, 그리고 시계로 읽기

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // delay shifts the whole stream — nothing dropped, spacing preserved.
  final watch = Stopwatch()..start();
  final shifted = await fxEvents(Stream.fromIterable([1, 2, 3]))
      .delay(const Duration(milliseconds: 200))
      .toList();
  print('$shifted after ~${(watch.elapsedMilliseconds / 100).round() * 100}ms');

// sample reads the newest value on its own clock, like sampleOn but
  // with the clock built in.
  final sensor = timed([(0, 'a'), (60, 'b'), (300, 'c')], 640);
  final read = await fxEvents(sensor)
      .sample(const Duration(milliseconds: 200))
      .toList();

print(read); // [b, c]
  // The 200ms tick saw b, the 400ms tick saw c, and the 600ms tick had
  // nothing new to report — so it stayed silent.
}

직접 해 보기

연습: 전송 속도와 보고 속도를 한 체인에.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise: a burst of outbound messages, paced to the wire and
  // then batched for the log.
  final outbox = Stream.fromIterable(['m1', 'm2', 'm3', 'm4', 'm5', 'm6']);

final logged = await fxEvents(outbox)
      .spaceBy(const Duration(milliseconds: 100))
      .chunkEvery(const Duration(milliseconds: 250))
      .toList();

print(logged); // [[m1, m2], [m3, m4], [m5, m6]] (roughly)
  // spaceBy sets the send rate; chunkEvery sets the reporting rate.
  // Because spaceBy queues rather than drops, an unbounded burst grows
  // an unbounded queue — reach for throttle or debounce when the input
  // is genuinely endless.
}

관련 항목: throttle — 손실이 있는 대응물: 창마다 하나, 즉시 · debounce — 버스트가 끝나기를 기다렸다가 마지막 값 취하기 · chunkEvery — 역시 모든 이벤트를 지키되, 펼치는 대신 묶기

delay, spaceBy & sample

강의

데모 1 · 버스트를 손실 없이 조절하기

데모 2 · 미루기, 그리고 시계로 읽기

직접 해 보기

`delay`, `spaceBy` & `sample`