`onErrorReturn`, `onErrorResume` & `retry`

Three depths of recovery: patch each error with a value, abandon the source for a fallback, or throw the whole stream away and rebuild it.

FxEvents<T> FxEvents<T>.onErrorReturn(T value) // chain (events) FxEvents<T> FxEvents<T>.onErrorResume(Stream<T> Function(Object, StackTrace) f) static FxEvents<T> FxEvents.retry<T>(Stream<T> Function() factory, [int? count])

Lecture

Errors behave differently on the push side, and the difference trips people up. In a pull pipeline an exception ends the iteration — there is one failure and then nothing. In a Dart Stream an error is just another event: it is delivered, and the subscription carries on. A stream can emit ten errors and forty values and still close normally.

That is why onErrorReturn(value) is a per-error substitution rather than a one-shot rescue. Every error becomes one value event and the stream keeps going — right for a flaky sensor where a bad reading should become a placeholder and the feed should survive.

onErrorResume(f) is the one-shot switch. On the first error the source is cancelled outright and the stream f builds from that error takes over for good — the cache-on-network-failure move. Nothing more of the original source is ever seen, and an error thrown by f itself is forwarded rather than swallowed.

FxEvents.retry(factory, [count]) works one level up: it does not patch a stream's errors, it rebuilds the stream. On error the failed attempt is thrown away and factory() is called again for a fresh subscription — the right shape when the failure is the connection itself. The budget counts re-subscriptions, so count: 2 allows at most three attempts; when it runs out the last error is forwarded and the stream closes. Events an attempt already emitted are not taken back, so the factory should produce something replayable.

fxdart events layer, after Rx's onErrorReturn, onErrorResume and Rx.retry. For failures you want to model rather than recover from, attempt moves them onto the value channel as a typed Left — the events-layer bridge to Either and Raise.

Demo 1 · A value per error

Demo 2 · Abandoning the source for a fallback

import 'dart:async';

import 'package:fxdart/fxdart.dart';

/// Emits each (offsetMs, value) pair at its offset, closing at [closeMs].
Stream<T> timed<T>(List<(int, T)> events, int closeMs) {
  final c = StreamController<T>();
  for (final (ms, v) in events) {
    Timer(Duration(milliseconds: ms), () => c.add(v));
  }
  Timer(Duration(milliseconds: closeMs), c.close);
  return c.stream;
}

Future<void> main() async {
  // A live feed that dies mid-session; fall back to the cache.
  final network = StreamController<String>();
  final out = fxEvents(network.stream)
      .onErrorResume((error, _) => Stream.fromIterable(['cached A', 'cached B']))
      .toList();

network
    ..add('live 1')
    ..addError(StateError('connection lost'))
    ..add('live 2'); // never arrives — the source was abandoned
  await Future<void>.delayed(const Duration(milliseconds: 20));
  await network.close();

print(await out); // [live 1, cached A, cached B]
  // Unlike onErrorReturn this is a ONE-SHOT switch: the source is
  // cancelled at the first error and the fallback takes over entirely.
}

Try it yourself

Exercise: rebuilding a flaky stream, with and without a budget.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise: retry rebuilds the whole stream instead of patching over
  // its errors — the right move when the failure is the connection.
  var attempts = 0;

final out = await FxEvents.retry<String>(() {
    attempts++;
    print('attempt $attempts');
    return attempts < 3
        ? Stream<String>.error(StateError('flaky'))
        : Stream.fromIterable(['ok']);
  }).toList();

print(out); // [ok] after three attempts

// With a budget, retry gives up and forwards the last error.
  var tries = 0;
  final seen = <Object>[];
  final done = Completer<void>();
  FxEvents.retry<String>(() {
    tries++;
    return Stream<String>.error(StateError('always down'));
  }, 2)
      .listen(seen.add, onError: seen.add, onDone: done.complete);
  await done.future;

print('gave up after $tries attempts'); // one attempt plus two retries
  print(seen.single.runtimeType); // StateError
}

Related: retry — the pull-layer original, with a backoff hook and per-element scope · attempt — the same failures, as typed Lefts on the value channel · Either — errors as typed values rather than events to recover from

onErrorReturn, onErrorResume & retry

Lecture

Demo 1 · A value per error

Demo 2 · Abandoning the source for a fallback

Try it yourself

`onErrorReturn`, `onErrorResume` & `retry`