Bound the stalled read

Toss-up async

Requirement

Read four probe values in sequence; the third read stalls for 500 ms. Give every read a 150 ms budget: print the readings that arrive in time, then reading timed out for the stall, and stop — the fourth read must not be reported. The stall is injected deterministically in the code; both versions must print the lines shown under Expected output.

Expected output

reading: 21.5
reading: 21.7
reading timed out

Side by side

RxDart

import 'package:rxdart/rxdart.dart';

/// Four probe reads; the third stalls far past the 150 ms budget.
const readMs = [30, 40, 500, 30];
const probeValues = [21.5, 21.7, 22.4, 21.9];

Future<double> readProbe(int i) async {
  await Future<void>.delayed(Duration(milliseconds: readMs[i]));
  return probeValues[i];
}

Future<void> main() async {
  final lines = await Stream.fromIterable([0, 1, 2, 3])
      // The budget is per read, so bound the Future itself. (The stream-
      // level .timeout operator measures a different thing: gaps between
      // events on the push side.)
      .asyncMap((i) =>
          readProbe(i).timeout(const Duration(milliseconds: 150)))
      .map((v) => 'reading: ${v.toStringAsFixed(1)}')
      // Replace the timeout error with a report line…
      .onErrorReturnWith((_, _) => 'reading timed out')
      // …and stop there: the paused source would otherwise resume pushing
      // readings once the stalled read finally lands.
      .takeWhileInclusive((line) => line != 'reading timed out')
      .toList();

lines.forEach(print);
}

FxDart

Why they differ

The per-read budget itself is easy on both sides — the RxDart panel bounds the Future inside asyncMap, FxDart's timeout bounds the pull — so the interesting difference is what each model's stream-level operator of the same name measures. Stream.timeout watches the gap between events: the producer decides when values arrive, so "too slow" can only mean "nothing has arrived lately". FxDart's timeout bounds demand-to-item time: the consumer asks, and the clock runs from the ask to the answer. In this finite, sequential task the two would coincide — but they are genuinely different quantities: a pull pipeline with no demand has no gaps to measure, and a push stream owes no answer to anyone's ask.

Each side then needs one real wrinkle to meet the "then stop" clause. On the push side the stalled source is still out there, and it would resume pushing readings once the slow read finally lands — so after onErrorReturnWith converts the error into the report line, takeWhileInclusive ends the stream and cancels the subscription. On the pull side stopping is free — the TimeoutException simply exits the loop and nothing pulls again — but keeping the readings that preceded the stall means collecting with each instead of a toList that would have discarded them when it threw.

A tie: one operator plus one wrinkle on each side, and the wrinkles are mirror images of each model's nature.

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

FxDart 145 µs

Peak memory Tie

RxDart 16.6 MB

FxDart 17.1 MB

N = 10,000

Time FxDart wins

RxDart 14.3 ms

FxDart 12.9 ms

Peak memory Tie

RxDart 27.4 MB

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