Streams in Dart

This is Part 4 of the Async & Concurrency in Dart series. A Future (Part 2) gives you one value, later. But tons of real-world data isn't one value — it's a sequence arriving over time: messages on a websocket, taps on a button, lines from a file, location updates, search-box keystrokes. For that, you want a Stream.

If a Future is a single restaurant buzzer, a Stream is a conveyor belt: items keep coming down the line until someone flips it off. This post covers what streams are, the two flavors (single-subscription vs broadcast), and the two ways to consume them.

Future vs Stream in one picture

Future<T>  ──────────────[ value ]X            (one event, then done)

Stream<T>  ──[v]──[v]──[v]──[v]──[v]──X         (many events, then done)

| | Future<T> | Stream<T> | | --- | --- | --- | | Delivers | exactly one value (or error) | zero-to-many values over time | | Consume with | await / .then() | await for / .listen() | | Real-world fit | "fetch this user" | "every message on this socket" |

A Stream<T> can emit any number of data events (each a T), interleave error events, and finally a single done event that signals "no more items." That's the whole vocabulary: data, error, done.

Creating a stream quickly (so we have something to listen to)

We'll go deep on creating streams in Part 5 and Part 6. For now, two one-liners:

// Emits 0,1,2,3,4 — one per call, from an iterable.
Stream<int> fromList = Stream.fromIterable([0, 1, 2, 3, 4]);

// Emits an incrementing int every second, forever.
Stream<int> ticker = Stream.periodic(const Duration(seconds: 1), (i) => i);

Two ways to listen

1. await for — read a stream like a loop

Inside an async function, await for pulls each event as it arrives, pausing the loop between them. It reads like a for loop, but it's asynchronous:

Future<int> sumStream(Stream<int> stream) async {
  var sum = 0;
  await for (final value in stream) { // waits for each event
    sum += value;
  }
  return sum; // reached only after the stream is DONE
}

void main() async {
  print(await sumStream(Stream.fromIterable([1, 2, 3, 4]))); // 10
}

await for is the most readable option when you want to process every event and then continue after the stream finishes. Two cautions:

• The line after the loop runs only when the stream completes (sends "done"). On an infinite stream (like Stream.periodic), await for never exits — you'd need break or a bounded stream.
• An error event thrown by the stream is thrown out of the await for, so you can wrap it in try/catch.

2. .listen() — register callbacks

.listen() is the lower-level, more flexible API. You hand it callbacks for each event type, and it returns a StreamSubscription — your handle to control the listening:

final subscription = ticker.listen(
  (value) => print('data: $value'), // onData — called per event
  onError: (e) => print('error: $e'),
  onDone: () => print('done'),
  cancelOnError: false,
);

.listen() is the right choice when you don't want to block in an await for loop — e.g. you're wiring up a subscription in initState and want to keep going. In fact, every stream method is built on .listen() under the hood; it's the foundational operation.

The StreamSubscription: pause, resume, cancel

.listen() returns a StreamSubscription<T>, which is your remote control for the conveyor belt:

final sub = ticker.listen((v) => print(v));

sub.pause();        // stop receiving (events buffer up)
sub.resume();       // start receiving again
await sub.cancel(); // stop for good and release resources

🔑 The cancel rule that prevents memory leaks. A subscription keeps the stream alive and keeps your callback referenced. If you don't cancel() it, you leak — the classic Flutter bug. Always cancel subscriptions in dispose():

StreamSubscription<int>? _sub;

@override
void initState() {
  super.initState();
  _sub = ticker.listen(_onTick);
}

@override
void dispose() {
  _sub?.cancel(); // ✅ stop listening; no leak, no "setState after dispose"
  super.dispose();
}

pause() is handy too: while paused, the stream's events are buffered (for single-subscription streams) until you resume() — useful for backpressure when a consumer can't keep up.

Single-subscription vs broadcast

This is the distinction the interview question hinges on. There are two kinds of streams:

Single-subscription streams (the default)

• Can be listened to exactly once. Call .listen() a second time and you get a StateError.
• Hold events until that one listener subscribes, then deliver them in order, no gaps.
• Model a complete sequence where every event matters: reading a file, the body of an HTTP response, a one-shot pipeline.

final s = Stream.fromIterable([1, 2, 3]);
s.listen((v) => print('A: $v'));
s.listen((v) => print('B: $v')); // ❌ StateError: already listened to

Broadcast streams

• Can have many listeners, and listeners can come and go at any time.
• Do not wait for a listener — events fired while nobody's listening are simply missed. A late subscriber only sees events from the moment it subscribes onward.
• Model ongoing happenings that exist whether or not anyone cares: mouse/keyboard events, app lifecycle, a shared bus.

Convert a single-subscription stream into a broadcast one with .asBroadcastStream(), or create one directly (we'll do that with StreamController.broadcast() in Part 5):

final broadcast = Stream.periodic(const Duration(seconds: 1), (i) => i)
    .asBroadcastStream();

broadcast.listen((v) => print('listener 1: $v'));
broadcast.listen((v) => print('listener 2: $v')); // ✅ allowed

| | Single-subscription | Broadcast | | --- | --- | --- | | Number of listeners | exactly one | many | | Waits for a listener? | yes (buffers until subscribed) | no (events fire regardless) | | Late subscriber sees past events? | n/a (only one listener) | no — only future events | | Typical use | files, HTTP bodies, one-shot pipelines | UI events, app-wide buses |

Why the default is single-subscription: it guarantees the listener sees the whole sequence from the start. Broadcast trades that guarantee for flexibility (multiple, transient listeners). Pick single-subscription unless you genuinely need multiple listeners.

Transforming streams: the pipeline

Streams have Iterable-like methods that each return a new stream, so you can build readable pipelines. They're lazy — nothing flows until something listens.

Stream.fromIterable([1, 2, 3, 4, 5, 6])
    .where((n) => n.isEven)   // keep evens: 2,4,6
    .map((n) => n * 10)        // transform: 20,40,60
    .take(2)                   // first two: 20,40
    .listen(print);            // 20  40

Handy members you'll reach for constantly:

• map, where, expand, take, skip, distinct — shape/filter the events.
• asyncMap — like map but the transform returns a Future and is awaited per event (great for async work per item).
• handleError, timeout — error/latency control.
• Terminal reducers that return a Future (they consume the whole stream): toList(), first, last, length, reduce, fold, join, contains, any, every.

// asyncMap: do async work per event, in order.
Stream<String> urls = Stream.fromIterable(['/a', '/b', '/c']);
urls.asyncMap((u) => http.get(u)).listen(print);

// Terminal reducer collapses a stream into a single Future.
Future<List<int>> all = Stream.fromIterable([1, 2, 3]).toList(); // Future<[1,2,3]>

Notice the symmetry with Part 2: a terminal operation turns a Stream<T> back into a Future<...>, because you've collapsed "many over time" into "one final answer."

Errors and done in await for

Because errors arrive as events, you handle them with ordinary try/catch around the loop, and the loop simply ends on "done":

Future<void> consume(Stream<int> stream) async {
  try {
    await for (final v in stream) {
      print('got $v');
    }
    print('stream finished cleanly'); // runs on the DONE event
  } catch (e) {
    print('stream errored: $e');
  }
}

With .listen(), the same three outcomes map to the three callbacks: onData, onError, onDone. (More on stream error nuances — like whether an error ends the stream — in Part 7.)

Flutter tie-in: StreamBuilder

Just as FutureBuilder renders a future, StreamBuilder rebuilds your UI on every stream event — the idiomatic way to show live data:

StreamBuilder<int>(
  stream: ticker, // a Stream<int>
  initialData: 0,
  builder: (context, snapshot) {
    if (snapshot.hasError) return Text('Error: ${snapshot.error}');
    return Text('Tick: ${snapshot.data}');
  },
)

⚠️ A StreamBuilder calls .listen() internally, so feed it a broadcast stream if multiple widgets must listen, and beware re-creating the stream inside build (same trap as FutureBuilder in Part 3).

Practice Challenges

Challenge 1 — Sum with await for. Write a function that returns the sum of a Stream<int>.

Future<int> total(Stream<int> s) async {
  var sum = 0;
  await for (final v in s) {
    sum += v;
  }
  return sum;
}

Challenge 2 — Cancel to stop an infinite stream. Use .listen() on Stream.periodic and stop after 3 events.

import 'dart:async';

void main() {
  var count = 0;
  late StreamSubscription<int> sub;
  sub = Stream.periodic(const Duration(milliseconds: 200), (i) => i)
      .listen((v) {
    print(v);
    if (++count == 3) sub.cancel(); // stop after three
  });
}

Challenge 3 — Why does the second listen throw? Fix it so both listeners work.

final s = Stream.fromIterable([1, 2, 3]);
s.listen((v) => print('A $v'));
s.listen((v) => print('B $v')); // throws

final s = Stream.fromIterable([1, 2, 3]).asBroadcastStream();
s.listen((v) => print('A $v'));
s.listen((v) => print('B $v')); // now allowed

Challenge 4 — Build a pipeline. From a stream of integers 1..10, print the squares of the odd numbers, only the first three.

Stream.fromIterable(List.generate(10, (i) => i + 1))
    .where((n) => n.isOdd) // 1,3,5,7,9
    .map((n) => n * n)      // 1,9,25,49,81
    .take(3)                // 1,9,25
    .listen(print);

Challenge 5 — No-leak subscription. Sketch a Flutter State that listens to a stream and cleans up correctly.

StreamSubscription<int>? _sub;

@override
void initState() {
  super.initState();
  _sub = ticker.listen((v) {
    if (!mounted) return;
    setState(() => _value = v);
  });
}

@override
void dispose() {
  _sub?.cancel(); // critical: stop listening, release resources
  super.dispose();
}

Questions to test yourself

Q1 (basic). What's the core difference between a Future and a Stream?

Q2 (basic). Name the two ways to consume a stream and when you'd pick each.

Q3 (intermediate). What happens if you call .listen() twice on a single-subscription stream? On a broadcast stream?

Q4 (intermediate). Why must you cancel a StreamSubscription, and where in a Flutter widget do you do it?

Q5 (advanced). A late subscriber to a broadcast stream is missing earlier events. Why, and how do single-subscription streams differ here?

Q6 (advanced). What does asyncMap do that map can't, and what does a terminal operation like toList() return?

Wrapping up

A Stream is a Future that keeps on giving:

• It emits data / error / done events — zero-to-many values over time.
• Consume with await for (loop-like, continues after done) or .listen() (callbacks + a StreamSubscription you can pause/resume/cancel).
• Single-subscription (default): one listener, full ordered sequence — files, HTTP bodies. Broadcast: many listeners, no replay of past events — UI events, buses.
• Build lazy pipelines with map/where/take/asyncMap; terminal reducers collapse a stream into a Future.
• Always cancel subscriptions in dispose() to avoid leaks.

So far we've only consumed streams someone handed us. In Part 5 we learn to produce them imperatively with the StreamController — the workhorse behind custom event buses, BLoCs, and bridging callback APIs into the stream world.