guardStream<T> method

Stream<T> guardStream<T>(Stream<T> source, { String? sourceName, FutureOr<void> onError(BaseException e)?, })

Transform a stream so errors are captured by this boundary.

Implementation

Stream<T> guardStream<T>(
  Stream<T> source, {
  String? sourceName,
  FutureOr<void> Function(BaseException e)? onError,
}) {
  late StreamController<T> controller;
  controller = StreamController<T>(
    sync: true,
    onListen: () {
      final sub = source.listen(
        controller.add,
        onError: (Object error, StackTrace stack) async {
          final normalized = _normalize(error, stack, source: sourceName);
          await onError?.call(normalized);
          _setError(normalized);
          await _report(normalized);
          controller.addError(normalized, stack);
        },
        onDone: controller.close,
        cancelOnError: false,
      );
      controller.onCancel = sub.cancel;
    },
  );
  return controller.stream;
}