StreamController in Dart

This is Part 5 of the Async & Concurrency in Dart series. In Part 4 you learned to consume streams. Now you'll learn to produce them — to be the source pushing events down the conveyor belt. The tool for that is the StreamController, and it's the engine behind event buses, the BLoC pattern, and any "turn my callbacks/events into a stream" task.

If a Stream is the conveyor belt and a StreamSubscription is the listener at the end of it, a StreamController is the operator standing at the start, placing items on the belt whenever they're ready.

The shape of a controller

A StreamController<T> exposes two halves:

• controller.stream — the read end. You hand this out to consumers; they .listen() to it.
• controller.sink (and the shortcuts .add, .addError, .close) — the write end. You push events in.

import 'dart:async';

void main() {
  final controller = StreamController<int>();

  // Consumer side: listen to the read end.
  controller.stream.listen(
    (value) => print('received: $value'),
    onDone: () => print('stream closed'),
  );

  // Producer side: push events into the write end.
  controller.add(1);
  controller.add(2);
  controller.add(3);
  controller.close(); // send the "done" event — important!
}

Output:

received: 1
received: 2
received: 3
stream closed

Three rules you'll never regret memorizing:

1. add(value) emits a data event; addError(e) emits an error event.
2. close() emits the done event and frees the controller. Always close when finished — otherwise await for/StreamBuilder never know it's over, and you leak.
3. By default the stream is single-subscription — exactly one .listen().

A Sink is just "the write end"

controller.sink is a StreamSink<T> — an abstraction for "somewhere you put things." It has add, addError, and close. Why does it matter? Because you can pass the sink to producers without giving them the power to listen or close the whole controller. It's the classic read/write split:

void wireUpSensor(StreamSink<double> out) {
  // This code can only PUSH readings; it can't listen or tear down.
  sensor.onReading = (value) => out.add(value);
}

final controller = StreamController<double>();
wireUpSensor(controller.sink);          // give producers the write end
controller.stream.listen(print);         // keep the read end yourself

This is exactly how you bridge a callback-based API into a stream — the streaming analogue of the Completer trick from Part 2 (which bridged callbacks into a single Future).

Lifecycle callbacks: react to listeners

A StreamController lets you hook into when someone starts/stops listening. This is how you avoid doing work nobody wants:

late StreamController<int> controller;
Timer? timer;
int counter = 0;

controller = StreamController<int>(
  onListen: () {
    // Called when the FIRST listener subscribes — start producing now.
    timer = Timer.periodic(const Duration(seconds: 1), (_) {
      controller.add(counter++);
    });
  },
  onPause:  () => print('listener paused — consider stopping work'),
  onResume: () => print('listener resumed'),
  onCancel: () {
    // Called when the LAST listener leaves — stop & clean up.
    timer?.cancel();
    print('no more listeners; timer stopped');
  },
);

| Callback | Fires when | Use it to | | --- | --- | --- | | onListen | first listener subscribes | start producing (don't produce before anyone's listening) | | onPause | listener pauses | stop generating events (avoid buffering) | | onResume | listener resumes | resume producing | | onCancel | last listener cancels | clean up resources (timers, sockets) |

🔑 The golden rule: don't produce events until you have a listener, and stop when they pause or leave. Starting a timer in onListen and killing it in onCancel is the canonical pattern. Producing eagerly wastes work and overflows buffers.

Backpressure: don't outrun your consumer

What if you add() faster than the listener consumes? For a single-subscription controller, events are buffered until they're delivered. If the listener is paused (or slow) and you keep adding, the buffer grows unbounded — a memory leak waiting to happen.

That's why onPause/onResume exist: they let a well-behaved producer stop generating when the consumer can't keep up.

// ❌ Ignores pause — buffer can grow without limit.
Timer.periodic(d, (_) => controller.add(expensiveReading()));

// ✅ Respects backpressure — pause/resume gate production.
controller = StreamController<int>(
  onListen: startProducing,
  onPause: stopProducing,   // consumer is overwhelmed → stop
  onResume: startProducing, // consumer caught up → continue
  onCancel: stopProducing,
);

Avoid StreamController(sync: true). A synchronous controller delivers events immediately inside your add() call, which can re-enter listener code at surprising times and break stream guarantees. Stick with the default (asynchronous) controller unless you have a specific, well-understood reason.

Broadcast controllers: many listeners

By default a controller's stream is single-subscription. When you need multiple listeners (an app-wide event bus, a shared notifier), create a broadcast controller:

final bus = StreamController<String>.broadcast();

bus.stream.listen((e) => print('analytics: $e'));
bus.stream.listen((e) => print('logger: $e'));

bus.add('user_signed_in'); // both listeners receive it

Remember the Part 4 rules carry over: a broadcast controller doesn't buffer for future listeners — events added while nobody's listening are gone, and a late listener only sees subsequent events. Broadcast onListen/onCancel semantics also shift (they fire on going from zero↔non-zero listeners).

A real-world example: a tiny event bus

Putting it together — a reusable, type-safe event bus, the kind you'll see in countless Flutter apps:

import 'dart:async';

class EventBus<T> {
  final _controller = StreamController<T>.broadcast();

  Stream<T> get events => _controller.stream; // read end (public)
  void emit(T event) => _controller.add(event); // write end (public)

  Future<void> dispose() => _controller.close(); // always close!
}

void main() {
  final bus = EventBus<String>();

  final sub = bus.events.listen((e) => print('heard: $e'));

  bus.emit('hello');
  bus.emit('world');

  sub.cancel();
  bus.dispose();
}

Note the encapsulation: consumers get events (a Stream, read-only) and emit (push only) — they can't close the controller or peek at its internals. This read/write separation is exactly what BLoC and similar patterns formalize.

When to reach for async* instead

A StreamController is the right tool when events come from outside — callbacks, timers, sockets, UI. But when you're generating a sequence with your own logic (a loop, a computation), an async* generator is far cleaner and handles listen/pause/cancel for you automatically. That's the subject of Part 6. Quick rule of thumb:

• Pushing external events into a stream → StreamController.
• Pulling a sequence out of your own code → async* generator.

Practice Challenges

Challenge 1 — Basic controller. Create a StreamController<String>, listen to it, push "a", "b", then close.

import 'dart:async';

void main() {
  final c = StreamController<String>();
  c.stream.listen(print, onDone: () => print('done'));
  c.add('a');
  c.add('b');
  c.close();
}

Challenge 2 — Bridge a callback. A sensor calls onReading(double). Expose its readings as a Stream<double>.

import 'dart:async';

Stream<double> sensorReadings(void Function(void Function(double)) register) {
  final controller = StreamController<double>();
  register((value) => controller.add(value)); // push each callback into the sink
  return controller.stream;
}

Challenge 3 — Produce only when listened. Emit an incrementing int every 500ms, but only while there's a listener, and stop cleanly when they leave.

import 'dart:async';

Stream<int> counter() {
  late StreamController<int> c;
  Timer? t;
  var n = 0;
  c = StreamController<int>(
    onListen: () => t = Timer.periodic(
        const Duration(milliseconds: 500), (_) => c.add(n++)),
    onCancel: () => t?.cancel(),
  );
  return c.stream;
}

Challenge 4 — Broadcast bus. Make a broadcast controller with two listeners and show both receive the same event.

import 'dart:async';

void main() {
  final c = StreamController<int>.broadcast();
  c.stream.listen((v) => print('one: $v'));
  c.stream.listen((v) => print('two: $v'));
  c.add(99); // one: 99 / two: 99
  c.close();
}

Challenge 5 — Spot the leak. What's wrong here, and how do you fix it?

final c = StreamController<int>();
c.stream.listen(print);
c.add(1);
c.add(2);
// ... function returns

Questions to test yourself

Q1 (basic). What are the two "ends" of a StreamController and what is each used for?

Q2 (basic). Which method emits the "done" event, and why must you call it?

Q3 (intermediate). What's the purpose of the onListen and onCancel callbacks?

Q4 (intermediate). Why pass controller.sink to a producer rather than the whole controller?

Q5 (advanced). What is backpressure, and how does a StreamController help you handle it?

Q6 (advanced). When should you choose a StreamController over an async* generator, and vice versa?

Wrapping up

StreamController is how you become a stream source:

• It splits into a stream (read end, for listeners) and a sink (write end: add/addError/close).
• Always close() to emit "done" and release resources — forgetting is a classic leak.
• onListen/onPause/onResume/onCancel tie event production to real demand and let you respect backpressure.
• .broadcast() controllers support multiple listeners (with no replay for latecomers).
• Pass the sink to producers to enforce read/write encapsulation — the basis of event buses and BLoC.

When the sequence comes from your own code rather than external events, there's a cleaner tool. In Part 6 we meet generators — sync*, async*, yield and yield* — which produce iterables and streams lazily, with none of the manual bookkeeping.