`groupsBy`

Grupos en vivo a medida que se abren: una clave y un stream interno de cada valor posterior que la comparte.

class GroupedEvents<K, T> { final K key; final FxEvents<T> events; } FxEvents<GroupedEvents<K, T>> FxEvents<T>.groupsBy<K>(K Function(T value) keyOf, {Stream<void> Function(K key)? lastFor}) // chain (events)

Lección

El groupBy de la capa pull es un terminal: tira de todo y te entrega un Map. groupsBy es la versión en vivo. El primer valor de cada clave emite un GroupedEvents — un registro de esa key y un stream interno events — y cada valor posterior con la misma clave se reenvía a ese interno. Los grupos aparecen en el orden de primera aparición de la clave, a medida que se abren, no después de que la fuente se haya cerrado.

Es la misma idea de FxEvents anidados que window*, claveada por valor en vez de por tiempo. Un widget puede suscribirse al interno de un grupo en el momento en que aparece y ver los valores posteriores conforme llegan; no tiene que esperar a un lote.

Si se define lastFor, el primer evento (o la completación) de lastFor(key) completa ese grupo. Un valor posterior con la misma clave abre uno nuevo — un timeout de inactividad por usuario, una señal de «sesión terminada» por sala. Sin lastFor, los grupos se quedan abiertos hasta que la fuente completa (o falla, lo que hace fallar cada grupo en vivo y luego el exterior).

Cancelar el exterior completa los grupos en vivo en silencio, la misma regla de RxJS 9 que sigue la familia window. Capa de eventos de fxdart, según el groupBy de Rx.

Demo 1 · Grupos a medida que se abren

Demo 2 · lastFor cerrando un grupo

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // lastFor closes a group: the first event (or completion) of
  // lastFor(key) completes that inner. A later value with the same
  // key opens a new one.
  final source = StreamController<String>();
  final lasts = <String, StreamController<void>>{};
  final future = fxEvents(source.stream)
      .groupsBy(
        (s) => s[0],
        lastFor: (key) => (lasts[key] = StreamController<void>()).stream,
      )
      .toList();

source.add('a1');
  source.add('a2');
  await Future<void>.delayed(Duration.zero);
  lasts['a']!.add(null); // close group a
  await Future<void>.delayed(Duration.zero);
  source.add('a3'); // reopens a
  source.add('b1');
  await source.close();

final groups = await future;
  print([for (final g in groups) g.key]); // [a, a, b]
  for (final g in groups) {
    print('${g.key}: ${await g.events.toList()}');
  }
  // a: [a1, a2]
  // a: [a3]
  // b: [b1]
  //
  // Cancelling the outer completes live groups silently, the same
  // RxJS 9 rule window* follows.
}

Pruébalo tú

Ejercicio: agrupa SKUs por el prefijo de departamento.

Relacionado: groupBy — el mismo agrupamiento como un Map con claves, en la capa pull · groupedBy — grupos de la capa pull como registros (key, items) encadenables · window* — internos en vivo rotados por conteo, disparador o reloj