`retryOn`, `repeat` & `whenComplete`

Resubscribe on error or on completion — with a budget, a delay, or a notifier that decides when to try again.

FxEvents<T> FxEvents<T>.retryOnError({int? count, Duration Function(int attempt)? delay}) // chain (events) FxEvents<T> FxEvents<T>.retryOn(Stream<void> Function(FxEvents<Object> errors) notifier) FxEvents<T> FxEvents<T>.repeat({int? count, Duration Function(int repeatCount)? delay}) FxEvents<T> FxEvents<T>.repeatOn(Stream<void> Function(FxEvents<void> completions) notifier) FxEvents<T> FxEvents<T>.whenComplete(void Function() callback)

Lecture

The pull layer's retry rebuilds an iterable. On the push side the same idea is a resubscribe, and there are two shapes. One rebuilds the stream from a factory — FxEvents.retry, already covered — which is the right move for a one-shot StreamController. The other re-listens the same stream, and that is this page: retryOnError, retryOn, repeat, repeatOn.

Re-listening needs a source that allows it. Stream.multi, Stream.fromIterable, and a broadcast all do; a spent single-subscription controller will error on the second listen. Reach for FxEvents.retry (or FxEvents.defer) when the source is one-shot.

retryOnError({count, delay}) resubscribes on error — forever when count is null, or up to that many retries (so count: 2 is three attempts). delay if set is consulted with the 1-based retry number before each retry. When the budget is spent the last error is forwarded and the stream closes.

retryOn(notifier) is Rx's retryWhen: the error is not forwarded; it is pushed into notifier, and a next on that stream resubscribes. Notifier complete completes the result without the error; notifier error is forwarded. repeat / repeatOn are the same pair on completion rather than error — errors forward and stop. whenComplete is Rx finalize: the callback runs exactly once on done, error, or cancel. fxdart events layer, after Rx's retryWhen, retry, repeat, repeatWhen, and finalize.

Demo 1 · retryOnError, with delay

Demo 2 · Repeat a short stream twice

Try it yourself

Exercise: retryOn — the notifier decides when to resubscribe.

import 'dart:async';

import 'package:fxdart/fxdart.dart';

Future<void> main() async {
  // Exercise: retryOn is retryWhen — on error the source is not
  // forwarded; the error is pushed into a notifier, and a next on that
  // notifier resubscribes. Here the notifier is the errors themselves,
  // so each failure retries immediately.
  var attempts = 0;
  final source = Stream<String>.multi((c) {
    attempts++;
    if (attempts == 1) {
      c
        ..addError(StateError('down'))
        ..close();
    } else {
      c
        ..add('connected')
        ..close();
    }
  });

final out = await fxEvents(source)
      .retryOn((errors) => errors.map((_) {
            print('retrying');
          }).stream)
      .toList();

print('attempts: $attempts'); // 2
  print(out); // [connected]
  // Notifier complete completes the result without forwarding the
  // error; notifier error is forwarded. Same re-listen caveat as
  // retryOnError: use Stream.multi, not a spent controller.
}

Related: FxEvents.retry — the factory form, for sources you cannot re-listen · retry — the pull-layer original, with a backoff hook and per-element scope · timeout — bound how long a pull may take, rather than how often it retries

retryOn, repeat & whenComplete

Lecture

Demo 1 · retryOnError, with delay

Demo 2 · Repeat a short stream twice

Try it yourself

`retryOn`, `repeat` & `whenComplete`