Streams in Dart

What is Stream object?

Stream is to handle source of asynchronous data events.

A Stream provides a way to receive a sequence of events. It is like an asynchronous Iterable—where, instead of getting the next event when you ask for it, the stream tells you that there is an event when it is ready.

Each event is either a data event or an error event.

Data event also called an element of the stream.
Error event, which is a notification that something has failed.

When a stream has emitted all its event, a single "done" event will notify the listener that the end has been reached.

Generating a simple stream of Strings

Generating a simple stream of strings using an async* function:

List<String> productsData() {
  // Generates a list of values of length 10
  return List.generate(10, (index) => 'Product ${index}');
}

// Generating a simple stream of Strings
Stream<String> productsStream() async* {
  for (var item in productsData()) {
    yield item;
  }
}

Receiving stream events

The asynchronous for loop (commonly just called await for) iterates over the events of a stream like the for loop iterates over an Iterable. For example:

// Receiving stream events
Future<List<String>> listOfProductStream(Stream<String> stream) async {
  List<String> list = [];

  // iterates over the events of a stream
  await for (var item in stream) {
    // receives each event of a stream of string events, adds them up to list
    list.add(item);
  }
  // return (a future of) the list, which is delayed for 1 second
  return Future.delayed(Duration(seconds: 1), () => list);
}

This code simply receives each event of a stream of string events, adds them up to list, and returns (a future of) the list. When the loop body ends, the function is paused until the next event arrives or the stream is done.

The function is marked with the async keyword, which is required when using the await for loop.

Stream sample code

Example:

The following code snippet shows hwo to generate and receive stream events.

import 'dart:async';

List<String> productsData() {
  // Generates a list of values of length 10
  return List.generate(10, (index) => 'Product ${index}');
}

// Generating a simple stream of Strings
Stream<String> productsStream() async* {
  for (var item in productsData()) {
    yield item;
  }
}

// Receiving stream events
Future<List<String>> listOfProductStream(Stream<String> stream) async {
  List<String> list = [];

  // iterates over the events of a stream
  await for (var item in stream) {
    // receives each event of a stream of string events, adds them up to list
    list.add(item);
  }
  // return (a future of) the list, which is delayed for 1 second
  return Future.delayed(Duration(seconds: 1), () => list);
}

main() async {
  var stream = productsStream();

  var result = await listOfProductStream(stream);

  // print to console
  result.forEach(print);
}

/* Output:

Product 0
Product 1
Product 2
Product 3
Product 4
Product 5
Product 6
Product 7
Product 8
Product 9
*/