Streams

26.1 Stream Basics

Future<void> main() async {
  final stream = Stream.fromIterable([1, 2, 3, 4, 5]);

  // Consume with await for
  await for (final value in stream.where((n) => n.isOdd)) {
    print(value);
  }

  // Or with listen
  final subscription = Stream.periodic(const Duration(seconds: 1), (i) => i).take(3).listen(
        (tick) => print('tick $tick'),
        onError: (e) => print('error $e'),
        onDone: () => print('done'),
        cancelOnError: false,
      );
  // subscription.pause(); subscription.resume(); await subscription.cancel();
}

26.2 Single-Subscription vs Broadcast

Buffers events until listened to Events are lost if nobody is listening File reads, HTTP responses UI events, notifications

26.3 StreamController

class ChatRoom {
  final _controller = StreamController<String>.broadcast();
  Stream<String> get messages => _controller.stream;
  void send(String msg) => _controller.add(msg);
  void fail(Object error) => _controller.addError(error);
  Future<void> close() => _controller.close();
}

void main() {
  final room = ChatRoom();
  room.messages.listen((m) => print('A got: $m'));
  room.messages.listen((m) => print('B got: $m'));
  room.send('Hello!');
}

26.4 Transforming Streams

The rxdart package adds BehaviorSubject, debounceTime, switchMap, combineLatest, throttleTime, and more. async (a Dart team package) adds StreamGroup, StreamQueue, CancelableOperation, AsyncMemoizer.

final words = Stream.fromIterable(['apple', 'banana', 'cherry']);
words
    .map((w) => w.toUpperCase())
    .where((w) => w.length > 5)
    .asyncMap((w) async => await translate(w))          // async per-event transform
    .distinct()
    .handleError((e) => print('handled $e'))
    .take(10)
    .listen(print);

final total = await Stream.fromIterable([1, 2, 3]).reduce((a, b) => a + b);
final list = await Stream.fromIterable([1, 2]).toList();
final first = await stream.first;
final expanded = stream.expand((x) => [x, x]);
final timeout = stream.timeout(const Duration(seconds: 5));

// Custom StreamTransformer
final debounce = StreamTransformer<String, String>.fromHandlers(
  handleData: (data, sink) => sink.add(data.trim()),
);
stream.transform(debounce);

// Decoding byte streams (dart:convert)
final lines = File('data.txt').openRead().transform(utf8.decoder).transform(const LineSplitter());
await for (final line in lines) print(line);