Async Programming in Dart

From Custom Streams to Optimization – Everything an Intermediate Dart Developer Needs to Know

Asynchronous programming is essential for building responsive applications. In Dart, the concept of asynchronous execution is closely tied with Future and Stream. While Future is used for single event async operations, Stream is used when you expect a sequence of asynchronous events. Whether you’re reading files, handling user input, or working with APIs, understanding Streams is crucial for writing efficient Dart applications.

What is a Stream?

A Stream represents a sequence of asynchronous data events. Unlike a Future, which completes once, a Stream can emit multiple values over time. This makes it perfect for handling data that arrives gradually like user inputs, sensor data, or API responses in chunks.

There are two main types of Streams:

1. Single-subscription Streams

These allow only one listener at a time. They're ideal when you have a known, linear flow of data like reading a file or making an HTTP request that delivers a sequence of values.

Example:

final stream = Stream.fromIterable([1, 2, 3]);

stream.listen((value) => print('Received: \$value'));

This stream emits the numbers 1, 2, and 3 in order. It supports only one listener. If you try to listen again, it will throw an error.

2. Broadcast Streams

These allow multiple listeners to subscribe to the same stream. Broadcast streams are useful when you want multiple parts of your app to react to the same events, such as user interactions or state changes like tracking app lifecycle changes, sensor updates, or user interactions.

Example:

final controller = StreamController<int>.broadcast();

controller.stream.listen((val) => print('Listener A: \$val'));
controller.stream.listen((val) => print('Listener B: \$val'));

controller.sink.add(1);
controller.sink.add(2);

Here, two listeners are added to a single stream. Both will receive the same events emitted by the controller.

Creating Streams in Dart

1. Using Stream.fromIterable()

• Why? Turns an existing collection into a stream of events, emitting each element sequentially.

• When? You have a fixed list of data (e.g., a list of IDs, file paths, or static configuration) and want to process items asynchronously.

final stream = Stream.fromIterable(["user1", "user2", "user3"]);

stream.listen((user) => print('Processing user: $user'));  // Processes each user one by one

This creates a stream that emits elements from a list one by one.

2. Using Stream.periodic()

• Why? Emits events at a fixed interval, ideal for polling or timers.

• When? You need repeated actions over time, such as refreshing data every few seconds or updating a stopwatch UI.

// Emits 0,1,2,... every second, for 5 ticks
final stream = Stream.periodic(Duration(seconds: 1), (count) => count).take(5);

stream.listen((tick) => print('Tick #$tick'));  // Tick #0, Tick #1, ... Tick #4

This emits a new value (based on count) every second. Useful for timers or polling.

3. Async Generator Function with async*

• Why? Lets you yield events from within an async function, combining asynchronous delay or I/O with stream emission.

• When? You need to fetch or compute data incrementally—e.g., loading paginated API results one page at a time, or reading large files chunk by chunk.

Stream<String> fetchPages(int totalPages) async* {
  for (int i = 1; i <= totalPages; i++) {
    final pageData = await fetchPageFromApi(i);  // hypothetical network call
    yield pageData;
  }
}

// In your code:
fetchPages(3).listen((page) => print('Received page: $page'));

This is an asynchronous generator. It yields a value after each delay. Ideal for step-by-step processing with delays or awaiting async tasks.

Custom Streams with StreamController

StreamController provides fine-grained control over when and how events are emitted. Use it when events originate from multiple sources or when you need broadcast capabilities.

• Why? To manually add, pause, resume, or close a stream; useful for event buses, user interactions, or combining multiple inputs.

• When? You have custom data flows, such as mixing user clicks and network events, or implementing global notification systems.

final controller = StreamController<String>();
controller.stream.listen((event) => print("Received: \$event"));

controller.sink.add("Hello");
controller.sink.add("World");
await controller.close();

This approach is helpful when you're manually handling events—such as user inputs or system data.

Managing StreamSubscription

A StreamSubscription gives you control over how and when to listen, pause, resume, or cancel stream events.

final subscription = controller.stream.listen((data) => print(data));

// Pause
subscription.pause();

// Resume
subscription.resume();

// Cancel
subscription.cancel();

Use this to manage the lifecycle of your listeners, especially in Flutter where you might need to stop listening on widget disposal.

Stream Transformations

map, where, take, skip

stream.map((e) => e * 2).where((e) => e > 5);

Transforms values by doubling them and filters those greater than 5.

asyncMap

stream.asyncMap((e) async => await fetchData(e));

Used when transforming stream data into another async value. Useful for network calls or async parsing.

expand and asyncExpand

stream.expand((e) => [e, e * 10]);
stream.asyncExpand((e) async* {
  yield e;
  yield await computeSomethingElse(e);
});

expand flattens items into multiple values; asyncExpand does this with async logic. Useful when each item needs to result in multiple outputs.

Error Handling in Streams

stream.listen(
  (data) => print(data),
  onError: (err) => print('Error: \$err'),
  onDone: () => print('Stream closed'),
);

Always handle errors to prevent unhandled exceptions, especially in production. onDone lets you clean up resources.

Stream Optimization Tips

1. Debounce Rapid Inputs

Use debounceTime to prevent handling every keystroke (useful for search input).

Before:

searchController.stream.listen((query) {
  print('Searching for: \$query');
});

After:

searchController.stream
  .debounceTime(Duration(milliseconds: 300))
  .listen((query) {
    print('Searching for: \$query');
  });

This waits 300ms after the last input before emitting, reducing API calls.

2. Eliminate Redundant Events

Use distinct() to ignore repeated consecutive values.

stream.distinct().listen((value) {
  print('Received: \$value');
});

This avoids processing the same value multiple times in a row.

3. Cancel Unused Streams

Free up memory and processing by canceling listeners when they’re no longer needed.

final subscription = stream.listen(print);
// Later in code
await subscription.cancel();

4. Always Close Controllers

Avoid memory leaks by closing controllers when done.

final controller = StreamController<int>();
controller.sink.add(1);
await controller.close();

Real-World Example: Debounced Search in a Flutter App

final searchController = StreamController<String>.broadcast();

TextField(
  onChanged: (value) => searchController.sink.add(value),
);

searchController.stream
  .debounceTime(Duration(milliseconds: 500))
  .listen((searchQuery) {
    fetchResults(searchQuery);
  });

In this example, we debounce the user input in a TextField. If the user types quickly, the stream emits only the final input after 500ms of inactivity.

Conclusion

Streams are a powerful part of Dart’s asynchronous capabilities. Understanding how to create, consume, transform, and optimize them will make your apps faster and more maintainable. This post focused strictly on concepts useful for intermediate developers—those who’ve moved beyond basics but are not yet diving into Dart isolates or advanced custom stream transformers.

Whether you're building reactive UIs or processing continuous data, mastering Streams is essential.

Stay tuned for the next post in our Dart Bootcamp, where we’ll integrate Streams with Flutter widgets using StreamBuilder and build a fully reactive UI.