Dart Stream Transformations

A stream transformation builds a new stream from an existing stream. Each transformation filters, changes, or trims the values as they pass through.

Filtering and changing values

The where call keeps the even numbers 2 and 4. The map call then multiplies each kept number by ten.

import 'dart:async';

Future<void> main() async {
  var stream = Stream.fromIterable([1, 2, 3, 4]);
  await for (var n in stream.where((n) => n.isEven).map((n) => n * 10)) {
    print(n);
  }
}

Removing consecutive duplicates

The list repeats 1 and 2 in a row but returns to 1 later. Only the repeats that sit next to an equal value are dropped.

import 'dart:async';

Future<void> main() async {
  var stream = Stream.fromIterable([1, 1, 2, 2, 3, 1]);
  await for (var n in stream.distinct()) {
    print(n);
  }
}

Trimming with skip and take

The skip call drops the first value 10. The take call then lets two values through and ends the stream.

import 'dart:async';

Future<void> main() async {
  var stream = Stream.fromIterable([10, 20, 30, 40]);
  await for (var n in stream.skip(1).take(2)) {
    print(n);
  }
}

Mapping with asyncMap

The tag function returns a future for each number. The values still arrive in the order of the source stream, as 3 then 1 then 2.

import 'dart:async';

Future<String> tag(int n) async => 'tag$n';

Future<void> main() async {
  var stream = Stream.fromIterable([3, 1, 2]).asyncMap(tag);
  await for (var t in stream) {
    print(t);
  }
}

Example

import 'dart:async';

Future<void> main() async {
  var stream = Stream.fromIterable([8, 3, 12, 5, 20]);
  await for (var n in stream.where((n) => n >= 8).map((n) => n + 2)) {
    print(n);
  }
}

Key points

A transformation takes a stream and returns a new stream.
The where method keeps only the values that pass its test.
The map method changes each kept value into a new value.
The distinct method drops a value only when it matches the value just before it.
The skip method drops a set number of opening values.
The take method ends the stream after a set number of values.
The asyncMap method maps each value with a function that returns a future, and it emits the results in source order.

In this example, where keeps 8, 12, and 20 because each one is at least 8. The map call adds 2 to each kept value. The loop prints 10, 14, and 22 in that order.

Note

Transformations chain in order, so each method in a chain receives the values the earlier methods passed on.

Try it Yourself

Edit the code, then press Run.

Running is simulated · DartPad

import 'dart:async';

Future<void> main() async {
  var stream = Stream.fromIterable([8, 3, 12, 5, 20]);
  await for (var n in stream.where((n) => n >= 8).map((n) => n + 2)) {
    print(n);
  }
}

Exercise

A stream holds 2, 2, 3, 2 in that order. What does distinct print, value by value?

2, 3, 2 2, 3 2, 2, 3, 2 3, 2