`switchLatest`, `flattenMerge` & friends

A stream of streams, flattened: keep the newest, run them all, play them in order, or ignore the extra — and start later sources immediately with 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)

Lecture

Sometimes the events already are inner streams — a socket per session, a request already built. There is no mapper to write; the only question is the flattening policy. That is what these identity forms are: switchLatest is switchMap((s) => s), flattenMerge is mergeMap, flattenConcat is concatMap, exhaustLatest is exhaustMap. An FxEvents of FxEvents flattens as .map((e) => e.stream).switchLatest().

switchLatest mirrors only the newest inner stream: a fresh one cancels the previous mid-flight. Use it when older inners become worthless — the current tab's feed, the current query's results. The chain closes when the outer has closed and the last inner completes.

The other three are the rest of the policy table. flattenMerge runs every inner at once (cap with concurrent: n); flattenConcat plays each to completion before the next begins; exhaustLatest keeps the first and ignores inners that arrive while one is still running.

concatEager is the sibling of FxEvents.concat. Both emit in source order, but concat waits to subscribe to the next source until the current one completes — a cold later source has not even started. concatEager subscribes to every source immediately and buffers later events until their turn. That is how you start a request now and still play the responses in order. fxdart events layer, after Rx's switchAll, mergeAll, concatAll, exhaustAll and concatEager.

Demo 1 · switchLatest — newest inner wins

Demo 2 · flattenConcat vs 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.
}

Try it yourself

Exercise: concatEager vs concat — later starts immediately.

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]
}

Related: switchMap — the mapped form of switchLatest · mergeMap — mergeMap, concatMap, exhaustMap · FxEvents.concat — subscribe-later sibling of concatEager

switchLatest, flattenMerge & friends

Lecture

Demo 1 · switchLatest — newest inner wins

Demo 2 · flattenConcat vs switchLatest

Try it yourself

`switchLatest`, `flattenMerge` & friends