🎓 A1Learner
🔍
0 of 24 lessons completed

Dart Stream Controllers

A StreamController creates a stream and feeds values into it by hand. Code adds values to the controller, and listeners receive them from its stream.

Values added before listening

Both values are added before any listener exists. The default controller holds them and delivers a then b once listening starts, and close then ends the stream.

import 'dart:async';

Future<void> main() async {
  var controller = StreamController<String>();
  controller.add('a');
  controller.add('b');
  controller.stream.listen((v) => print('heard: $v'), onDone: () => print('closed'));
  await controller.close();
}

Delivering values synchronously

Each add call delivers its value before the next line runs. The heard lines therefore sit between the prints of the adding code.

import 'dart:async';

Future<void> main() async {
  var controller = StreamController<int>(sync: true);
  controller.stream.listen((v) => print('heard $v'));
  print('before');
  controller.add(1);
  print('middle');
  controller.add(2);
  await controller.close();
  print('after');
}

A second listen is refused

The first listen call takes the only subscription. The second call throws a StateError at once, and the catch block prints its message.

import 'dart:async';

Future<void> main() async {
  var controller = StreamController<int>();
  controller.stream.listen((v) => print('got $v'));
  try {
    controller.stream.listen((v) => print('again $v'));
  } on StateError catch (e) {
    print(e.message);
  }
  await controller.close();
}

Listening and cancelling callbacks

The onListen callback runs as the subscription starts. The value arrives next, and onCancel runs when the subscription is cancelled.

import 'dart:async';

Future<void> main() async {
  var controller = StreamController<int>(
    onListen: () => print('listening now'),
    onCancel: () => print('cancelled now'),
  );
  print('before listen');
  var sub = controller.stream.listen((v) => print('v $v'));
  controller.add(5);
  await Future.delayed(Duration.zero);
  await sub.cancel();
  await controller.close();
}

Example

import 'dart:async';

Future<void> main() async {
  var controller = StreamController<String>();
  controller.add('opened');
  controller.add('saved');
  controller.stream.listen((event) => print('heard: $event'), onDone: () => print('log closed'));
  await controller.close();
}

Key points

  • A StreamController creates a stream that code feeds by hand with add.
  • A default controller buffers values added before listening starts and delivers them in order once a listener arrives.
  • A controller built with the sync option set to true delivers each value during the add call itself.
  • The stream of a controller allows one subscription only, and a second listen call throws a StateError.
  • The close method ends the stream, and listeners hear the end through the onDone callback.
  • The onListen callback runs when listening starts.
  • The onCancel callback runs when the subscription is cancelled.

In this example, the controller works as an event log that collects short text events. Both events are added before listening starts, so the buffered values arrive in order when the listener attaches. Closing the controller then prints the final line through onDone.

Note

Code closes a controller once it has added every value.

Try it Yourself

Edit the code, then press Run.

Running is simulated · DartPad

Exercise

What happens to values added to a default StreamController before any listener attaches?