Dart Streams — Continuous Data Flows and Reactive Programming

In this tutorial, you will learn about Dart Streams. We cover key concepts, practical examples, and best practices to help you master this topic.

Dart streams are asynchronous sequences of data events that arrive over time, enabling reactive programming patterns for handling user input, network data, file I/O, and real-time updates.

What You Will Learn

• Creating streams with StreamController and async generators
• Listening to streams with listen and await for
• Transforming streams with map, where, and transform
• Single-subscription vs broadcast streams
• Stream subscriptions and cancellation
• Handling stream errors and completion
• Using StreamBuilder in Flutter

Why It Matters

Many real-world data sources are continuous rather than one-shot: user taps, GPS location updates, Websocket messages, file system changes, and sensor data. Streams model these naturally. Flutter relies heavily on streams for form validation, animation controllers, and state management with BLoC. Understanding streams unlocks reactive programming patterns that lead to cleaner, more maintainable code.

Real-World Use

The DodaTech app uses streams for real-time notifications, search-as-you-type functionality, and chat messaging. The WebSocket client exposes a stream of messages. The search bar transmits keystroke events through a stream that is debounced and transformed before triggering API calls.

Learning Path

flowchart LR
  A[Async Programming] --> B[Dart Streams\nYou are here]
  B --> C[Dart Isolates]
  style B fill:#f90,color:#fff

What is a Stream?

A stream is an asynchronous sequence of data. It delivers events (data, errors, or done) over time:

import 'dart:async';

void main() {
  var stream = Stream.fromIterable([1, 2, 3, 4, 5]);

  var subscription = stream.listen(
    (data) => print('Data: $data'),
    onError: (error) => print('Error: $error'),
    onDone: () => print('Stream closed'),
  );

  // Subscription can be paused, resumed, or cancelled
  // subscription.pause();
  // subscription.resume();
  // subscription.cancel();
}

Output:

Data: 1
Data: 2
Data: 3
Data: 4
Data: 5
Stream closed

The listen method returns a StreamSubscription that controls the flow. The stream delivers data events, then a done event when complete.

Creating Streams

Several ways to create streams:

import 'dart:async';

void main() async {
  // From iterable
  var stream1 = Stream.fromIterable(['a', 'b', 'c']);
  await for (var value in stream1) {
    print('From iterable: $value');
  }

  // From future
  var stream2 = Stream.fromFuture(Future.value('From future'));
  await for (var value in stream2) {
    print(value);
  }

  // Periodic stream
  var counter = 0;
  var periodic = Stream.periodic(Duration(milliseconds: 500), (i) => i);
  var sub = periodic.take(3).listen((i) {
    print('Periodic: $i');
  });
  await Future.delayed(Duration(seconds: 2));
}

Output:

From iterable: a
From iterable: b
From iterable: c
From future: From future
Periodic: 0
Periodic: 1
Periodic: 2

Stream.fromIterable converts a collection to a stream. Stream.fromFuture wraps a single Future as a stream. Stream.periodic emits values at regular intervals.

StreamController

StreamController gives you manual control over stream events:

import 'dart:async';

Stream<int> createNumberStream() {
  var controller = StreamController<int>();

  // Simulate data arriving over time
  Future(() async {
    for (var i = 1; i <= 5; i++) {
      await Future.delayed(Duration(milliseconds: 500));
      controller.add(i);
    }
    controller.close();
  });

  return controller.stream;
}

void main() async {
  var stream = createNumberStream();

  await for (var num in stream) {
    print('Received: $num');
  }

  print('Stream finished');
}

Output:

Received: 1
(500ms pause)
Received: 2
...
Received: 5
Stream finished

StreamController provides add() to emit data, addError() to emit errors, and close() to signal completion. The stream property is what listeners subscribe to.

Single-Subscription vs Broadcast

Streams come in two flavors:

import 'dart:async';

void main() {
  // Single-subscription stream: only one listener allowed
  var singleController = StreamController<int>();
  var singleStream = singleController.stream;

  singleStream.listen((v) => print('Listener 1: $v'));
  // singleStream.listen((v) => print('Listener 2')); // THROWS error

  // Broadcast stream: multiple listeners allowed
  var broadcastController = StreamController<int>.broadcast();
  var broadcastStream = broadcastController.stream;

  broadcastStream.listen((v) => print('Broadcast 1: $v'));
  broadcastStream.listen((v) => print('Broadcast 2: $v'));

  // Add data to both
  singleController.add(1);
  broadcastController.add(10);
  broadcastController.add(20);

  singleController.close();
  broadcastController.close();
}

Output:

Listener 1: 1
Broadcast 1: 10
Broadcast 2: 10
Broadcast 1: 20
Broadcast 2: 20

Single-subscription streams (default) are for one-to-one communication. Broadcast streams are for one-to-many. Use broadcast when multiple widgets or services need to react to the same events.

Transforming Streams

Streams support functional transformations similar to collections:

import 'dart:async';

void main() async {
  var numbers = Stream.fromIterable([1, 2, 3, 4, 5, 6]);

  // Map: transform each element
  var doubled = numbers.map((n) => n * 2);
  await for (var n in doubled) {
    print('Doubled: $n');
  }

  // Where: filter elements
  var stream2 = Stream.fromIterable([1, 2, 3, 4, 5, 6]);
  var evens = stream2.where((n) => n.isEven);
  await for (var n in evens) {
    print('Even: $n');
  }

  // Distinct: remove consecutive duplicates
  var stream3 = Stream.fromIterable([1, 1, 2, 2, 3, 1]);
  var distinct = stream3.distinct();
  await for (var n in distinct) {
    print('Distinct: $n');
  }
}

Output:

Doubled: 2
Doubled: 4
Doubled: 6
...
Even: 2
Even: 4
Even: 6
Distinct: 1
Distinct: 2
Distinct: 3
Distinct: 1

Transformation methods return new streams. They are lazy: no data flows until someone listens. Common transforms: map, where, take, skip, distinct, expand.

Stream Transformers

For complex transformations, use StreamTransformer:

import 'dart:async';

// Custom transformer that debounces events
StreamTransformer<T, T> debounce<T>(Duration duration) {
  return StreamTransformer<T, T>.fromHandlers(
    handleData: (data, sink) {
      Timer? timer;
      timer?.cancel();
      timer = Timer(duration, () {
        sink.add(data);
      });
    },
  );
}

void main() async {
  var controller = StreamController<String>.broadcast();

  controller.stream
    .transform(debounce(Duration(milliseconds: 300)))
    .listen((value) => print('Debounced: $value'));

  // Simulating rapid input
  controller.add('h');
  await Future.delayed(Duration(milliseconds: 50));
  controller.add('he');
  await Future.delayed(Duration(milliseconds: 50));
  controller.add('hel');
  await Future.delayed(Duration(milliseconds: 50));
  controller.add('hell');
  await Future.delayed(Duration(milliseconds: 50));
  controller.add('hello');

  // Wait for debounce delay
  await Future.delayed(Duration(milliseconds: 400));
  controller.close();
}

Output:

Debounced: hello

StreamTransformer receives each event and forwards it based on custom logic. The debounce transformer delays forwarding until 300ms after the last event, which is useful for search-as-you-type inputs.

Await For Loop

The await for loop consumes a stream elegantly:

import 'dart:async';

Stream<String> getMessages() async* {
  var messages = ['Hello', 'World', 'From', 'Stream'];
  for (var msg in messages) {
    await Future.delayed(Duration(milliseconds: 300));
    yield msg;
  }
}

Future<void> main() async {
  print('Messages:');

  await for (var message in getMessages()) {
    print('  $message');
  }

  print('All messages received');
}

Output:

Messages:
  Hello
  World
  From
  Stream
All messages received

await for exits when the stream closes. If the stream never closes, the loop runs indefinitely. Use .first, .last, .single, or .toList() for one-shot stream consumption.

Error Handling in Streams

Streams can emit errors that must be handled:

import 'dart:async';

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

  controller.stream.listen(
    (data) => print('Data: $data'),
    onError: (error) => print('Error: $error'),
    onDone: () => print('Done'),
    cancelOnError: false, // Continue after error
  );

  controller.add(1);
  controller.addError('Something went wrong');
  controller.add(2);
  controller.add(3);
  controller.close();
}

Output:

Data: 1
Error: Something went wrong
Data: 2
Data: 3
Done

With cancelOnError: false, the stream continues after an error. With true (default for single-subscription streams), the subscription cancels on the first error.

Stream in Flutter

Flutter's StreamBuilder widget rebuilds when the stream emits:

import 'package:flutter/material.dart';

class CounterScreen extends StatelessWidget {
  final Stream<int> counterStream;

  CounterScreen({required this.counterStream});

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: Text('Stream Counter')),
      body: Center(
        child: StreamBuilder<int>(
          stream: counterStream,
          builder: (context, snapshot) {
            if (snapshot.hasError) {
              return Text('Error: ${snapshot.error}');
            }
            if (!snapshot.hasData) {
              return CircularProgressIndicator();
            }
            return Column(
              mainAxisAlignment: MainAxisAlignment.center,
              children: [
                Text('Count:', style: TextStyle(fontSize: 24)),
                Text(
                  '${snapshot.data}',
                  style: TextStyle(fontSize: 48, fontWeight: FontWeight.bold),
                ),
              ],
            );
          },
        ),
      ),
    );
  }
}

StreamBuilder subscribes to the stream and calls <a href="/design-patterns/builder/">Builder</a> on each event. It handles connection state (waiting, active, done) and errors automatically.

Combining Streams

Combine multiple streams into one:

import 'dart:async';

void main() async {
  var stream1 = Stream.periodic(
    Duration(milliseconds: 300), (i) => 'Stream1: $i',
  ).take(3);

  var stream2 = Stream.periodic(
    Duration(milliseconds: 500), (i) => 'Stream2: $i',
  ).take(3);

  // Merge: interleave events
  var merged = StreamGroup.merge([stream1, stream2]);
  await for (var value in merged) {
    print('Merged: $value');
  }

  print('---');

  // Zip: combine pairs
  var s1 = Stream.fromIterable(['A', 'B', 'C']);
  var s2 = Stream.fromIterable([1, 2, 3]);
  var zipped = StreamZip([s1, s2]);
  await for (var pair in zipped) {
    print('Zipped: $pair');
  }
}

Output:

Merged: Stream1: 0
Merged: Stream2: 0
Merged: Stream1: 1
Merged: Stream1: 2
Merged: Stream2: 1
Merged: Stream2: 2
---
Zipped: [A, 1]
Zipped: [B, 2]
Zipped: [C, 3]

StreamGroup.merge interleaves events from multiple streams. StreamZip pairs events by index.

Common Mistakes

1. Not canceling stream subscriptions: Uncanceled subscriptions cause memory leaks. Always cancel subscriptions in dispose() or use StreamBuilder which manages subscriptions automatically.

2. Using single-subscription stream for multiple listeners: Only one listener can subscribe to a single-subscription stream. Use .asBroadcastStream() or create a StreamController.broadcast() for multiple listeners.

3. Blocking the event loop inside stream handlers: Stream handlers run in the event loop. Long synchronous operations inside a handler block all other processing. Offload heavy work to isolates.

4. Forgetting to handle stream errors: Unhandled stream errors propagate to the zone level and may crash the application. Always provide an onError handler or use .handleError().

5. Creating streams with no listeners: A stream that is created but never listened to still executes. Lazy streams (from async*) only execute when listened to. Eager streams (from StreamController) execute regardless.

Practice Questions

1. What is the difference between single-subscription and broadcast streams?
2. How does StreamTransformer differ from map?
3. When would you use await for vs .listen()?
4. How does StreamBuilder manage stream subscriptions in Flutter?
5. Challenge: Implement a searchSuggestions function that takes a stream of user keystrokes, debounces by 300ms, filters out short queries (less than 3 characters), and maps each query to a simulated API call result.

Mini Project

Build a real-time search system:

• Simulate a stream of user keystrokes (characters arriving every 100ms)
• Debounce the stream by 300ms
• Filter out queries shorter than 3 characters
• Transform each query to upper case
• Collect results into a list (last 5 unique queries)
• Print the final search suggestions list

FAQ

What is Next

Now that you understand streams, learn about isolates for parallel execution. Proceed to Dart Isolates for multi-threading and handling CPU-intensive work. Then explore Dart Extensions for adding functionality to existing types.