Streams in Dart

Analogy categoryAnalogies

What Is a Stream?

A Stream<T> models a sequence of asynchronous events — zero, one, or many values of type T, plus optionally an error, arriving over time — in contrast to a Future<T> which resolves exactly once. Streams underlie things like incoming WebSocket messages, chunks read from a large file, keystrokes from a text field, or periodic timer ticks from Stream.periodic(). Internally, most custom streams are built and fed data with a StreamController, whose .sink lets you add() events and .stream exposes the read side to consumers, with close() signalling that no further events will arrive.

🏏

Cricket analogy: A Stream is like ball-by-ball commentary during a T20 over — you don't get one final score, you get a sequence of events (each delivery, each boundary, each wicket) arriving one after another until the over ends, unlike a Future which only reports the final result.

Listening to Streams: listen() and await for

You consume a Stream either by calling .listen(onData, onError: ..., onDone: ...), which returns a StreamSubscription you can pause(), resume(), or cancel() to stop receiving events, or — inside an async function — with an await for (var event in stream) { ... } loop that processes events sequentially and completes when the stream closes. The await-for form is convenient for simple sequential processing but, unlike listen(), it can't easily be paused or run alongside other work in the same function without additional coordination.

🏏

Cricket analogy: Using .listen() is like a TV director cutting to different camera angles as each ball is bowled, keeping a subscription open until stumps, while await-for is like a scorer manually logging every ball in order on a paper scoresheet, one at a time.

dart

Stream<int> countStream(int max) async* {
  for (int i = 1; i <= max; i++) {
    await Future.delayed(Duration(milliseconds: 200));
    yield i;
  }
}

Future<void> main() async {
  // Consume with await-for
  await for (final value in countStream(5)) {
    print('await-for: $value');
  }

  // Consume with listen()
  final subscription = countStream(3).listen(
    (value) => print('listen: $value'),
    onError: (e) => print('Stream error: $e'),
    onDone: () => print('Stream done'),
  );
  await Future.delayed(Duration(seconds: 1));
  await subscription.cancel();
}

Single-Subscription vs Broadcast Streams

By default, streams returned by APIs like File.openRead() or an http response body are single-subscription: they can only be listened to once, and attempting a second .listen() throws a StateError, because the underlying source (like a file handle) typically can't be replayed. Broadcast streams, created with StreamController.broadcast() or converted with .asBroadcastStream(), allow multiple simultaneous listeners — useful for things like a shared WebSocket connection or an app-wide event bus — but they don't buffer events for listeners that subscribe late, and calling listen() before any data flows is essential if you don't want to miss early events.

🏏

Cricket analogy: A single-subscription stream is like a courtside seat pass valid for exactly one match — once used, it's gone — while a broadcast stream is like the TV broadcast of that same match, which any number of households can tune into simultaneously, though latecomers miss the overs already bowled.

StreamController.broadcast() is the right choice for things like an app-wide event bus or a shared WebSocket connection that multiple widgets or services need to observe simultaneously — just remember late subscribers won't see events emitted before they called listen().

Transforming Streams: map, where, and StreamTransformer

Streams support the same functional-style operators as Iterables — .map() to transform each event, .where() to filter, .take()/.skip() to limit — all applied lazily so nothing runs until a listener subscribes. For more involved logic, such as debouncing rapid events or decoding chunked bytes into UTF-8 strings, you implement or reuse a StreamTransformer via stream.transform(transformer), which is exactly how utf8.decoder or const LineSplitter() work on byte streams. You can also produce a Stream directly by writing an async* generator function that uses yield to emit values one at a time, which Dart compiles into a properly lazy, pull-based Stream.

🏏

Cricket analogy: Using .where() on a stream to filter only boundary-scoring deliveries is like a highlights editor cutting a full day's play down to just fours and sixes, while a StreamTransformer decoding chunked bytes is like a translator converting raw stadium PA announcements into subtitles in real time.

Forgetting to call subscription.cancel() when a widget or object is disposed is a common source of memory leaks and 'setState() called after dispose()' style errors — it also risks trying to add events to an already-closed StreamController, which throws a StateError.

Stream<T> models a sequence of async events (data or errors) arriving over time, unlike Future's single result.
listen(onData, onError, onDone) returns a StreamSubscription for pause/resume/cancel control.
await for provides simple sequential consumption inside async functions.
Single-subscription streams (files, HTTP bodies) support exactly one listener; broadcast streams support many.
Broadcast streams don't replay missed events to late subscribers — subscribe early.
map(), where(), and StreamTransformer transform events lazily, only when listened to.
Forgetting to cancel a subscription can cause memory leaks or errors on already-closed controllers.