#Lecture 95 — Stream Transformations

Doubling, filtering, or totalling stream events used to mean an await for loop full of ifs and counters. With map, where, fold and friends, you describe each step in one line — exactly as you did with lists.

Contents

Prerequisites · Connection · Objectives · Lesson · Code Examples · Common Mistakes · Best Practices · Mini Challenge · Practice Questions · Answers · Takeaways

Every output and error message in this lecture is real. Each one was produced by running the code shown on Dart 3.12.2.

#Prerequisites

Students should already understand, from Lectures 043–094:

• map, where, take, skip, expand, reduce, and fold on lists (Lectures 043–044)
• Future, async, await, and try/catch around await (Lectures 088–090)
• streams, Stream.fromIterable, Stream.periodic, and await for (Lecture 092)
• StreamController and addError (Lecture 093)

#Connection With Previous Lectures

Here is how we have handled stream events so far — Sara's marks arrive one by one, and we want the passing marks, plus 5 bonus points each:

void main() async {
  var marks = Stream.fromIterable([72, 35, 90, 48]);
  await for (var mark in marks) {
    if (mark >= 50) {
      print(mark + 5);
    }
  }
}

77
95

It works, but filtering, changing, and printing are tangled in one loop. In Lecture 043 we solved the same problem for lists with where and map. Streams have those methods too.

#Learning Objectives

By the end of this lecture you will be able to:

• Transform stream events with map and filter them with where
• Limit a stream with take, skip, takeWhile, and skipWhile
• Drop repeated events with distinct and split events with expand
• Run a Future for each event with asyncMap
• Recover from stream errors with handleError
• Chain several transformations into one pipeline
• Collect a result with toList, first, length, reduce, fold, any, and contains
• Write a small custom step with StreamTransformer.fromHandlers

#1. Streams Have List Methods

Analogy — a stream is a conveyor belt. A transformation is a station along the belt: it changes, removes, or adds items as they pass, and the belt carries the result onward.

A transformation returns a new stream; the original is not changed. Put where and map on the example above:

void main() async {
  var marks = Stream.fromIterable([72, 35, 90, 48]);
  var boosted = marks.where((m) => m >= 50).map((m) => m + 5);
  await for (var mark in boosted) {
    print(mark);
  }
}

77
95

The loop now only prints. The same names you learned in Lectures 043 and 044 work here.

#2. A Transformation Is Not a Result

A list's map gives a lazy Iterable. A stream's map gives another stream — events still arrive over time, so you cannot print them directly:

void main() {
  var marks = Stream.fromIterable([72, 35, 90]);
  print(marks.map((m) => m + 5));
}

Instance of '_MapStream<int, int>'

You must listen — with await for or one of the result methods in Steps 9 and 10.

#3. take and skip

take(n) keeps the first n events and then ends the stream. That matters for endless streams like Stream.periodic:

void main() async {
  var ticks = Stream.periodic(Duration(milliseconds: 200), (i) => i + 1);
  await for (var tick in ticks.skip(1).take(3)) {
    print('Tick $tick');
  }
  print('Done');
}

Tick 2
Tick 3
Tick 4
Done

Without take, Done would never print. skip(1) dropped the first tick.

#4. takeWhile and skipWhile

These work by condition. takeWhile stops at the first event that fails the test; skipWhile drops events until the first one that fails:

void main() async {
  var readings = [20, 24, 31, 22, 35];
  print(await Stream.fromIterable(readings).takeWhile((t) => t < 30).toList());
  print(await Stream.fromIterable(readings).skipWhile((t) => t < 30).toList());
}

[20, 24]
[31, 22, 35]

22 is below 30, but takeWhile had already stopped at 31. toList() is explained in Step 9.

#5. distinct — Drop Repeats

distinct() drops an event that equals the one right before it. A thermometer that sends the same value again and again only reports changes:

void main() async {
  var temps = Stream.fromIterable([25, 25, 26, 26, 26, 25]);
  await for (var t in temps.distinct()) {
    print('Now $t°C');
  }
}

Now 25°C
Now 26°C
Now 25°C

The last 25 survives: it differs from the 26 before it. distinct is not a set.

#6. expand — Many Events from One

expand turns each event into a list of events, like expand on lists:

void main() async {
  var orders = Stream.fromIterable(['pen,book', 'bag']);
  await for (var item in orders.expand((o) => o.split(','))) {
    print(item);
  }
}

pen
book
bag

#7. asyncMap — A Future per Event

Sometimes changing an event needs to wait — for example, looking up a price on a server. Give map an async function and you get futures, not prices:

Future<double> priceOf(String item) async {
  await Future.delayed(Duration(milliseconds: 300));
  return {'pen': 50.0, 'book': 400.0}[item]!;
}

void main() async {
  var items = Stream.fromIterable(['pen', 'book']);
  await for (var p in items.map(priceOf)) {
    print(p);
  }
}

Instance of 'Future<double>'
Instance of 'Future<double>'

asyncMap waits for each future and passes on its value (same priceOf):

void main() async {
  var items = Stream.fromIterable(['pen', 'book']);
  await for (var p in items.asyncMap(priceOf)) {
    print(p);
  }
}

50.0
400.0

asyncMap handles one event at a time, so the order is always kept.

#8. Chaining a Pipeline

Like list methods, stream transformations chain. Write one step per line:

void main() async {
  var words = Stream.fromIterable(['hi', 'hi', 'dart', 'is', 'fun', 'ok']);
  var result = words
      .distinct()
      .where((w) => w.length > 2)
      .map((w) => w.toUpperCase())
      .take(2);
  await for (var w in result) {
    print(w);
  }
}

DART
FUN

Each event travels through every station before the next event starts.

#9. Collecting: toList, first, length

Some methods do not return a stream. They listen to the whole stream and give one answer — so they return a Future, and you await it:

void main() async {
  var marks = [72, 35, 90, 48];
  print(await Stream.fromIterable(marks).toList());
  print(await Stream.fromIterable(marks).first);
  print(await Stream.fromIterable(marks).length);
}

[72, 35, 90, 48]
72
4

Each line builds its own stream. A stream from Stream.fromIterable happens to allow more than one listener, but most streams — a controller's, for example — allow only one (Mistake 3).

#10. Combining: reduce, fold, any, contains

The aggregation methods from Lecture 044 work the same way, ending in a Future:

void main() async {
  var marks = [72, 35, 90, 48];
  var stream = Stream.fromIterable(marks);
  var highest = await stream.reduce((a, b) => a > b ? a : b);
  var total = await Stream.fromIterable(marks).fold(0, (sum, m) => sum + m);
  print(highest);
  print(total);
  print(await Stream.fromIterable(marks).any((m) => m < 40));
  print(await Stream.fromIterable(marks).contains(90));
}

90
245
true
true

Store the fold result in a variable before printing: inside print(...), Dart cannot work out the type of sum and reports The operator '+' isn't defined for the type 'Object?'. As with lists, reduce fails on an empty stream; fold returns its starting value.

#11. handleError — Recover and Continue

If an event causes an error, the stream delivers it, and await for throws it. handleError catches errors inside the pipeline, so the remaining events still arrive:

Future<int> loadMark(String name) async {
  if (name == 'Zayan') throw Exception('No record for $name');
  return {'Ali': 80, 'Sara': 92}[name]!;
}

void main() async {
  var names = Stream.fromIterable(['Ali', 'Zayan', 'Sara']);
  var marks = names.asyncMap(loadMark).handleError((e) => print('Skipped: $e'));
  await for (var m in marks) {
    print(m);
  }
}

80
Skipped: Exception: No record for Zayan
92

Without handleError, the error would end the loop at Zayan; you would need try/catch around it, and Sara's mark would never print.

#12. A Custom Step with StreamTransformer.fromHandlers

When no built-in method fits, build your own station. handleData receives each event and a sink — anything you add to the sink goes out on the new stream:

import 'dart:async';

void main() async {
  var grades = StreamTransformer<int, String>.fromHandlers(
    handleData: (mark, sink) {
      if (mark >= 50) sink.add('Pass ($mark)');
    },
  );
  var marks = Stream.fromIterable([72, 35, 90]);
  await for (var g in marks.transform(grades)) {
    print(g);
  }
}

Pass (72)
Pass (90)

transform attaches it. StreamTransformer lives in dart:async, hence the import. For everyday work, the built-in methods are clearer; reach for this only when they cannot do it.

#13. A Real-World Example

A shop's checkout scanner sends barcodes. Some scans repeat, and one barcode is unknown. We look up each price, skip errors, and total the bill:

Future<double> lookUp(String code) async {
  await Future.delayed(Duration(milliseconds: 100));
  var prices = {'A1': 250.0, 'B2': 100.0, 'C3': 75.0};
  if (!prices.containsKey(code)) throw Exception('Unknown code $code');
  return prices[code]!;
}

void main() async {
  var scans = Stream.fromIterable(['A1', 'A1', 'B2', 'X9', 'C3']);
  var total = await scans
      .distinct()
      .asyncMap(lookUp)
      .handleError((e) => print('Warning: $e'))
      .fold(0.0, (sum, price) => sum + price);
  print('Total: $total');
}

Warning: Exception: Unknown code X9
Total: 425.0

distinct removed the double scan, asyncMap waited for each lookup, handleError kept the bad code from stopping the checkout, and fold produced one Future with the total.

#Code Examples

#Example 1 — Squares with map

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

1
4
9
16

#Example 2 — Even Numbers with where

void main() async {
  var numbers = Stream.fromIterable([1, 2, 3, 4, 5, 6]);
  print(await numbers.where((n) => n.isEven).toList());
}

[2, 4, 6]

#Example 3 — Three Ticks from an Endless Stream

void main() async {
  var ticks = Stream.periodic(Duration(milliseconds: 100), (i) => i);
  print(await ticks.take(3).toList());
}

[0, 1, 2]

#Example 4 — Only Changes with distinct

void main() async {
  var lights = Stream.fromIterable(['red', 'red', 'green', 'green', 'red']);
  print(await lights.distinct().toList());
}

[red, green, red]

#Example 5 — Letters with expand

void main() async {
  var names = Stream.fromIterable(['Ali', 'Sara']);
  print(await names.expand((n) => n.split('')).length);
}

7

#Example 6 — A Delayed Lookup with asyncMap

Future<int> ageOf(String name) =>
    Future.delayed(Duration(milliseconds: 200), () => name.length * 5);

void main() async {
  var names = Stream.fromIterable(['Ali', 'Zayan']);
  print(await names.asyncMap(ageOf).toList());
}

[15, 25]

#Example 7 — Student Marks: Class Average

void main() async {
  var marks = [80, 65, 92, 47];
  var total = await Stream.fromIterable(marks).fold(0, (sum, m) => sum + m);
  var passed = await Stream.fromIterable(marks).where((m) => m >= 50).length;
  print('Average: ${total / marks.length}');
  print('Passed: $passed');
}

Average: 71.0
Passed: 3

#Example 8 — Shopping Total with Tax

void main() async {
  var prices = Stream.fromIterable([500.0, 1200.0, 300.0]);
  var total = await prices.map((p) => p * 1.1).fold(0.0, (sum, p) => sum + p);
  print('Total with tax: ${total.toStringAsFixed(2)}');
}

Total with tax: 2200.00

#Example 9 — Bank Balance: First Large Withdrawal

void main() async {
  var amounts = [-200.0, 1500.0, -3000.0, -50.0];
  var large = await Stream.fromIterable(amounts).where((p) => p < -1000).first;
  var hasDeposit = await Stream.fromIterable(amounts).any((p) => p > 0);
  print('First large withdrawal: $large');
  print('Any deposit: $hasDeposit');
}

First large withdrawal: -3000.0
Any deposit: true

first and any stop listening as soon as they have their answer.

#Example 10 — Temperature: Highest Fahrenheit Reading

void main() async {
  var celsius = Stream.periodic(
    Duration(milliseconds: 100),
    (i) => [21.0, 25.0, 23.0][i],
  );
  var highest = await celsius
      .take(3)
      .map((c) => c * 9 / 5 + 32)
      .reduce((a, b) => a > b ? a : b);
  print('Highest: $highest°F');
}

Highest: 77.0°F

#Common Beginner Mistakes

1. Forgetting await on a result method

void main() {
  var list = Stream.fromIterable([1, 2, 3]).toList();
  print(list);
}

Instance of 'Future<List<int>>'

toList, first, fold and the rest return a Future. Await it.

2. Using map with an async function

items.map(priceOf) gives a stream of futures, as Step 7 showed. Use asyncMap.

3. Listening to the same stream twice

import 'dart:async';

void main() async {
  var controller = StreamController<int>();
  controller.add(72);
  controller.add(35);
  controller.close();
  var marks = controller.stream;
  print(await marks.length);
  print(await marks.first);
}

2
Unhandled exception:
Bad state: Stream has already been listened to.

length listened to the whole stream; first cannot listen again. Collect once with toList() and work on the list.

4. Expecting distinct to remove every duplicate

void main() async {
  print(await Stream.fromIterable([1, 2, 1]).distinct().toList());
}

[1, 2, 1]

It only compares each event with the one just before it.

5. reduce or first on an empty stream

void main() async {
  var marks = Stream.fromIterable([35, 40]).where((m) => m >= 50);
  var best = await marks.reduce((a, b) => a > b ? a : b);
  print(best);
}

Unhandled exception:
Bad state: No element

No mark passes, so there is nothing to reduce. Use fold when the stream might be empty.

#Best Practices

• Think "list method, but awaited": the names from Lectures 043–044 carry over.
• Put transformations in a chain, one step per line, and the result method last.
• Always take from an endless stream before collecting it.
• Use asyncMap, not map, when each step returns a Future.
• Add handleError when one bad event should not stop the rest.
• Prefer fold over reduce when a stream could be empty.
• Write a StreamTransformer only when no built-in method fits.

#Mini Challenge

Ali's step counter sends readings: 1200, 1200, 3400, 0, 5600, 5600, 8000. Ignore repeated readings and the 0 (a sensor glitch), then print the readings in thousands (1.2k) and the count of real readings.

Try it yourself before opening the solution.

void main() async {
  var readings = [1200, 1200, 3400, 0, 5600, 5600, 8000];
  var steps = Stream.fromIterable(readings);
  var clean = await steps
      .distinct()
      .where((r) => r > 0)
      .map((r) => '${r / 1000}k')
      .toList();
  print(clean);
  print('Readings: ${clean.length}');
}

[1.2k, 3.4k, 5.6k, 8.0k]
Readings: 4

#Practice Questions

Solve each one before looking at the answers.

Beginner

1. From Stream.fromIterable([3, 6, 9]), print each number times 10 with map and await for.
2. Keep only the names longer than 3 letters from ['Ali', 'Sara', 'Zayan'] and print them as a list.
3. Print the first 2 events of Stream.periodic(Duration(milliseconds: 100), (i) => i * 5) as a list.

Application

4. What does this print?

   dartCopy

   void main() async {
     var s = Stream.fromIterable([4, 4, 5, 4, 6, 6]);
     print(await s.distinct().toList());
   }

5. Print the total of [250.0, 100.0, 75.0] using a stream and fold.

6. Using skipWhile, drop the readings below 30 from [18, 25, 33, 29, 40] until the first hot one, and print the rest as a list.

7. Use contains to check whether 'Sara' is in a stream of ['Ali', 'Sara'], and any to check whether any name starts with 'Z'.

Challenge

8. square(n) waits 100 ms and returns n * n. Use asyncMap to print the squares of [2, 3, 4] as a list.
9. A stream of ['10', 'x', '30'] must be turned into numbers with an async function that throws on 'x'. Use handleError to print Bad value and print the sum 40.
10. Write a StreamTransformer<String, String> that sends each word twice, and apply it to ['hi', 'bye'].

#Practice Answers

#Question 1 — Solution

void main() async {
  await for (var n in Stream.fromIterable([3, 6, 9]).map((n) => n * 10)) {
    print(n);
  }
}

30
60
90

#Question 2 — Solution

void main() async {
  var names = Stream.fromIterable(['Ali', 'Sara', 'Zayan']);
  print(await names.where((n) => n.length > 3).toList());
}

[Sara, Zayan]

#Question 3 — Solution

void main() async {
  var s = Stream.periodic(Duration(milliseconds: 100), (i) => i * 5);
  print(await s.take(2).toList());
}

[0, 5]

#Question 4 — Solution

[4, 5, 4, 6]

Only neighbouring repeats are dropped, so the second 4 (after 5) stays.

#Question 5 — Solution

void main() async {
  var prices = Stream.fromIterable([250.0, 100.0, 75.0]);
  var total = await prices.fold(0.0, (sum, p) => sum + p);
  print(total);
}

425.0

#Question 6 — Solution

void main() async {
  var readings = Stream.fromIterable([18, 25, 33, 29, 40]);
  print(await readings.skipWhile((t) => t < 30).toList());
}

[33, 29, 40]

29 stays: once skipWhile reaches 33, it stops checking.

#Question 7 — Solution

void main() async {
  var names = ['Ali', 'Sara'];
  print(await Stream.fromIterable(names).contains('Sara'));
  print(await Stream.fromIterable(names).any((n) => n.startsWith('Z')));
}

true
false

#Question 8 — Solution

Future<int> square(int n) =>
    Future.delayed(Duration(milliseconds: 100), () => n * n);

void main() async {
  print(await Stream.fromIterable([2, 3, 4]).asyncMap(square).toList());
}

[4, 9, 16]

#Question 9 — Solution

Future<int> toNumber(String text) async {
  var n = int.tryParse(text);
  if (n == null) throw FormatException(text);
  return n;
}

void main() async {
  var texts = Stream.fromIterable(['10', 'x', '30']);
  var sum = await texts
      .asyncMap(toNumber)
      .handleError((e) => print('Bad value'))
      .fold(0, (total, n) => total + n);
  print(sum);
}

Bad value
40

#Question 10 — Solution

import 'dart:async';

void main() async {
  var twice = StreamTransformer<String, String>.fromHandlers(
    handleData: (word, sink) {
      sink.add(word);
      sink.add(word);
    },
  );
  print(await Stream.fromIterable(['hi', 'bye']).transform(twice).toList());
}

[hi, hi, bye, bye]

Adding to the sink twice sends two events — a transformer can output more, fewer, or the same number of events.

#Quick Revision

1. What does a stream's map return — a list, a future, or a stream?
2. Why is take important on Stream.periodic?
3. Which events does distinct drop?
4. When do you use asyncMap instead of map?
5. What do toList, first, and fold return, and what must you write before them?
6. What does handleError change about an error in the middle of a stream?
7. Why does the second await marks.first fail after await marks.length?
8. What is the sink in StreamTransformer.fromHandlers for?

#Key Takeaways

• Streams have the list methods from Lectures 043–044; a transformation returns a new stream.
• map, where, expand, take, skip, takeWhile, skipWhile work like their list versions.
• take(n) ends a stream — essential for endless streams.
• distinct drops an event equal to the previous one, not every duplicate.
• asyncMap waits for a Future per event, in order; map with async gives futures.
• handleError catches errors inside the pipeline so later events still arrive.
• toList, first, length, reduce, fold, any, contains return a Future — await it.
• Most streams allow one listener; collect once and reuse the result.
• StreamTransformer.fromHandlers builds a custom step that adds events to a sink.

#What's Next?

Lecture 096 — Async Generators. Every stream so far came ready-made from Stream.fromIterable, Stream.periodic, or a controller. Next we will write functions that produce a stream themselves, with async* and yield — sending each event exactly when it is ready.