`switchLatest`, `flattenMerge` 및 관련 함수

스트림의 스트림을 평탄화합니다: 최신만 남기거나, 전부 실행하거나, 순서대로 재생하거나, 여분을 무시하거나 — 그리고 concatEager로 나중 소스를 즉시 시작합니다.

FxEvents<T> FxEvents<Stream<T>>.switchLatest() FxEvents<T> FxEvents<Stream<T>>.flattenMerge({int? concurrent}) FxEvents<T> FxEvents<Stream<T>>.flattenConcat() FxEvents<T> FxEvents<Stream<T>>.exhaustLatest() FxEvents<T> concatEager<T>(Iterable<Stream<T>> sources)

강의

이벤트가 이미 내부 스트림인 경우가 있습니다 — 세션마다 소켓 하나, 이미 만들어진 요청. 쓸 매퍼는 없고, 질문은 평탄화 정책 하나뿐입니다. 이 항등 형태들이 그것입니다: switchLatest는 switchMap((s) => s), flattenMerge는 mergeMap, flattenConcat는 concatMap, exhaustLatest는 exhaustMap. FxEvents의 FxEvents는 .map((e) => e.stream).switchLatest()로 평탄화합니다.

switchLatest는 가장 새로운 내부 스트림만 비춥니다: 새것이 오면 이전 것을 진행 도중에 취소합니다. 오래된 내부가 무가치해질 때 쓰세요 — 현재 탭의 피드, 현재 쿼리의 결과. 체인은 아우터가 닫히고 그리고 마지막 내부가 완료됐을 때 닫힙니다.

나머지 셋이 정책 표의 나머지입니다. flattenMerge는 모든 내부를 한꺼번에 실행하고 (concurrent: n으로 한도); flattenConcat는 각각을 완료까지 재생한 뒤 다음을 시작합니다; exhaustLatest는 첫 번째를 지키고, 하나가 도는 동안 도착한 내부를 무시합니다.

concatEager는 FxEvents.concat의 형제입니다. 둘 다 소스 순서로 내보내지만, concat은 현재 것이 완료될 때까지 다음 소스를 구독하지 않습니다 — 콜드인 나중 소스는 시작조차 하지 않은 상태입니다. concatEager는 모든 소스를 즉시 구독하고 나중 이벤트를 자기 차례까지 버퍼에 담습니다. 요청은 지금 시작하고 응답은 여전히 순서대로 재생하는 방법이 그것입니다. fxdart 이벤트 레이어, Rx의 switchAll, mergeAll, concatAll, exhaustAll, concatEager를 따랐습니다.

데모 1 · switchLatest — 최신 내부가 이긴다

데모 2 · flattenConcat 대 switchLatest

import 'dart:async';

import 'package:fxdart/fxdart.dart';

/// Starts its timers on listen, so a later inner is not already spent.
Stream<int> inner(int n) {
  final c = StreamController<int>();
  c.onListen = () {
    Timer(const Duration(milliseconds: 30), () {
      if (c.hasListener) c.add(n);
    });
    Timer(const Duration(milliseconds: 70), () {
      if (c.hasListener) c.add(n * 10);
      if (!c.isClosed) c.close();
    });
  };
  return c.stream;
}

Future<void> main() async {
  List<Stream<int>> twoInners() => [inner(1), inner(2)];

print(await fxEvents(Stream.fromIterable(twoInners()))
      .switchLatest()
      .toList()); // [2, 20]
  // Second inner arrived while the first was still running — first
  // CANCELLED, so 1 and 10 never land.

print(await fxEvents(Stream.fromIterable(twoInners()))
      .flattenConcat()
      .toList()); // [1, 10, 2, 20]
  // flattenConcat waits for each inner to finish; nothing is dropped.
  // flattenMerge would interleave both; exhaustLatest would ignore 2.
}

직접 해 보기

연습: concatEager 대 concat — 나중 소스가 즉시 시작됩니다.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Stream<int> first() {
  final c = StreamController<int>();
  c.onListen = () {
    Timer(const Duration(milliseconds: 40), () {
      c.add(1);
      c.close();
    });
  };
  return c.stream;
}

Stream<int> later(String tag) {
  final c = StreamController<int>();
  c.onListen = () {
    print('started $tag');
    c.add(2);
    c.close();
  };
  return c.stream;
}

Future<void> main() async {
  // concat waits to subscribe to the next source until the current one
  // completes — later does not start until 1 has landed.
  print('concat:');
  final concatOut = FxEvents.concat([first(), later('concat')]).toList();
  print('subscribed concat');
  print(await concatOut); // [1, 2]

// concatEager subscribes to EVERY source immediately and buffers later
  // events until their turn — 'started eager' prints before we even
  // return from toList().
  print('concatEager:');
  final eagerOut = concatEager([first(), later('eager')]).toList();
  print('subscribed eager');
  print(await eagerOut); // [1, 2]
}

관련 항목: switchMap — switchLatest의 매핑 형태 · mergeMap — mergeMap, concatMap, exhaustMap · FxEvents.concat — concatEager의 나중에-구독하는 형제

switchLatest, flattenMerge 및 관련 함수

강의

데모 1 · switchLatest — 최신 내부가 이긴다

데모 2 · flattenConcat 대 switchLatest

직접 해 보기

`switchLatest`, `flattenMerge` 및 관련 함수