`fxEvents`

Envuelve un Stream normal de Dart en un FxEvents encadenable: la puerta de entrada al lado push de FxDart.

FxEvents<T> fxEvents<T>(Stream<T> stream) class FxEvents<T> { Stream<T> get stream; // unwrap FxEvents<R> map<R>(R Function(T a) f); FxEvents<T> where(bool Function(T a) f); FxEvents<R> asyncMap<R>(FutureOr<R> Function(T a) f); FxEvents<T> startWith(T value); FxEvents<R> scan<R>(R Function(R acc, T a) f, R seed); // seed first FxEvents<T> uniqAdjacent() / uniqAdjacentBy<B>(B Function(T a) f); FxEvents<(T, T)> pairwise(); FxEvents<T> take(int count) / drop(int count) / skip(int count); static FxEvents<T> merge<T>(Iterable<Stream<T>> sources); Future<List<T>> toList(); // terminal Future<T?> head() / firstOrNull(); // terminal FxAsync<T> pull(); // cross into the pull model }

Lección

Todo lo anterior a esta sección es pull: un pipeline se queda quieto hasta que un operador terminal exige el siguiente elemento. Pero algunos problemas son genuinamente push — las pulsaciones de teclado, las lecturas de sensores, los mensajes de socket llegan cuando llegan, los pida alguien o no. Eso es lo que modela el Stream de Dart, y fxEvents(stream) le da a ese mundo el mismo tratamiento encadenable: map, where, asyncMap, startWith, FxEvents.merge — más los operadores de tiempo y de combinación que cubre el resto de esta sección.

Decisiones de diseño que conviene conocer. FxEvents es un wrapper fino, deliberadamente no un conjunto de extensiones sobre Stream — así sus operadores nunca pueden chocar con rxdart ni con ninguna otra librería de streams en el mismo archivo. La única excepción es el getter de entrada .fxEvents, un solo nombre que nadie más reclama; se compara con .fx en Puentes de Stream. La cadena se mantiene fría: envolver no escucha nada; solo un terminal (toList, head, listen) pone los eventos a fluir. Y es una extensión de fxdart inspirada en Rx, no parte de FxTS — las ideas vienen de Rx, pero cuando un nombre chocaría con el de la capa pull gana la grafía pull: uniqAdjacent en vez de distinctUntilChanged, stopOn en vez de takeUntil, head en vez de first. Una palabra significa una cosa en ambos lados.

Dos vías de escape te mantienen sin ataduras. .stream desenvuelve de vuelta a un Stream normal para cualquier API basada en Stream, en cualquier punto de la cadena. Y .pull() cruza al mundo pull tipado: los eventos se convierten en una cadena FxAsync, consumida bajo demanda a partir de ahí — push en el borde donde nacen los eventos, pull en el núcleo donde tú controlas la demanda.

Demo 1 · Una cadena fría sobre un Stream

Demo 2 · merge, y el cruce al mundo pull

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 {
  // Two sensors, one feed: merge interleaves in arrival order and closes
  // once every source has closed.
  final merged = await FxEvents.merge([
    timed([(0, 'kitchen 21.0'), (100, 'kitchen 21.5')], 200),
    timed([(50, 'attic 15.2'), (150, 'attic 15.1')], 220),
  ]).toList();
  print(merged); // [kitchen 21.0, attic 15.2, kitchen 21.5, attic 15.1]

// .pull() crosses into the typed pull world: from here on the events
  // are an FxAsync chain, pulled on demand.
  final odds = await fxEvents(Stream.fromIterable([1, 2, 3, 4, 5]))
      .pull()
      .filter((v) => v.isOdd)
      .map((v) => v * 2)
      .toList();
  print(odds); // [2, 6, 10]
}

Pruébalo tú

Ejercicio: limpia un feed de sensor con fallos.

Relacionado: Puentes de Stream — el lado pull de la frontera y stream.fx frente a stream.fxEvents · debounce & throttle — ambos tienen forma FxEvents · LiveValue — el compañero de valor actual de esta cadena