`onErrorReturn`, `onErrorResume` & `retry`

Tres profundidades de recuperación: parchear cada error con un valor, abandonar la fuente por una alternativa, o tirar el stream entero y reconstruirlo.

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])

Lección

Los errores se comportan de otra forma en el lado push, y la diferencia hace tropezar. En un pipeline pull una excepción termina la iteración: hay un fallo y después nada. En un Stream de Dart un error es solo otro evento: se entrega y la suscripción continúa. Un stream puede emitir diez errores y cuarenta valores y aun así cerrarse con normalidad.

Por eso onErrorReturn(value) es una sustitución por error y no un rescate de una sola vez. Cada error se convierte en un evento value y el stream sigue adelante: lo correcto para un sensor inestable, donde una lectura mala debería convertirse en un marcador de posición y el feed debería sobrevivir.

onErrorResume(f) es el cambio de una sola vez. Al primer error la fuente se cancela de golpe y el stream que f construye a partir de ese error toma el relevo para siempre: la jugada de tirar de caché cuando falla la red. No se vuelve a ver nada de la fuente original, y un error lanzado por el propio f se reenvía en vez de tragarse.

FxEvents.retry(factory, [count]) trabaja un nivel más arriba: no parchea los errores de un stream, reconstruye el stream. Ante un error se tira el intento fallido y se vuelve a llamar a factory() para una suscripción nueva: la forma correcta cuando el fallo es la conexión misma. El presupuesto cuenta resuscripciones, así que count: 2 permite tres intentos como mucho; cuando se agota, el último error se reenvía y el stream se cierra. Los eventos que un intento ya emitió no se retiran, así que la factoría debería producir algo reproducible.

Capa de eventos de fxdart, siguiendo a onErrorReturn, onErrorResume y Rx.retry de Rx. Para fallos que quieres modelar en vez de recuperar, attempt los mueve al canal de valores como un Left tipado — el puente de la capa de eventos hacia Either y Raise.

Demo 1 · Un valor por error

Demo 2 · Abandonar la fuente por una alternativa

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.
}

Pruébalo tú

Ejercicio: reconstruir un stream inestable, con y sin presupuesto.

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
}

Relacionado: retry — el original de la capa pull, con gancho de backoff y ámbito por elemento · attempt — los mismos fallos, como Lefts tipados en el canal de valores · Either — errores como valores con tipo en vez de eventos de los que recuperarse

onErrorReturn, onErrorResume & retry

Lección

Demo 1 · Un valor por error

Demo 2 · Abandonar la fuente por una alternativa

Pruébalo tú

`onErrorReturn`, `onErrorResume` & `retry`