Streams & Reactive Data

Streams — Asynchronous Sequences

A Stream<T> is like a Future<T>, but instead of delivering one value once, it delivers zero or more values over time. Think of it as an asynchronous Iterable.

Creating Streams

dart
// 1. async* generator function — yields values over time
Stream<int> countdown(int from) async* {
  for (int i = from; i >= 0; i--) {
    await Future.delayed(const Duration(seconds: 1));
    yield i; // 'yield' sends a value downstream
  }
}

// 2. StreamController — manual control
final controller = StreamController<String>();
controller.sink.add('Hello');
controller.sink.add('World');
controller.close();

// 3. Periodic stream
final ticks = Stream.periodic(
  const Duration(seconds: 1),
  (tick) => tick, // transform each tick
);

// 4. From a list
final stream = Stream.fromIterable([1, 2, 3, 4, 5]);
Stream.fromFuture(fetchUser()); // wraps a Future as a single-value Stream

Consuming Streams

dart
// Method 1: await for (most readable)
Future<void> printCountdown() async {
  await for (final value in countdown(5)) {
    print(value); // 5, 4, 3, 2, 1, 0
  }
  print('Liftoff!');
}

// Method 2: listen (lower-level, more control)
final subscription = countdown(10).listen(
  (value) => print('Tick: $value'),
  onError: (e) => print('Error: $e'),
  onDone: () => print('Stream complete'),
  cancelOnError: false,
);

// Later, unsubscribe
await subscription.cancel();

Stream Transformations

Streams have a rich set of transformation operators, similar to Iterable:

dart
final userStream = fetchUserStream();

userStream
  .where((user) => user.isActive)            // filter
  .map((user) => user.name.toUpperCase())    // transform
  .distinct()                                // skip duplicates
  .take(10)                                  // take first 10
  .timeout(const Duration(seconds: 5))       // timeout each event
  .listen(print);

Single vs Broadcast Streams

dart
// Single-subscription stream — only one listener allowed (default)
final singleStream = fetchDataStream();

// Broadcast stream — multiple listeners allowed
final broadcastStream = singleStream.asBroadcastStream();
broadcastStream.listen(listenerA);
broadcastStream.listen(listenerB); // Both receive all events

StreamController for State Management

dart
class UserBloc {
  final _controller = StreamController<UserState>.broadcast();
  Stream<UserState> get stream => _controller.stream;

  Future<void> loadUser(String id) async {
    _controller.add(UserState.loading());
    try {
      final user = await fetchUser(id);
      _controller.add(UserState.success(user));
    } catch (e) {
      _controller.add(UserState.error(e.toString()));
    }
  }

  void dispose() => _controller.close(); // Always close!
}

StreamBuilder in Flutter

dart
StreamBuilder<UserState>(
  stream: userBloc.stream,
  builder: (context, snapshot) {
    final state = snapshot.data;
    return switch (state) {
      null || UserState.loading => const CircularProgressIndicator(),
      UserState.success(:final user) => UserCard(user: user),
      UserState.error(:final message) => ErrorWidget(message),
    };
  },
)

Summary

Streams are the backbone of reactive programming in Dart. WebSocket data, real-time database updates, sensor events, user input — all of these are naturally modeled as streams. In Flutter, StreamBuilder and packages like rxdart build on top of this foundation to power entire state management architectures.