`groupsBy`

Live groups as they open: a key and an inner stream of every later value that shares it.

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)

Lecture

The pull layer's groupBy is a terminal: it pulls everything and hands you a Map. groupsBy is the live version. The first value of each key emits a GroupedEvents — a record of that key and an inner events stream — and every later value with the same key is forwarded to that inner. Groups appear in first-seen key order, as they open, not after the source has closed.

That is the same nested-FxEvents idea as window*, keyed by value instead of by time. A widget can subscribe to one group's inner the moment it appears and see later values as they arrive; it does not have to wait for a batch.

If lastFor is set, the first event (or completion) of lastFor(key) completes that group. A later value with the same key opens a new one — an idle timeout per user, a "session ended" signal per room. Without lastFor, groups stay open until the source completes (or errors, which errors every live group then the outer).

Cancelling the outer completes live groups silently, the same RxJS 9 rule the window family follows. fxdart events layer, after Rx's groupBy.

Demo 1 · Groups as they open

Demo 2 · lastFor closing a group

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

Try it yourself

Exercise: group SKUs by department prefix.

Related: groupBy — the same grouping as a keyed Map, on the pull layer · groupedBy — pull-layer groups as chainable (key, items) records · window* — live inners rotated by count, trigger, or clock