Dart / Contracts: patterns, generic APIs, and asynchronous results / Lesson 12

Own stream subscriptions and delivery semantics

Handle values, errors, completion, cancellation, and single-subscription or broadcast behavior without assuming replay or bounded buffering.

About 5 min readContent reviewed 2026-10-28

On this page

Lesson goals

Distinguish a Stream source from an active subscription and its resource lifetime.
Choose event handling that preserves required data when listeners are slow or absent.

A stream carries more than successful values

A Stream can deliver data events, error events, and a done event. An error does not universally imply that the source can never emit another value; the subscription and consumption policy determine how it is handled. An async* generator produces a stream, while await for consumes it using sequential asynchronous control flow. A listen call instead returns a StreamSubscription that the owner must manage. Identify whether the source owns a file, socket, timer, or no external resource at all. The Stream type alone cannot specify durability, replay, ordering across independent sources, or how much buffering the implementation may use.

Consume a finite stream with visible completion

The generator below yields two fixed counts, and await for transforms and prints them in order before the completion marker. There is no timer, external I/O, controller, or broadcast sharing. The simple form makes the sequential consumption contract clear. If the body awaits a slow operation, processing of later values follows the stream and subscription behavior rather than magically creating parallel workers. Introduce concurrency only with an explicit ordering, error, and resource policy. This program is complete source, but its output is reasoned because no Dart runtime was available.

dart An owned finite async generator

import 'dart:async';
Stream<int> counts() async* {
  yield 1;
  yield 2;
}
Future<void> main() async {
  await for (final count in counts()) {
    print(count * 10);
  }
  print('done');
}

Expected output

10
20
done

Expected output is 10, 20, done. The loop waits for stream completion before printing done. No stream runtime, scheduler, or timing experiment was executed.

Single-subscription and broadcast solve different needs

A single-subscription stream is intended for one listener, including later attempts after an earlier listener finishes. A broadcast stream permits multiple listeners, but does not automatically replay old events to a listener that arrives late. No-listener behavior and resource activation depend on the source. Converting to broadcast is therefore not a universal fix for a repeated-listen error: it may change loss and lifetime semantics. Decide whether every consumer requires every event, only future events, or current state. A stream of required batch mutations needs a stronger delivery design than a transient notification channel.

Handle a stream error at its consumption boundary

The example uses Stream.error to make one synthetic error and catches it around await for. It does not infer that all production stream failures are FormatException or that every error should end an application. With listen, provide the intended error and done handling and retain the subscription if lifecycle cancellation is needed. Avoid ignoring errors while presenting a complete-success state to the user. Record enough safe context to distinguish malformed data, a disconnected source, and a consumer defect; those cases can need different recovery actions and owners.

dart A synthetic error consumed through await for

import 'dart:async';
Future<void> main() async {
  final source = Stream<int>.error(const FormatException('synthetic sample'));
  try {
    await for (final value in source) {
      print(value);
    }
  } on FormatException {
    print('stream rejected');
  }
  print('consumer finished');
}

Expected output

stream rejected
consumer finished

Expected output is stream rejected then consumer finished. No data event is printed. The program was not compiled or run and does not connect to a real stream source.

Pausing is not a universal source-throttling guarantee

StreamSubscription pause can defer delivery and may cause buffering if the source cannot pause its actual production. An unbounded slow-consumer backlog can consume memory even when the API appears asynchronous. Cancellation can require asynchronous cleanup, so await the returned cancellation Future where ownership requires completion. Breaking an await-for loop cancels its subscription according to the language’s consumption behavior; it does not roll back external effects already performed. Choose a finite capacity or drop/reject policy that matches the event meaning, and test the actual source in an isolated project before claiming reliable backpressure or resource closure.

Common errors and fixes

A second listener fails on a single-subscription stream.

Why it happens: The source contract allows only one subscription.
How to fix it: Create a fresh source or choose deliberate sharing with the required semantics.
Prevent it next time: Document subscription kind before exposing a stream publicly.

A late broadcast listener misses important earlier events.

Why it happens: Multiple-listener support was mistaken for durable replay.
How to fix it: Use retained state or an event-delivery mechanism that explicitly preserves the required history.
Prevent it next time: Specify late-listener and no-listener behavior in the data contract.

A paused subscriber accumulates an unbounded backlog.

Why it happens: Delivery pause was assumed to stop all upstream production.
How to fix it: Bound buffering and choose source-specific flow control or an explicit loss policy.
Prevent it next time: Verify slow-consumer behavior and memory limits with the actual source contract.

Check your understanding

1. Should a repeated-listen error always be fixed by calling asBroadcastStream?

Show worked solution

No. First determine whether consumers need fresh independent execution, shared future events, replayed history, or current state. Broadcasting changes subscription semantics and may introduce missed events or a longer source lifetime. Choose the design that preserves the product requirement and assign resource cancellation ownership explicitly.

2. Why is printing onDone insufficient proof that a source resource was released safely?