Creating streams in Dart

Written by Lasse Nielsen April 2013 (updated October 2018)

The dart:async library contains two typesthat are important for many Dart APIs:Stream andFuture.Where a Future represents the result of a single computation,a stream is a sequence of results.You listen on a stream to get notified of the results(both data and errors)and of the stream shutting down.You can also pause while listening or stop listening to the streambefore it is complete.

But this article is not about using streams.It’s about creating your own streams.You can create streams in a few ways:

• Transforming existing streams.
• Creating a stream from scratch by using an async* function.
• Creating a stream by using a StreamController.

This article shows the code for each approachand gives tips to help you implement your stream correctly.

For help on using streams, seeAsynchronous Programming: Streams.

Transforming an existing stream

The common case for creating streams is that you already have a stream,and you want to create a new stream based on the original stream’s events.For example you might have a stream of bytes thatyou want to convert to a stream of strings by UTF-8 decoding the input.The most general approach is to create a new stream thatwaits for events on the original stream and thenoutputs new events. Example:

/// Splits a stream of consecutive strings into lines.
///
/// The input string is provided in smaller chunks through
/// the `source` stream.
Stream<String> lines(Stream<String> source) async* {
  // Stores any partial line from the previous chunk.
  var partial = '';
  // Wait until a new chunk is available, then process it.
  await for (var chunk in source) {
    var lines = chunk.split('\n');
    lines[0] = partial + lines[0]; // Prepend partial line.
    partial = lines.removeLast(); // Remove new partial line.
    for (var line in lines) {
      yield line; // Add lines to output stream.
    }
  }
  // Add final partial line to output stream, if any.
  if (partial.isNotEmpty) yield partial;
}

For many common transformations,you can use Stream-supplied transforming methodssuch as map(), where(), expand(), and take().

For example, assume you have a stream, counterStream,that emits an increasing counter every second.Here’s how it might be implemented:

var counterStream =
    Stream<int>.periodic(Duration(seconds: 1), (x) => x).take(15);

To quickly see the events, you can use code like this:

counterStream.forEach(print); // Print an integer every second, 15 times.

To transform the stream events, you can invoke a transforming methodsuch as map() on the stream before listening to it.The method returns a new stream.

// Double the integer in each event.
var doubleCounterStream = counterStream.map((int x) => x * 2);
doubleCounterStream.forEach(print);

Instead of map(), you could use any other transforming method,such as the following:

.where((int x) => x.isEven) // Retain only even integer events.
.expand((var x) => [x, x]) // Duplicate each event.
.take(5) // Stop after the first five events.

Often, a transforming method is all you need.However, if you need even more control over the transformation,you can specify aStreamTransformerwith Stream’s transform() method.The platform libraries provide stream transformers for many common tasks.For example, the following code uses the utf8.decoder and LineSplittertransformers provided by the dart:convert library.

Stream<List<int>> content = File('someFile.txt').openRead();
List<String> lines =
    await content.transform(utf8.decoder).transform(LineSplitter()).toList();

Creating a stream from scratch

One way to create a new stream is withan asynchronous generator (async) function.The stream is created when the function is called,and the function’s body starts running when the stream is listened to.When the function returns, the stream closes.Until the function returns, it can emit events on the stream byusing yield or yield statements.

Here’s a primitive example that emits numbers at regular intervals:

Stream<int> timedCounter(Duration interval, [int maxCount]) async* {
  int i = 0;
  while (true) {
    await Future.delayed(interval);
    yield i++;
    if (i == maxCount) break;
  }
}

This function returns a Stream.When that stream is listened to, the body starts running.It repeatedly delays for the requested interval and then yields the next number.If the count parameter is omitted, there is no stop condition on the loop,so the stream outputs increasingly larger numbers forever -or until the listener cancels its subscription.

When the listener cancels(by invoking cancel() on the StreamSubscriptionobject returned by the listen() method),then the next time the body reaches a yield statement,the yield instead acts as a return statement.Any enclosing finally block is executed,and the function exits.If the function attempts to yield a value before exiting,that fails and acts as a return.

When the function finally exits, the future returned bythe cancel() method completes.If the function exits with an error, the future completes with that error;otherwise, it completes with null.

Another, more useful example is a function that convertsa sequence of futures to a stream:

Stream<T> streamFromFutures<T>(Iterable<Future<T>> futures) async* {
  for (var future in futures) {
    var result = await future;
    yield result;
  }
}

This function asks the futures iterable for a new future,waits for that future, emits the resulting value, and then loops.If a future completes with an error, then the stream completes with that error.

It’s rare to have an async function building a stream from nothing.It needs to get its data from somewhere,and most often that somewhere is another stream.In some cases, like the sequence of futures above,the data comes from other asynchronous event sources.In many cases, however, an async function is too simplistic toeasily handle multiple data sources.That’s where the StreamController class comes in.

Using a StreamController

If the events of your stream comes from different parts of your program,and not just from a stream or futures that can traversed by an async function,then use aStreamControllerto create and populate the stream.

A StreamController gives you a new streamand a way to add events to the stream at any point, and from anywhere.The stream has all the logic necessary to handle listeners and pausing.You return the stream and keep the controller to yourself.

The following example(from stream_controller_bad.dart)shows a basic, though flawed, usage of StreamControllerto implement the timedCounter() function from the previous examples.This code creates a stream to return,and then feeds data into it based on timer events,which are neither futures nor stream events.

// NOTE: This implementation is FLAWED!
// It starts before it has subscribers, and it doesn't implement pause.
Stream<int> timedCounter(Duration interval, [int maxCount]) {
  var controller = StreamController<int>();
  int counter = 0;
  void tick(Timer timer) {
    counter++;
    controller.add(counter); // Ask stream to send counter values as event.
    if (maxCount != null && counter >= maxCount) {
      timer.cancel();
      controller.close(); // Ask stream to shut down and tell listeners.
    }
  }
 
  Timer.periodic(interval, tick); // BAD: Starts before it has subscribers.
  return controller.stream;
}

As before, you can use the stream returned by timedCounter() like this:

var counterStream = timedCounter(const Duration(seconds: 1), 15);
counterStream.listen(print); // Print an integer every second, 15 times.

This implementation of timedCounter() hasa couple of problems:

• It starts producing events before it has subscribers.
• It keeps producing events even if the subscriber requests a pause.

As the next sections show,you can fix both of these problems by specifyingcallbacks such as onListen and onPausewhen creating the StreamController.

Waiting for a subscription

As a rule, streams should wait for subscribers before starting their work.An async* function does this automatically,but when using a StreamController,you are in full control and can add events even when you shouldn’t.When a stream has no subscriber,its StreamController buffers events,which can lead to a memory leakif the stream never gets a subscriber.

Try changing the code that uses the stream to the following:

void listenAfterDelay() async {
  var counterStream = timedCounter(const Duration(seconds: 1), 15);
  await Future.delayed(const Duration(seconds: 5));
 
  // After 5 seconds, add a listener.
  await for (int n in counterStream) {
    print(n); // Print an integer every second, 15 times.
  }
}

When this code runs,nothing is printed for the first 5 seconds,although the stream is doing work.Then the listener is added,and the first 5 or so events are printed all at once,since they were buffered by the StreamController.

To be notified of subscriptions, specify anonListen argument when you create the StreamController.The onListen callback is calledwhen the stream gets its first subscriber.If you specify an onCancel callback,it’s called when the controller loses its last subscriber.In the preceding example,Timer.periodic()should move to an onListen handler,as shown in the next section.

Honoring the pause state

Avoid producing events when the listener has requested a pause.An async* function automatically pauses at a yield statementwhile the stream subscription is paused.A StreamController, on the other hand, buffers events during the pause.If the code providing the events doesn’t respect the pause,the size of the buffer can grow indefinitely.Also, if the listener stops listening soon after pausing,then the work spent creating the buffer is wasted.

To see what happens without pause support,try changing the code that uses the stream to the following:

void listenWithPause() {
  var counterStream = timedCounter(const Duration(seconds: 1), 15);
  StreamSubscription<int> subscription;
 
  subscription = counterStream.listen((int counter) {
    print(counter); // Print an integer every second.
    if (counter == 5) {
      // After 5 ticks, pause for five seconds, then resume.
      subscription.pause(Future.delayed(const Duration(seconds: 5)));
    }
  });
}

When the five seconds of pause are up,the events fired during that time are all received at once.That happens because the stream’s source doesn’t honor pausesand keeps adding events to the stream.So the stream buffers the events,and it then empties its buffer when the stream becomes unpaused.

The following version of timedCounter()(from stream_controller.dart)implements pause by using theonListen, onPause, onResume, and onCancel callbackson the StreamController.

Stream<int> timedCounter(Duration interval, [int maxCount]) {
  StreamController<int> controller;
  Timer timer;
  int counter = 0;
 
  void tick(_) {
    counter++;
    controller.add(counter); // Ask stream to send counter values as event.
    if (counter == maxCount) {
      timer.cancel();
      controller.close(); // Ask stream to shut down and tell listeners.
    }
  }
 
  void startTimer() {
    timer = Timer.periodic(interval, tick);
  }
 
  void stopTimer() {
    if (timer != null) {
      timer.cancel();
      timer = null;
    }
  }
 
  controller = StreamController<int>(
      onListen: startTimer,
      onPause: stopTimer,
      onResume: startTimer,
      onCancel: stopTimer);
 
  return controller.stream;
}

Run this code with the listenWithPause() function above.You’ll see that it stops counting while paused,and it resumes nicely afterwards.

You must use all of the listeners—onListen,onCancel, onPause, and onResume—to benotified of changes in pause state.The reason is that if thesubscription and pause states both change at the same time,only the onListen or onCancel callback is called.

Final hints

When creating a stream without using an async* function,keep these tips in mind:

• Be careful when using a synchronous controller—for example,one created using StreamController(sync: true).When you send an event on an unpaused synchronous controller(for example, using the add(), addError(), or close() methods defined byEventSink),the event is sent immediately to all listeners on the stream.Stream listeners must never be called untilthe code that added the listener has fully returned,and using a synchronous controller at the wrong time canbreak this promise and cause good code to fail.Avoid using synchronous controllers.

• If you use StreamController,the onListen callback is called beforethe listen call returns the StreamSubscription.Don’t let the onListen callback dependon the subscription already existing.For example, in the following code,an onListen event fires(and handler is called)before the subscription variablehas a valid value.

subscription = stream.listen(handler);

• The onListen, onPause, onResume, and onCancelcallbacks defined by StreamController arecalled by the stream when the stream’s listener state changes,but never during the firing of an eventor during the call of another state change handler.In those cases, the state change callback is delayed untilthe previous callback is complete.

• Don’t try to implement the Stream interface yourself.It’s easy to get the interaction between events, callbacks,and adding and removing listeners subtly wrong.Always use an existing stream, possibly from a StreamController,to implement the listen call of a new stream.

• Although it’s possible to create classes that extend Stream withmore functionality by extending the Stream class andimplementing the listen method and the extra functionality on top,that is generally not recommended becauseit introduces a new type that users have to consider.Instead you can often make a class that has a Stream (and more) —instead of one that is a Stream (and more).