Streams

A Future delivers one value. A Stream delivers many, over time: lines from a file, messages from a socket, events from a user. The same await machinery works on it, with await for.

Stream<int> countTo(int n) async* {
  for (var i = 1; i <= n; i++) {
    await Future<void>.delayed(const Duration(milliseconds: 10));
    yield i;                       // async* + yield = a stream, written as a loop
  }
}

Future<void> main() async {
  await for (final value in countTo(3)) {
    print('got $value');
  }
  print('stream closed');
}

async* and yield are the readable way to produce a stream. yield* delegates to another stream, the way ... spreads a list.

listen, and the three things a stream does

await for is sugar. Underneath, a stream tells you about data, errors and the end - and listen is where all three are visible.

import 'dart:async';

Stream<int> numbers() async* {
  yield 1;
  yield 2;
  throw StateError('something went wrong');
}

Future<void> main() async {
  final done = Completer<void>();

  numbers().listen(
    (value) => print('data: $value'),
    onError: (Object e) => print('error: $e'),
    onDone: () {
      print('done');
      done.complete();
    },
    cancelOnError: false,
  );

  await done.future;
}

Completer is how you turn "tell me when this is finished" into a future to await. Note that onDone still runs after the error, because cancelOnError is false - set it to true and the subscription ends at the first error instead.

Single-subscription and broadcast

A stream from a file or a socket can be listened to once: the data is consumed as it arrives and there is nowhere to keep it for a second listener. That is a single-subscription stream, and it is the default.

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

  print(await single.length);
  try {
    await single.first;            // the second listen fails
  } on StateError catch (e) {
    print('second listen: ${e.message}');
  }

  // A broadcast stream allows many listeners, and delivers only what arrives
  // after each one subscribes.
  final broadcast = Stream<int>.fromIterable([1, 2, 3]).asBroadcastStream();
  broadcast.listen((v) => print('listener A: $v'));
  broadcast.listen((v) => print('listener B: $v'));
  await Future<void>.delayed(const Duration(milliseconds: 20));
}

StreamController, for when the events come from elsewhere

A controller is a stream you push into - the bridge between a callback-shaped API and the stream world.

import 'dart:async';

Future<void> main() async {
  final controller = StreamController<String>();

  // Something else - a socket, a button, a timer - pushes events in.
  Future<void>(() async {
    for (final word in ['first', 'second', 'third']) {
      await Future<void>.delayed(const Duration(milliseconds: 10));
      controller.add(word);
    }
    await controller.close();      // without this, await for never finishes
  });

  await for (final word in controller.stream) {
    print(word);
  }
  print('closed');
}

Forgetting close() is the classic stream bug: the program simply never finishes, with no error to explain why.

Transforming a stream

The Iterable vocabulary is here too, and it is still lazy.

Stream<int> ticks(int n) async* {
  for (var i = 1; i <= n; i++) {
    yield i;
  }
}

Future<void> main() async {
  final result = await ticks(10)
      .where((n) => n.isEven)
      .map((n) => n * n)
      .take(3)
      .toList();
  print(result);

  // asyncMap awaits each result before emitting it, keeping order.
  final resolved = await ticks(3).asyncMap((n) async {
    await Future<void>.delayed(Duration(milliseconds: 10 * (4 - n)));
    return 'item $n';
  }).toList();
  print(resolved);

  print(await ticks(5).fold<int>(0, (sum, n) => sum + n));
}

What to take away

• A stream is many values over time; await for consumes one at a time.
• async* with yield writes a producer as an ordinary loop.
• listen exposes all three channels: data, error, and done.
• Single-subscription is the default; asBroadcastStream allows more listeners.
• A StreamController bridges callback APIs into streams - and must be closed.
• Stream transformations are lazy, like Iterable's, and asyncMap preserves order.