Single & Broadcast Streams

## `listen` is the primitive Every way of consuming a Dart stream, including `await for`, `first`, `toList` and `forEach`, is built on `Stream.listen`: ```dart StreamSubscription<T> listen( void onData(T event)?, { Function? onError, void onDone()?, bool? cancelOnError, }); ``` ## The handlers - **`onData`** is called once per data event. It may be `null`, in which case data events are ignored. - **`onError`** must be a `void Function(Object error)` or a `void Function(Object error, StackTrace stackTrace)`; the stream checks which shape you passed and supplies the stack trace only to the second. If `onError` is **omitted**, error events are **unhandled**: they are passed to the current zone's error handler, which by default treats them like uncaught top-level errors. - **`onDone`** is called once, when the stream sends its done event. - **`cancelOnError`** decides what an error does to the subscription. The default is **`false`**: the error is delivered and the subscription carries on receiving later events. With `true`, the subscription is cancelled automatically when the first error is delivered. Whether a stream keeps sending after an error depends on its producer: an `async*` function that throws emits the error and closes, while other sources may continue. ## The returned `StreamSubscription` `listen` returns a `StreamSubscription<T>`, the live connection between listener and stream: | Member | What it does | |---|---| | `cancel()` | stops delivery and returns a `Future<void>` that completes when the stream has cleaned up | | `pause([Future<void>? resumeSignal])` | stops delivery; events that still arrive are buffered | | `resume()` | undoes one `pause` | | `isPaused` | `true` while pauses outnumber resumes | | `onData`, `onError`, `onDone` | replace the corresponding handler | | `asFuture([value])` | turns done into a completing future and error into a failing one | ## Pausing in detail 1. While paused, **no handlers run**. Events the subscription receives are **buffered** until it resumes. 2. For a **non-broadcast** stream, the source is usually informed and can stop producing; an `async*` body, for instance, suspends at its next `yield`. 3. For a **broadcast** stream, the pause affects only that subscription, which buffers everything meanwhile. The SDK suggests cancelling and listening again instead, if intermediate events do not matter. 4. Pauses are **counted**: two `pause` calls need two `resume` calls. Passing a `resumeSignal` future resumes automatically when it completes. 5. `resume` on a subscription that is not paused is safe and does nothing. ## Cancelling, and why it matters A subscription to a **long-lived** stream, such as a temperature sensor's broadcast stream, holds a reference to your callbacks and everything they capture. If the listener's owner goes away without cancelling, the callbacks keep running and nothing is garbage-collected. The rule is simple: whoever calls `listen` keeps the subscription and cancels it when done. ```dart class TemperatureLogger { TemperatureLogger(Stream<double> temps) { _sub = temps.listen( (t) => _log.add(t), onError: (Object e, StackTrace st) => _errors.add(e), onDone: () => _log.add(double.nan), ); } final _log = <double>[]; final _errors = <Object>[]; late final StreamSubscription<double> _sub; Future<void> close() => _sub.cancel(); } ``` `cancel()` returning a future matters when cleanup is itself asynchronous: the SDK's example is a stream that must close a file before the listener can delete it, so the listener awaits `cancel()` first. ## Cancelling from inside a handler Sometimes the decision to stop is made by the data itself, such as "log readings until the temperature passes 30 degrees". The handler then needs the subscription it belongs to, which does not exist until `listen` returns. A `late` local solves it: ```dart late final StreamSubscription<double> sub; sub = temps.listen((t) { print(t); if (t > 30) sub.cancel(); }); ``` Alternatively, create the subscription with a `null` `onData` and install the handler afterwards with `sub.onData(...)`, which replaces the data handler on an existing subscription. ## Common mistakes - Omitting `onError` and then wondering why errors appear as uncaught. - Assuming the subscription ends on the first error when `cancelOnError` was left at `false`. - Pausing a broadcast subscription for a long time and building an unbounded buffer. - Never cancelling, which is the classic leak in long-lived screens and services.