`switchLatest`, `flattenMerge` & friends

Un stream de streams, aplanado: quédate con el más nuevo, ejecútalos todos, reprodúcelos en orden, o ignora los de más — y arranca fuentes posteriores de inmediato con 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)

Lección

A veces los eventos ya son streams internos — un socket por sesión, una petición ya construida. No hay mapper que escribir; la única pregunta es la política de aplanado. Eso es lo que son estas formas identidad: switchLatest es switchMap((s) => s), flattenMerge es mergeMap, flattenConcat es concatMap, exhaustLatest es exhaustMap. Un FxEvents de FxEvents se aplana como .map((e) => e.stream).switchLatest().

switchLatest refleja solo el stream interno más nuevo: uno fresco cancela el anterior en pleno vuelo. Úsalo cuando los internos viejos se vuelven inútiles — el feed de la pestaña actual, los resultados de la consulta actual. La cadena se cierra cuando el externo se ha cerrado y el último interno completa.

Los otros tres son el resto de la tabla de políticas. flattenMerge ejecuta todos los internos a la vez (acota con concurrent: n); flattenConcat reproduce cada uno hasta el final antes de que empiece el siguiente; exhaustLatest se queda con el primero e ignora los internos que llegan mientras uno sigue en marcha.

concatEager es el hermano de FxEvents.concat. Ambos emiten en el orden de la fuente, pero concat espera a suscribirse a la siguiente fuente hasta que la actual completa — una fuente posterior fría ni siquiera ha arrancado. concatEager se suscribe a todas las fuentes de inmediato y almacena en búfer los eventos posteriores hasta que les toca. Así es como arrancas una petición ahora y aun así reproduces las respuestas en orden. Capa de eventos de fxdart, siguiendo a switchAll, mergeAll, concatAll, exhaustAll y concatEager de Rx.

Demo 1 · switchLatest — gana el interno más nuevo

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

Pruébalo tú

Ejercicio: concatEager frente a concat — las posteriores arrancan de inmediato.

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

Relacionado: switchMap — la forma mapeada de switchLatest · mergeMap — mergeMap, concatMap, exhaustMap · FxEvents.concat — el hermano que se suscribe después de concatEager

switchLatest, flattenMerge & friends

Lección

Demo 1 · switchLatest — gana el interno más nuevo

Demo 2 · flattenConcat frente a switchLatest

Pruébalo tú

`switchLatest`, `flattenMerge` & friends