Resume from cache when the source dies

Toss-up async

Requirement

The live order feed delivers three updates and then the connection dies. The dashboard still needs its first six rows: keep everything the live feed managed to deliver, then continue from last night's cached snapshot, marking those rows (from cache). The failure is injected deterministically in the code; both versions must print the lines shown under Expected output.

Expected output

ORD-7011 packed
ORD-7012 shipped
ORD-7013 packed
ORD-7014 picking (from cache)
ORD-7015 picking (from cache)
ORD-7016 received (from cache)

Side by side

RxDart

FxDart

import 'package:fxdart/fxdart.dart';

/// The live order feed: three updates arrive, then the connection dies.
Stream<String> liveFeed() async* {
  yield 'ORD-7011 packed';
  yield 'ORD-7012 shipped';
  yield 'ORD-7013 packed';
  throw StateError('feed connection lost');
}

/// Last night's cached snapshot of the rest of the queue.
const cachedTail = [
  'ORD-7014 picking',
  'ORD-7015 picking',
  'ORD-7016 received',
  'ORD-7017 received',
];

Future<void> main() async {
  // Pull from the live feed until it dies, keeping everything it managed
  // to deliver — the loop is the explicit error boundary.
  final live = <String>[];
  try {
    await for (final update in liveFeed()) {
      live.add(update);
    }
  } on StateError {
    // The feed died — resume from the cached tail below.
  }

// The dashboard shows the first six updates; take(6) means the cache is
  // only pulled as far as needed — and nothing here is async anymore.
  final updates = fx(live)
      .concat(fx(cachedTail).map((u) => '$u (from cache)'))
      .take(6)
      .toList();

updates.forEach(print);
}

Why they differ

This is the error channel at its best. The recovery point is whole-stream — "when this source dies, switch to that one for the rest of the sequence" — which is exactly the shape a push error channel models. onErrorResumeNext says the entire requirement in one operator: values pass through untouched, and the first error swaps the subscription to the recovery stream mid-flight, with the three delivered updates already safely past.

The pull side has no operator that keeps the values of a source that later throws — a pull pipeline surfaces the error at the pull site, and a failed toList would discard what came before it. So the boundary is written out: an await for loop collects the live rows, the try/catch names the failure, and concat + map + take splice on the cached tail. Two or three honest lines more — the same recovery, minus the vocabulary word.

A toss-up, leaning RxDart on elegance here. Both sides stop at six rows, and neither reads the cache past what the page needs: take cancels the subscription on one side and simply stops pulling on the other.

Benchmark

Apple M1 Max, 32 GB RAM · Dart 3.12.2 (AOT-compiled) · 2026-08-18

Async case: the headline scale is N = 10,000, not 1,000,000. Every element costs an event-loop turn on both sides, so a million real awaits would measure Dart's event loop for minutes — not the pipeline. Delays are zero-length and the example's concurrency limit is kept; what the bars compare is the pipeline machinery.

N = 100

Time Tie

RxDart 28 µs

FxDart 26 µs

Peak memory Tie

RxDart 16.5 MB

FxDart 16.6 MB

N = 10,000

Time Tie

RxDart 1.80 ms

FxDart 1.97 ms

Peak memory Tie

RxDart 23.2 MB

FxDart 23.7 MB

Bars are medians of repeated timed iterations in fresh processes per side (small N is batched for timer resolution). Sides within 5% of each other — or within 0.6 ms, a difference no person can perceive — count as a tie; close relative races are re-measured up to 5 runs. In an app, anything under a few milliseconds is invisible to the user regardless of which bar is shorter. Memory is peak process RSS. The Dart VM and the dataset are identical on both sides, so the difference between the two bars is what the pipeline itself holds onto.