skip to content

In Dart, how do Stream.asyncMap and Stream.asyncExpand differ from map, and do they process events concurrently?

level: middleimportance: should knowfreq 33%

answer

  1. FutureOr per event, or a stream per event
  2. source paused while work is pending
  3. one at a time, order kept
  4. asyncExpand concatenates inner streams
  5. concurrency lives elsewhere

basics

~20 s

asyncMap lets the per-event function return a Future and waits for it; asyncExpand maps each event to a whole stream and emits all its events. Both pause the source while that work runs, so events are processed one at a time, in order.

solid answer

~40 s

`map` converts each event synchronously. `asyncMap` accepts a `FutureOr<E>` result: when it gets a `Future`, it pauses the source subscription, waits, emits the value (or the error), then resumes. `asyncExpand` maps each event to a `Stream<E>?`, pauses the source, relays that inner stream's data and errors through `addStream`, and resumes when it ends — so inner streams are concatenated, never interleaved. Neither runs work concurrently: order is preserved and throughput is bounded by the slowest step. That is the right default for saving chat messages to a database in order; if you need parallel or latest-wins processing, that is a different operator, not these two.

code

dart · 12 lines
dart
import 'dart:async';

Future<int> save(String message) async {
  await Future<void>.delayed(const Duration(milliseconds: 50));
  return message.length;
}

Future<void> main() async {
  final saved = Stream.fromIterable(['hi', 'hello', 'hey'])
      .asyncMap(save); // one save at a time, in order
  print(await saved.toList()); // [2, 5, 3]
}

go deeper

for a junior

Recall that asyncMap is map for functions returning a Future, and asyncExpand is for functions returning a Stream.

for a middle

Explain that both pause the source while work runs, so events are processed one at a time and in order, and that errors pass through as error events.

for a senior

Spot where sequential asyncMap limits throughput or delays fresh events, and choose batching or a concurrent operator deliberately when order is not required.

for a principal

Make ordering and concurrency an explicit decision for each processing stage, since the default here trades throughput for strict order.

## Three ways to map a stream All three are methods on `Stream<T>` in `dart:async`, and all three return a new stream that listens to the source lazily. | Method | Function type | Output per event | |---|---|---| | `map<S>(S convert(T event))` | Synchronous | Exactly one value | | `asyncMap<E>(FutureOr<E> convert(T event))` | May return a `Future` | Exactly one value, after the future completes | | `asyncExpand<E>(Stream<E>? convert(T event))` | Returns a stream, or `null` | Every event of that inner stream (none for `null`) | The returned stream is broadcast if the source is broadcast, and single-subscription otherwise. ## How asyncMap works Inside, `asyncMap` builds a controller whose `onListen` subscribes to the source. For each data event it calls `convert`: 1. If `convert` throws synchronously, the error is added to the output and the next event is processed. 2. If it returns a plain value, the value is added immediately. 3. If it returns a **`Future`**, the source subscription is **paused**, and when the future completes its value — or its error — is added and the source is **resumed**. Because of step 3, only one future is ever outstanding. Events are handled strictly **one at a time** and **in source order**; a slow future holds back everything behind it. Source errors and the done event pass straight through, and the output's pause, resume and cancel are forwarded to the source subscription. ## How asyncExpand works `asyncExpand` is the stream-returning sibling. For each data event it calls `convert`; if that returns a stream, it pauses the source and pipes the inner stream into the output with **`addStream`**. When the inner stream finishes, the source resumes and the next event produces the next inner stream. The effect is **concatenation**: all of the first inner stream, then all of the second. A `null` result skips the event. ## A chat example - **`asyncMap`**: each incoming `ChatMessage` is written to a local database with an `async` function returning the saved row. Order matters — message 12 must not be stored before message 11 — and one write at a time is exactly what `asyncMap` gives. - **`asyncExpand`**: when the client reconnects, each room id in a stream of room ids expands into a stream of that room's missed messages from a paginated history call; rooms are replayed one after another rather than mixed together. ## Consequences to state in an interview - **No parallelism.** Ten slow database writes take ten times as long; `asyncMap` never overlaps them. If ordering is not required and throughput is, you need a different design — batching, or an operator from a reactive library that runs work concurrently. - **No cancellation of in-flight work.** When a newer event arrives, `asyncMap` does not abandon the pending future; it waits for it. Latest-wins behaviour is not what these methods do. - **Backpressure for free.** Since the source is paused while work runs, a fast producer that honours pause is slowed to the consumer's pace. - **Errors do not stop the stream.** A failing future becomes an error event and processing continues with the next event, unless the listener subscribed with `cancelOnError: true`.

  • If the future returned for one event in asyncMap fails, does the stream end?
    No. The error is added to the output as an error event, the source is resumed, and the next event is processed. The stream ends only if the listener subscribed with `cancelOnError: true`, or when the source itself completes.
  • How is asyncExpand different from calling expand on the stream?
    `expand` maps each event to a synchronous `Iterable` and emits its elements immediately. `asyncExpand` maps each event to a `Stream`, waits for that stream to finish while the source is paused, and relays its data and error events, so the inner sequence can be asynchronous.

A single passport desk: each traveller is fully processed before the next one steps up, and a traveller whose form needs an extra check holds the whole queue until it is done.

saying these in an interview costs you the question

  • asyncMap starts every event's future at once and emits them as they finish.
  • asyncExpand interleaves events from all inner streams.
  • A failed future inside asyncMap closes the stream.
  • asyncMap cancels the pending future when a newer event arrives.
  • asyncMap always returns a single-subscription stream.