A price, or the list price

FxDart wins async

Requirement

Quote two purchase orders against the August promo price book. The async lookup throws for SKUs with no promo price — those lines must fall back to their list price, and every line must appear in the quote. An order with no lines quotes as the single line (no lines to quote). The data is in the code; both versions must print the lines shown under Expected output.

Expected output

PO-3391:
  SKU-201: 11.50 (promo)
  SKU-333: 9.50 (list)
  SKU-204: 8.25 (promo)
  SKU-207: 19.90 (promo)
  SKU-410: 5.20 (list)
PO-3402:
  (no lines to quote)

Side by side

RxDart

import 'package:rxdart/rxdart.dart';

/// August promo price book — looking up any other SKU throws.
const promoPrices = {'SKU-201': 11.50, 'SKU-204': 8.25, 'SKU-207': 19.90};

/// Purchase orders to quote: (order id, lines of sku + list price).
const orders = [
  (
    'PO-3391',
    [
      (sku: 'SKU-201', list: 14.00),
      (sku: 'SKU-333', list: 9.50),
      (sku: 'SKU-204', list: 9.75),
      (sku: 'SKU-207', list: 24.00),
      (sku: 'SKU-410', list: 5.20),
    ]
  ),
  ('PO-3402', <({String sku, double list})>[]),
];

Future<double> promoPrice(String sku) async {
  await Future<void>.delayed(const Duration(milliseconds: 5));
  final price = promoPrices[sku];
  if (price == null) throw StateError('no promo price for $sku');
  return price;
}

Future<void> main() async {
  for (final (id, lines) in orders) {
    print('$id:');
    final quotes = await Stream.fromIterable(lines)
        // One error channel for the whole stream: by the time an
        // onErrorReturnWith after asyncMap saw the failure, the line that
        // caused it would be gone. Recovery that still knows its line
        // means wrapping each lookup in its own inner stream.
        .flatMap((line) => Rx.fromCallable(() => promoPrice(line.sku))
            .map((p) => '  ${line.sku}: ${p.toStringAsFixed(2)} (promo)')
            .onErrorReturnWith((_, _) =>
                '  ${line.sku}: ${line.list.toStringAsFixed(2)} (list)'))
        .defaultIfEmpty('  (no lines to quote)')
        .toList();
    quotes.forEach(print);
  }
}

FxDart

import 'package:fxdart/fxdart.dart';

/// August promo price book — looking up any other SKU throws.
const promoPrices = {'SKU-201': 11.50, 'SKU-204': 8.25, 'SKU-207': 19.90};

Future<void> main() async {
  for (final (id, lines) in orders) {
    print('$id:');
    final quotes = await fx(lines)
        .toAsync()
        // Recovery is plain control flow inside the callback: catch the
        // typed error right beside the call, return the fallback value.
        .map((line) async {
          try {
            final p = await promoPrice(line.sku);
            return '  ${line.sku}: ${p.toStringAsFixed(2)} (promo)';
          } on StateError {
            return '  ${line.sku}: ${line.list.toStringAsFixed(2)} (list)';
          }
        })
        .defaultIfEmpty('  (no lines to quote)')
        .toList();
    quotes.forEach(print);
  }
}

Why they differ

A stream carries its values and its errors on separate channels, and the error channel belongs to the whole pipeline. By the time an onErrorReturnWith placed after an asyncMap saw the failure, the line that caused it would be gone — an error event carries the error, not the element. The idiomatic RxDart recovery is to give every lookup its own inner stream — flatMap into Rx.fromCallable — so each failure is terminal only to its own one-item stream, where the line is still in scope for the fallback.

In the pull model there is no second channel to fall between. The lookup is an await inside the map callback, so recovery is ordinary Dart control flow: catch the typed StateError right beside the call and return the list price — the element, its fallback, and the error handling all live in the same four lines. No wrapping, no re-merging, and the failure never touches the pipeline at all.

The empty second order lands the same way on both sides — defaultIfEmpty, an operator FxDart took from Rx. The verdict goes to FxDart for the main event: per-item error recovery is a sentence in the pull model and a construction in the push model.

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 455 µs

FxDart 453 µs

Peak memory Tie

RxDart 17.1 MB

FxDart 17.3 MB

N = 10,000

Time FxDart wins

RxDart 42.7 ms

FxDart 38.9 ms

Peak memory RxDart wins

RxDart 49.1 MB

FxDart 52.0 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.