Understanding Streams in Dart: A Complete Guide
Streams are one of the foundations of asynchronous programming in Dart. In its simplest form, a stream is a sequence of asynchronous events — a single value or a whole collection, arriving over time rather than all at once. Streams are how Dart and Flutter apps model user interactions, file I/O, and API responses that don't resolve in a single step.
Key concepts
Stream: the source of asynchronous data — a sequence of events that arrive over time.
StreamController: the object that manages a stream and lets you add events, errors, and a "done" signal to it.
StreamSubscription: what listen() returns — the handle representing an active listener, which is also what you cancel when you're done.
Streams come in two flavors: single-subscription streams allow exactly one listener over their lifetime (the default, used for one-shot things like an API response), and broadcast streams allow any number of simultaneous listeners (used for shared, ongoing data like a connectivity status or a chat message feed).
Creating a stream
Using StreamController — the general-purpose way to build a custom stream:
void main() {
final controller = StreamController<int>();
final stream = controller.stream;
final subscription = stream.listen((data) {
print('Data received: $data');
});
controller.sink.add(1);
controller.sink.add(2);
controller.close(); // signals "done" — no more events will arrive
subscription.cancel(); // releases the listener; see "cleaning up" below
}
Using Stream.fromIterable — for turning an existing collection into a stream:
final stream = Stream.fromIterable([1, 2, 3, 4, 5]);
stream.listen((event) => print('Event: $event'));
Using Stream.periodic — for events on a timer:
final stream = Stream.periodic(const Duration(seconds: 1), (count) => count);
stream.take(5).listen((event) => print('Periodic event: $event'));
Note take(5) here — without it, Stream.periodic never stops on its own, which matters later when combining streams.
Listening to streams
listen() returns a StreamSubscription, and that return value isn't just for show — it's what you cancel later:
final stream = Stream.fromIterable([1, 2, 3]);
final subscription = stream.listen(
(data) => print('Data: $data'),
onDone: () => print('Stream closed'),
);
Transforming streams
Mapping — transform each event:
stream.map((event) => event * 2).listen((data) => print('Mapped: $data'));
Filtering — pass through only matching events:
stream.where((event) => event % 2 == 0).listen((data) => print('Even: $data'));
Reducing — collapse the whole stream into a single value once it completes:
stream.reduce((acc, curr) => acc + curr).then((sum) => print('Sum: $sum'));
reduce only resolves once the stream is done — it won't work on a stream that never completes, like an un-take'd Stream.periodic.
Handling errors
stream.listen(
(data) => print('Data: $data'),
onError: (error) => print('Error: $error'),
onDone: () => print('Stream completed'),
);
An unhandled error on a stream doesn't crash your app the way an unhandled exception in synchronous code might, but it does silently stop delivering further events to that listener unless cancelOnError: false is passed to listen() — worth knowing explicitly, since the default behavior surprises people the first time a single bad event ends an otherwise-healthy stream.
Combining multiple streams — done correctly
Merging two streams — combining their events into one stream, interleaved as they actually arrive, not the events from one stream followed by the events from the other. The straightforward-looking approach using Stream.fromFutures and .toList() doesn't do this: .toList() waits for a stream to fully complete before it produces anything, so "merging" two streams that way just plays one stream's full output followed by the other's, and it silently never resolves at all if either stream is unbounded (like a Stream.periodic without .take()).
The correct tool is StreamGroup from Dart's own async package:
import 'package:async/async.dart';
Stream<int> stream1 = Stream.fromIterable([1, 2, 3]);
Stream<int> stream2 = Stream.periodic(const Duration(milliseconds: 500), (i) => i + 10).take(3);
final merged = StreamGroup.merge([stream1, stream2]);
merged.listen((event) => print('Merged: $event'));
Events from both sources now arrive on merged genuinely interleaved by real arrival time, and it works correctly even when one of the sources is a long-lived, never-completing stream — which is the entire point of merging in the first place.
Zipping — pairing up the nth event from each stream — isn't part of Dart's standard library. rxdart's Rx.zip2 is the standard tool:
import 'package:rxdart/rxdart.dart';
Rx.zip2(stream1, stream2, (int a, int b) => a + b)
.listen((sum) => print('Zipped sum: $sum'));
Async generators
The async* keyword is Dart's built-in way to write a stream as a function, using yield instead of manually managing a StreamController:
Stream<int> generateNumbers(int max) async* {
for (int i = 1; i <= max; i++) {
yield i;
await Future.delayed(const Duration(seconds: 1));
}
}
generateNumbers(5).listen(print);
This is usually the cleaner choice over a raw StreamController whenever the stream's logic is "loop, wait, produce a value" — you get cancellation and backpressure handling for free instead of managing them by hand.
Consuming a stream in Flutter: StreamBuilder
The single most common place a Flutter developer actually touches a stream isn't a manual listen() call — it's StreamBuilder, which rebuilds a widget automatically each time the stream emits, and manages subscription and cancellation for you as the widget is built and disposed:
class CounterDisplay extends StatelessWidget {
const CounterDisplay({super.key, required this.stream});
final Stream<int> stream;
@override
Widget build(BuildContext context) {
return StreamBuilder<int>(
stream: stream,
initialData: 0,
builder: (context, snapshot) {
if (snapshot.hasError) return Text('Error: ${snapshot.error}');
return Text('Count: ${snapshot.data}');
},
);
}
}
StreamBuilder subscribes when it's built and unsubscribes automatically when it's removed from the tree — this is why it's the preferred way to consume a stream directly in UI code, and manual listen() calls are more for streams you're combining or transforming in a service layer, not directly rendering.
Cleaning up — the part "always close controllers" needs to actually show
Every manual listen() call in this guide returns a StreamSubscription, and that subscription needs to be cancelled when you're done with it, the same way a TextEditingController needs .dispose() — a subscription left open keeps its callback (and whatever it captured) alive indefinitely:
class _MyWidgetState extends State<MyWidget> {
StreamSubscription<int>? _subscription;
@override
void initState() {
super.initState();
_subscription = someStream.listen((data) { /* ... */ });
}
@override
void dispose() {
_subscription?.cancel();
super.dispose();
}
}
And any StreamController you create yourself needs .close() called on it once nothing will add to it anymore — typically also in dispose() if it's owned by a widget's state, or whenever the service or repository that created it is done being used.
Best practices
Close every StreamController you create, and cancel every StreamSubscription you manually listen() to — these are two separate objects with two separate cleanup calls, and skipping either one leaks.
Prefer StreamBuilder for UI consumption over manual listen() calls inside widgets, since it handles the subscribe/unsubscribe lifecycle automatically and removes an entire category of leak risk.
Use broadcast streams specifically for shared, ongoing data — a connectivity status, a chat feed — and single-subscription streams for one-shot data, like a single API call's response.
Reach for async* generators over a raw StreamController whenever the underlying logic is a loop that produces values over time — it's less code and gets cancellation handling for free.
Use StreamGroup (from async) or rxdart for combining streams rather than hand-rolling it with Future-based tricks — merging and zipping have real edge cases (unbounded streams, error propagation) that these packages have already solved correctly.
References
