`switchMap`

각 이벤트를 내부 스트림으로 매핑하고 최신 것 하나만 비춥니다 — 새 이벤트가 오면 이전 내부 스트림을 진행 도중에 취소합니다.

FxEvents<R> FxEvents<T>.switchMap<R>(Stream<R> Function(T a) f) // chain (events)

강의

모든 UI가 최소 한 번은 출시해 본 검색창 버그입니다: 사용자가 da를 치고 dart를 칩니다. da 요청이 더 느립니다. 그 낡은 결과가 마지막에 도착해 좋은 결과를 덮어씁니다. 해법은 오래된 응답을 무시하는 것이 아니라 아예 불가능하게 만드는 것입니다: switchMap(f)는 각 이벤트를 내부 스트림으로 매핑하고, 더 새 이벤트가 도착하는 순간 이전 내부 스트림을 즉시 취소합니다. 하류로 비춰지는 것은 언제나 가장 새로운 내부 스트림뿐입니다.

새것에-의한-취소는 하나의 정책이고, 더 새로운 입력이 생기는 순간 오래된 작업이 무가치해지는 경우에 정확히 옳은 정책입니다 — 검색, 자동 완성, 내비게이션, "선택된 행의 상세 불러오기". 모든 내부 스트림의 출력이 중요한 경우(예: 파일마다 업로드 하나)에는 틀린 정책입니다 — 그것은 아무것도 취소되지 않는 pull 쪽 mapConcurrent의 팬아웃 작업입니다.

가장자리의 의미론: 체인은 소스가 닫히고 그리고 마지막 내부 스트림이 완료됐을 때 닫힙니다 — 소스가 끝난다고 이미 화면에 오른 작업이 잘리는 일은 없습니다. 매퍼가 던지면 에러 이벤트가 되고 소스는 계속 갑니다. fxdart 이벤트 계층이며, Rx의 switchMap을 따랐습니다.

데모 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 search box: the user refines the query while the first search is
  // still in flight (each search takes ~150ms).
  final queries = timed([(0, 'da'), (60, 'dart')], 400);
  final started = <String>[];

Stream<String> search(String q) {
    started.add(q);
    return timed([(150, 'results for "$q"')], 200);
  }

final out = await fxEvents(queries).switchMap(search).toList();

print('started:   $started'); // [da, dart]
  print('delivered: $out'); // [results for "dart"]
  // Both searches started — but 'dart' arrived while 'da' was in flight,
  // so the older inner stream was CANCELLED mid-request. No stale results
  // can ever overwrite newer ones.
}

데모 2 · 마지막 내부 스트림은 말을 끝맺는다

직접 해 보기

연습: 낡은 제안을 보여 줄 수 없는 자동 완성.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Autocomplete: every keystroke kicks off a ~100ms suggestion fetch.
  final keys = timed([(0, 's'), (40, 'sw'), (80, 'swi')], 300);

Stream<String> suggest(String q) => timed([(100, '$q…suggestions')], 130);

// TODO: only the NEWEST query's suggestions should reach the screen —
  // replace asyncExpand (which dutifully shows every stale result) with
  //   fxEvents(keys).switchMap(suggest).toList()
  final out = await keys.asyncExpand(suggest).toList();

print(out);
  // currently every query's suggestions — want [swi…suggestions]:
  // 's' and 'sw' are superseded before their fetch comes back.
}

관련 항목: debounce — 자연스러운 상류 파트너: 들어가는 쿼리는 줄이고, 나오는 낡은 결과는 없앰 · race — 연속된 스트림이 아니라 형제 스트림 사이의 취소 · mapConcurrent — 모든 결과가 중요할 때, pull 쪽 팬아웃