# Streams

Course: [Dart](https://www.devtutorials.ai/learn/dart) · Section: Asynchronous Programming
Source: https://www.devtutorials.ai/learn/dart/async/streams

> How Streams deliver many values over time: creating them, listening with listen and await for, making your own with StreamController and async*, and changing them with map, where and friends.

## What Is a Stream?

**Stream** = a source of **many** values that arrive **one after another**, over time.

- A **Future** gives **one** value later. A **Stream** gives **many** values later.
- A Stream sends three kinds of **events**:
  - **data** event: a value
  - **error** event: a problem
  - **done** event: "no more values"
- You **listen** to a Stream to react to each event.

**Simple lesson:** a Future is one parcel, and a Stream is a conveyor belt of parcels.

### A Real-World Example

Think of the school's **live cricket score board** on Sports Day.

- The score does not come all at once. It changes ball by ball.
- Each new score is a **data event**.
- A rain stop is an **error event**.
- "Match over" is the **done event**.
- You keep watching the board until the match is over. That is **listening**.

Every **Run** box on this page runs on the course's Dart runner in your browser.

This lesson builds on **Async Basics and Futures**. Make sure `async`, `await` and the event loop feel familiar.

## What You Will Learn

In this lesson, you will learn:

1. How to create Streams with `fromIterable`, `value` and `periodic`.
2. How to listen with `listen` and with `await for`.
3. How to make your own Streams with `async*` and `StreamController`.
4. How to change Streams with `map`, `where`, `take` and more.
5. Single-subscription and broadcast Streams.

## Future vs Stream

| | Future | Stream |
|---|---|---|
| How many values? | one | zero, one or many |
| Ends with | the value, or an error | a done event |
| Read with | `await` or `then` | `await for` or `listen` |
| Example | one download | a timer ticking, chat messages |

## Creating Streams

### Ready-Made Streams

| Code | Gives |
|---|---|
| `Stream.fromIterable([1, 2, 3])` | each item of the list, one by one, then done |
| `Stream.value(5)` | one value, then done |
| `Stream.periodic(Duration(seconds: 1), (i) => i)` | 0, 1, 2 ... one per second, forever |

```dart runnable title="A stream of roll calls"
void main() {
  var names = Stream.fromIterable(['Asha', 'Ravi', 'Priya']);
  names.listen((name) => print('$name: present'));
  print('Roll call started');
}
```

`Roll call started` prints **first**. The Stream's events come later, from the event loop.

### Ticking with periodic

```dart runnable title="A countdown that ticks every second"
Future<void> main() async {
  var ticks = Stream.periodic(Duration(seconds: 1), (i) => 3 - i).take(3);
  await for (var t in ticks) {
    print('$t...');
  }
  print('Go!');
}
```

- `periodic` never stops by itself.
- `take(3)` keeps only the first 3 values, then sends done.

**Simple lesson:** use `fromIterable` for known values, and `periodic` for values that come with time.

## Listening to Streams

### listen

**`listen`** = register functions for each kind of event:

```text
stream.listen(
  (value) { ... },        // data
  onError: (e) { ... },   // error
  onDone: () { ... },     // done
);
```

```dart runnable title="Data, error and done"
Stream<int> scores() async* {
  yield 4;
  yield 6;
  throw StateError('Rain stopped play');
}

void main() {
  scores().listen(
    (runs) => print('Runs: $runs'),
    onError: (e) => print('Problem: ${e.message}'),
    onDone: () => print('Match over'),
  );
}
```

### StreamSubscription and cancel

`listen` returns a **`StreamSubscription`**. Use it to **stop** listening:

```dart runnable title="Stop listening after the third tick"
import 'dart:async';

void main() {
  late StreamSubscription<int> sub;
  sub = Stream.periodic(Duration(seconds: 1), (i) => i + 1).listen((n) {
    print('Tick $n');
    if (n == 3) {
      sub.cancel();
      print('Cancelled');
    }
  });
}
```

### await for

**`await for`** = a loop over a Stream inside an `async` function. Each turn waits for the next value. The loop ends on **done**.

`await for` on a Stream that never ends (like `periodic` without `take`) never finishes. The code after the loop never runs.

**Simple lesson:** use `listen` to react to events, and `await for` to loop over values in order.

## Making Your Own Streams

### async* and yield

**Generator** = a function that produces values one by one.

- **`async*`** marks a function that returns a `Stream`.
- **`yield value`** sends one value to the Stream.
- **`yield* otherStream`** sends every value of another Stream.
- `await` works inside too.

```dart runnable title="A generator with waiting and yield*"
Stream<String> morning() async* {
  yield 'Bell rings';
  await Future.delayed(Duration(seconds: 1));
  yield 'Prayer';
}

Stream<String> schoolDay() async* {
  yield* morning();
  yield 'Classes';
  yield 'Home time';
}

Future<void> main() async {
  await for (var event in schoolDay()) {
    print(event);
  }
}
```

### StreamController

**`StreamController`** = a Stream you push values into **by hand**, from anywhere in your code.

- `controller.stream` = the Stream others listen to.
- `controller.add(value)` = send a data event.
- `controller.addError(e)` = send an error event.
- `controller.close()` = send done.

```dart runnable title="A notice board you post to"
import 'dart:async';

void main() {
  var board = StreamController<String>();
  board.stream.listen(
    (notice) => print('NOTICE: $notice'),
    onDone: () => print('Board closed for today'),
  );

  board.add('Sports Day on Friday');
  board.add('Library closed at 3 pm');
  board.close();
}
```

**Simple lesson:** write `async*` with `yield` for a Stream of steps, and use a `StreamController` to push values when things happen.

## Transforming Streams

Many list methods (from **Working with Iterables**) work on Streams too. Each gives a **new** Stream.

| Method | Does |
|---|---|
| `map(f)` | changes each value |
| `where(test)` | keeps only values that pass |
| `expand(f)` | turns each value into many |
| `take(n)` / `skip(n)` | keeps the first `n` / skips the first `n` |
| `handleError(f)` | catches error events, so the Stream continues |

Turn steps on and off, and send the events:

```dart runnable title="Only even marks, times ten"
Future<void> main() async {
  var marks = Stream.fromIterable([3, 8, 5, 10, 7, 4]);
  var result = marks.where((m) => m.isEven).map((m) => m * 10).skip(1);
  await for (var r in result) {
    print(r);
  }
}
```

### Results That Are Futures

Some Stream methods wait for the **whole** Stream and give **one** answer, as a Future:

```dart runnable title="first, length, reduce and fold"
Stream<int> runs() => Stream.fromIterable([4, 1, 6, 0, 2]);

Future<void> main() async {
  print(await runs().first);
  print(await runs().last);
  print(await runs().length);
  print(await runs().isEmpty);
  print(await runs().reduce((a, b) => a + b));
  print(await runs().fold(100, (total, r) => total + r));
  print(await runs().toList());
}
```

`runs()` makes a **fresh** Stream each time, because a normal Stream can be listened to only once (next section).

**Simple lesson:** chain `where`, `map` and friends to shape a Stream, and use `first`, `fold` or `toList` to boil it down to one answer.

## Single-Subscription and Broadcast

- **Single-subscription** Stream (the default) = **one** listener only. It keeps values until someone listens.
- **Broadcast** Stream = **many** listeners at once. Values sent with nobody listening are lost.

```dart runnable expect-error title="Listening twice to a normal stream"
void main() {
  var s = Stream.fromIterable([1, 2, 3]);
  s.listen((v) => print('A got $v'));
  s.listen((v) => print('B got $v'));
}
```

```dart runnable title="A broadcast stream for many listeners"
import 'dart:async';

void main() {
  var announcements = StreamController<String>.broadcast();
  announcements.stream.listen((m) => print('Class 7A hears: $m'));
  announcements.stream.listen((m) => print('Class 7B hears: $m'));

  announcements.add('Assembly in 5 minutes');
  announcements.close();
}
```

| Use | Kind |
|---|---|
| reading a file, one download | single-subscription |
| button taps, announcements, live scores | broadcast |

**Easy way to remember**

- **Future** = one parcel. **Stream** = a conveyor belt.
- **listen** = watch the belt. **await for** = take parcels one by one.
- **async* + yield** = put parcels on your own belt.
- **broadcast** = many people may watch the same belt.

**Simple lesson:** a normal Stream has one listener, and a broadcast Stream can have many.

## TL;DR

1. **Stream** = many values over time: data, error and done events.
2. Create: `Stream.fromIterable`, `Stream.value`, `Stream.periodic`, `async*` with `yield`, `StreamController`.
3. Read: **`listen`** (with `onError:` and `onDone:`, returns a subscription you can `cancel()`), or **`await for`**.
4. Transform: `map`, `where`, `expand`, `take`, `skip`, `handleError`. Boil down: `first`, `length`, `reduce`, `fold`.
5. **Single-subscription** = one listener. **Broadcast** = many.

## Common mistakes

| Mistake | Wrong idea | Correct idea | What goes wrong |
|---|---|---|---|
| Listening twice to a normal Stream | Any Stream allows many listeners | Use `.broadcast()` or a new Stream | `Bad state: Stream has already been listened to.` |
| `await for` on a never-ending Stream | The loop stops by itself | Use `take`, or `cancel` the subscription | Code after the loop never runs. |
| Forgetting `close()` on a controller | Streams end by themselves | `close()` sends done | Listeners wait forever for done. |
| `yield` in a normal function | Any function can yield | Mark it `async*` (or `sync*`) | A compile error. |
| No `onError:` | Streams cannot fail | Handle errors with `onError:` or `handleError` | An unhandled exception. |

## Check Yourself

## Exercises

## What to Remember

- A Stream gives many values over time, then a done event.
- Read it with `listen` (with `onError:` and `onDone:`) or with `await for`.
- Make your own with `async*` and `yield`, or push values with a `StreamController`.
- `map`, `where`, `take` and friends make new Streams, and `fold` or `toList` give one answer.
- A normal Stream allows one listener, and a broadcast Stream allows many.

Next: **Isolates**. You will run heavy work in the background without freezing your app.