Stream Controllers in Dart

What is a StreamController?

A StreamController<T> gives you fine-grained control over a stream — you decide when to add events, errors, and when to close the stream. The controller exposes a sink for adding events and a stream for listening.

Basic Usage

import 'dart:async';

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

  // Add events to the sink
  controller.sink.add(1);
  controller.sink.add(2);
  controller.sink.add(3);
  controller.sink.close();  // close when done

  // Listen to the stream
  await for (var value in controller.stream) {
    print(value);  // 1, 2, 3
  }

  // Always close the controller to free resources
  await controller.close();
}

Broadcast Controllers

// Regular: only one listener
// Broadcast: multiple listeners
var broadcast = StreamController<String>.broadcast();

broadcast.stream.listen((msg) => print('Listener 1: $msg'));
broadcast.stream.listen((msg) => print('Listener 2: $${msg.toUpperCase()}'));

broadcast.sink.add('hello');
broadcast.sink.add('world');
await broadcast.close();
// Listener 1: hello
// Listener 2: HELLO
// Listener 1: world
// Listener 2: WORLD

Event Bus Pattern

class EventBus {
  final _controller = StreamController<Map<String, dynamic>>.broadcast();

  Stream<Map<String, dynamic>> get stream => _controller.stream;

  // Filter stream to specific event type
  Stream<Map<String, dynamic>> on(String eventType) =>
      stream.where((e) => e['type'] == eventType);

  void emit(String type, Map<String, dynamic> data) {
    _controller.add({'type': type, ...data});
  }

  Future<void> close() => _controller.close();
}

void main() async {
  var bus = EventBus();

  bus.on('login').listen((e) => print('User logged in: $${e['user']}'));
  bus.on('logout').listen((e) => print('User logged out: $${e['user']}'));
  bus.stream.listen((e) => print('All events: $${e['type']}'));

  bus.emit('login', {'user': 'alice'});
  bus.emit('login', {'user': 'bob'});
  bus.emit('logout', {'user': 'alice'});

  await Future.delayed(Duration(milliseconds: 50));
  await bus.close();
}

All events: login
User logged in: alice
All events: login
User logged in: bob
All events: logout
User logged out: alice

🧠 Quiz

Summary

StreamController<T> gives manual control over stream events. Add events via controller.sink.add(), add errors via controller.sink.addError(), and close with controller.sink.close(). Use StreamController.broadcast() for multiple listeners. Always close controllers to avoid memory leaks.