`fxEvents`

평범한 Dart Stream을 체이닝 가능한 FxEvents로 감쌉니다 — FxDart push 세계로 들어가는 입구입니다.

FxEvents<T> fxEvents<T>(Stream<T> stream) class FxEvents<T> { Stream<T> get stream; // unwrap FxEvents<R> map<R>(R Function(T a) f); FxEvents<T> where(bool Function(T a) f); FxEvents<R> asyncMap<R>(FutureOr<R> Function(T a) f); FxEvents<T> startWith(T value); FxEvents<R> scan<R>(R Function(R acc, T a) f, R seed); // seed first FxEvents<T> uniqAdjacent() / uniqAdjacentBy<B>(B Function(T a) f); FxEvents<(T, T)> pairwise(); FxEvents<T> take(int count) / drop(int count) / skip(int count); static FxEvents<T> merge<T>(Iterable<Stream<T>> sources); Future<List<T>> toList(); // terminal Future<T?> head() / firstOrNull(); // terminal FxAsync<T> pull(); // cross into the pull model }

강의

이 섹션 이전의 모든 것은 pull이었습니다: 파이프라인은 종결 연산자가 다음 항목을 요구할 때까지 가만히 있습니다. 하지만 어떤 문제는 본질적으로 push입니다 — 키 입력, 센서 판독값, 소켓 메시지는 누가 요청했든 아니든 올 때가 되면 옵니다. 그것이 바로 Dart의 Stream이 모델링하는 세계이고, fxEvents(stream)은 그 세계에 같은 체이닝 경험을 제공합니다: map, where, asyncMap, startWith, FxEvents.merge — 그리고 이 섹션의 나머지가 다루는 시간·결합 연산자들까지.

알아 둘 만한 설계 결정들입니다. FxEvents는 얇은 래퍼이며, 의도적으로 Stream 확장의 집합이 아닙니다 — 그래서 그 연산자들이 같은 파일 안에서 rxdart나 다른 어떤 스트림 라이브러리와도 절대 충돌하지 않습니다. 예외는 진입용 .fxEvents getter 하나뿐이고, 다른 어디서도 쓰지 않는 이름입니다. .fx와 나란히 놓고 비교한 표는 Stream 다리에 있습니다. 체인은 콜드로 유지됩니다: 감싸는 것만으로는 아무것도 듣지 않고, 종결 연산자(toList, head, listen)만이 이벤트를 흐르게 합니다. 그리고 이것은 Rx에서 영감을 받은 fxdart의 확장이지 FxTS의 일부가 아닙니다 — 아이디어는 Rx에서 왔지만, 이름이 pull 계층과 겹칠 때는 pull 쪽 표기가 이깁니다. uniqAdjacent가 distinctUntilChanged를, stopOn이 takeUntil을, head가 first를 대신합니다. 한 단어는 양쪽에서 한 가지를 뜻합니다.

두 개의 탈출구가 여러분을 가두지 않게 지켜 줍니다. .stream은 체인의 어느 지점에서든 Stream 기반 API를 위해 평범한 Stream으로 되돌려 풉니다. 그리고 .pull()은 타입 있는 pull 세계로 건너갑니다: 이벤트들이 FxAsync 체인이 되어 그때부터는 요청에 따라 pull됩니다 — 이벤트가 태어나는 가장자리에서는 push, 수요를 통제하는 중심부에서는 pull입니다.

데모 1 · Stream 위의 콜드 체인

데모 2 · merge, 그리고 pull 세계로 건너가기

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 {
  // Two sensors, one feed: merge interleaves in arrival order and closes
  // once every source has closed.
  final merged = await FxEvents.merge([
    timed([(0, 'kitchen 21.0'), (100, 'kitchen 21.5')], 200),
    timed([(50, 'attic 15.2'), (150, 'attic 15.1')], 220),
  ]).toList();
  print(merged); // [kitchen 21.0, attic 15.2, kitchen 21.5, attic 15.1]

// .pull() crosses into the typed pull world: from here on the events
  // are an FxAsync chain, pulled on demand.
  final odds = await fxEvents(Stream.fromIterable([1, 2, 3, 4, 5]))
      .pull()
      .filter((v) => v.isOdd)
      .map((v) => v * 2)
      .toList();
  print(odds); // [2, 6, 10]
}

직접 해 보기

연습: 튀는 값이 섞인 센서 피드를 정리해 보세요.

관련 항목: Stream 다리 — 경계의 pull 쪽, 그리고 stream.fx와 stream.fxEvents의 비교 · debounce & throttle — 둘 다 FxEvents 형태가 있음 · LiveValue — 이 체인의 현재-값 동반자