`onErrorReturn`, `onErrorResume` & `retry`

복구의 세 깊이: 에러마다 값으로 때우거나, 소스를 버리고 대체물로 갈아타거나, 스트림 전체를 버리고 다시 만들거나.

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

강의

푸시 쪽에서 에러는 다르게 행동하고, 그 차이가 사람들을 넘어뜨립니다. 풀 파이프라인에서 예외는 순회를 끝냅니다 — 실패 하나, 그리고 아무것도 없음. Dart Stream에서 에러는 그저 또 하나의 이벤트입니다: 전달되고, 구독은 계속됩니다. 스트림 하나가 에러 열 개와 값 마흔 개를 내보내고도 정상적으로 닫힐 수 있습니다.

그래서 onErrorReturn(value)은 한 번의 구조가 아니라 에러마다의 대체입니다. 모든 에러가 하나의 value 이벤트가 되고 스트림은 계속 갑니다 — 나쁜 판독이 자리표시자가 되고 피드는 살아남아야 하는 불안정한 센서에 맞습니다.

onErrorResume(f)이 한 번의 전환입니다. 첫 에러에서 소스는 그 자리에서 취소되고, f가 그 에러로부터 만든 스트림이 영원히 넘겨받습니다 — 네트워크 실패 시 캐시로 가는 수입니다. 원래 소스의 것은 더 이상 보이지 않고, f 자신이 던진 에러는 삼켜지지 않고 전달됩니다.

FxEvents.retry(factory, [count])는 한 층 위에서 작동합니다. 스트림의 에러를 때우는 것이 아니라 스트림을 다시 만듭니다. 에러가 나면 실패한 시도를 버리고 factory()를 다시 불러 새 구독을 얻습니다 — 실패한 것이 연결 자체일 때 맞는 모양입니다. 예산은 재구독을 세므로 count: 2는 최대 세 번의 시도를 허용하고, 다 쓰면 마지막 에러를 전달하고 스트림을 닫습니다. 어떤 시도가 이미 내보낸 이벤트는 되돌려지지 않으므로, 팩토리는 다시 재생 가능한 것을 만들어야 합니다.

fxdart 이벤트 레이어, Rx의 onErrorReturn, onErrorResume와 Rx.retry를 따랐습니다. 복구할 대상이 아니라 모델링하고 싶은 실패라면 attempt가 그것을 값 채널 위의 타입 있는 Left로 옮깁니다 — 이벤트 레이어에서 Either와 Raise로 가는 다리입니다.

데모 1 · 에러마다 값 하나

데모 2 · 소스를 버리고 대체물로

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

직접 해 보기

연습: 불안정한 스트림 다시 만들기, 예산이 있을 때와 없을 때.

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
}

관련 항목: retry — 백오프 훅과 원소 단위 범위를 가진 풀 레이어의 원본 · attempt — 같은 실패를, 값 채널 위의 타입 있는 Left로 · Either — 복구할 이벤트가 아니라 타입 있는 값으로서의 에러

onErrorReturn, onErrorResume & retry

강의

데모 1 · 에러마다 값 하나

데모 2 · 소스를 버리고 대체물로

직접 해 보기

`onErrorReturn`, `onErrorResume` & `retry`