`stopOn` & `startOn`

두 번째 스트림이 조종하는 두 개의 게이트: stopOn은 트리거가 발화하면 체인을 닫고, startOn은 엽니다.

FxEvents<T> FxEvents<T>.stopOn(Stream<void> trigger) // chain (events) FxEvents<T> FxEvents<T>.startOn(Stream<void> trigger) // chain (events)

강의

푸시 체인에는 자연스러운 끝이 없습니다. 풀 파이프라인은 소비자가 묻기를 멈추면 멈추지만, 라이브 피드 — 소켓, 센서, Stream.periodic — 는 무언가가 취소할 때까지 계속 생산합니다. 그걸 잊는 것이 전형적인 누수입니다: 화면은 폐기되었고, 위젯은 사라졌는데, 구독은 여전히 깨어 있고, 여전히 할당하고, 여전히 시체를 향해 setState를 호출합니다.

stopOn(trigger)은 종료를 "기억해서 취소해야 하는 변수"가 아니라 체인의 일부로 만듭니다. trigger의 첫 이벤트가 출력을 닫고 두 구독을 모두 취소합니다 — 소스의 것과 트리거의 것 모두. 트리거가 무엇을 실어 나르는지는 전혀 읽지 않고 발화했다는 사실만 보므로, 타입은 Stream<void>이고 어떤 스트림이든 쓸 수 있습니다.

startOn(trigger)은 거울상입니다: 트리거가 한 번 발화할 때까지 소스 이벤트는 버려지고, 그 후로는 영원히 통과합니다. "준비될 때까지 기다리는" 게이트입니다 — 세션이 로드되기 전의 탭에는 반응하지 마세요. 값을 앞에 덧붙이는 startWith와는 무관하다는 점에 유의하세요. 두 이름은 가깝게 붙어 있지만 상당히 다른 것을 뜻합니다.

…On 접미사는 "트리거 스트림이 조종한다"는 이벤트 레이어의 관례로, sampleOn, chunkOn과 공유합니다. fxdart 이벤트 레이어, Rx의 takeUntil과 skipUntil을 따랐습니다 — 이름이 다른 이유는 Fx.takeUntil이 풀 쪽에서 이미 술어 기반 takeUntilInclusive를 뜻하기 때문입니다. 한 라이브러리 안에서 한 이름이 두 가지를 뜻할 수는 없습니다.

데모 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 {
  // A telemetry feed with no natural end — it would run for 5 seconds.
  final telemetry = timed([
    (0, 'cpu 12%'),
    (60, 'cpu 31%'),
    (120, 'cpu 44%'),
    (300, 'cpu 51%'),
  ], 5000);

// The screen is torn down at 200ms.
  final disposed = timed([(200, null)], 240);

final watch = Stopwatch()..start();
  final seen = await fxEvents(telemetry).stopOn(disposed).toList();
  watch.stop();

print(seen); // [cpu 12%, cpu 31%, cpu 44%]
  print('closed after ~${(watch.elapsedMilliseconds / 100).round() * 100}ms');
  // stopOn did not wait out the source's 5 seconds: the first trigger
  // event closed the chain and cancelled BOTH subscriptions.
}

데모 2 · 켜는 스위치

직접 해 보기

연습: 두 게이트로 만드는 세션 윈도.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise: record only the ticks that fall inside a session.
  final ticks = timed([
    (0, 1),
    (150, 2),
    (300, 3),
    (450, 4),
    (600, 5),
    (750, 6),
    (900, 7),
    (1050, 8),
  ], 1200);

final loggedIn = timed([(375, null)], 1300);
  final loggedOut = timed([(825, null)], 1300);

final session = await fxEvents(ticks)
      .startOn(loggedIn)
      .stopOn(loggedOut)
      .toList();

print(session); // [4, 5, 6]
  // The two gates compose: startOn opens at 375ms, stopOn closes at
  // 825ms, and the ticks outside that window never reach the chain.
}

관련 항목: FxSubscriptions — 정리의 나머지 절반: 여러 구독을 한 번에 취소 · sampleOn — 같은 트리거 관례, 멈추는 대신 읽기 위해 · race — 누가 먼저 말하는지가 결정하는 취소

stopOn & startOn

강의

데모 1 · 끄는 스위치

데모 2 · 켜는 스위치

직접 해 보기

`stopOn` & `startOn`