Give up after three failures

FxDart wins async

Requirement

A feed of ten health probes runs in order; probes 2, 5, 7, 8 and 9 throw. Stop the run the moment the third failure is seen (including it), then print three counts: processed — probes that entered the pipeline before the cut; failures — how many of those threw; and probes run — probe bodies actually executed, tallied by a side-effect counter inside the probe itself. The later probes must never execute, so the run count has to match the processed count. The schedule is in the code; both versions must print the lines shown under Expected output.

Expected output

processed: 7
failures: 3
probes run: 7

Side by side

RxDart

import 'package:rxdart/rxdart.dart';

const probeIds = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
const failing = {2, 5, 7, 8, 9};

int runs = 0;

// One health probe — throws when the endpoint is down.
Future<void> probe(int id) async {
  runs++;
  await Future.delayed(const Duration(milliseconds: 5));
  if (failing.contains(id)) throw StateError('probe $id failed');
}

Future<void> main() async {
  // Failures must become data before scan can count them — each probe
  // gets an inner stream that recovers the error as a marker value.
  final states = await Stream.fromIterable(probeIds)
      .asyncExpand((id) => Rx.fromCallable(() async {
            await probe(id);
            return true;
          }).onErrorReturn(false))
      .scan<({int done, int fails})>(
          (acc, ok, _) => (done: acc.done + 1, fails: acc.fails + (ok ? 0 : 1)),
          (done: 0, fails: 0))
      .takeWhileInclusive((s) => s.fails < 3)
      .toList();

final last = states.last;
  print('processed: ${last.done}');
  print('failures: ${last.fails}');
  print('probes run: $runs');
}

FxDart

Why they differ

The counting core is the same on both sides — scan folds a running (done, fails) state, and a take-inclusive operator cuts the pipeline at the third failure (takeUntilInclusive(fails == 3) on one side, takeWhileInclusive(fails < 3) on the other). Both also genuinely stop the work: probes run: 7 proves that cancelling the subscription and ceasing to pull are equally effective brakes.

The difference is what each side had to do before scan could count. A thrown probe lives on the stream's error channel, where scan cannot see it — and where it would end the stream at failure number one. So the RxDart side first converts every probe into an inner stream (Rx.fromCallable + onErrorReturn(false)) to smuggle failures back onto the data channel as marker values. The FxDart side needs no conversion step, because there is nothing to convert from: a try/catch inside map makes the outcome a bool right where it happens, and the rest of the pipeline is arithmetic. Same operators, one fewer model boundary to cross.

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

FxDart 413 µs

Peak memory Tie

RxDart 16.6 MB

FxDart 17.0 MB

N = 10,000

Time FxDart wins

RxDart 65.5 ms

FxDart 37.6 ms

Peak memory Tie

RxDart 22.2 MB

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