订阅者可以在不关闭流的情况下处理 dart 流中抛出的异常吗？

Question

我无法理解的简短示例：

Stream<int> getNumbersWithException() async* {
  for (var i = 0; i < 10; i++) {
    yield i;
    if (i == 3) throw Exception();
  }
}

与用法：

getNumbersWithException()
    .handleError((x) => print('Exception caught for $x'))
    .listen((event) {
  print('Observed: $event');
});

这将在 3 处停止并输出：

Observed: 0
Observed: 1
Observed: 2
Observed: 3
Exception caught for Exception: foo

从文档（https://dart.dev/tutorials/language/streams）和（https://api.dart.dev/stable/2.9.1/dart-async/Stream/handleError.html）来看，这是这样的预期的，因为抛出的异常将自动关闭流。

1. 这是否意味着处理流中的异常（以便订阅可以在此类事件中长期存在）的正确方法是处理流本身内部的异常？从外部无法做到这一点吗？
2. 这对于广播流来说是一样的吗？
3. 如果我以错误的方式思考这个问题，有哪些指示可以让我开始正确思考？

我目前认为流是异步数据事件的来源，有时可能是错误事件。从文档和示例来看，这一切看起来都很整洁，但我认为想要处理错误并继续观察数据流是一个正常的用例。我很难编写代码来做到这一点。但是，我可能会做错事。任何见解将不胜感激。

---

编辑：我可以补充一点，我尝试了各种方法，例如使用流转换器，结果相同：

var transformer = StreamTransformer<int, dynamic>.fromHandlers(
  handleData: (data, sink) => sink.add(data),
  handleError: (error, stackTrace, sink) =>
      print('Exception caught for $error'),
  handleDone: (sink) => sink.close(),
);
getNumbersWithException().transform(transformer).listen((data) {
  print('Observed: $data');
});

另外，listen()有一个可选参数cancelOnError，看起来很有希望，但它默认为false，所以这里没有雪茄。

Answer 1

生成器方法

Stream<int> getNumbersWithException() async* {
  for (var i = 0; i < 10; i++) {
    yield i;
    if (i == 3) throw Exception();
  }
}

当你抛出异常时将终止。工作throw正常，它不会直接将异常添加到流中。因此，它通过循环和方法体传播出去，直到整个方法体以抛出异常结束。此时，未处理的异常将添加到流中，然后流将关闭，因为主体已结束。

因此，问题不在于处理，而在于流的生成。您确实必须在本地处理错误，以避免它结束流生成主体。

throw您不能在方法中使用向流添加多个错误async*，并且该错误将是流执行的最后一件事。

实际上发出多个错误的可用黑客方法是产生异常：

  if (i == 3) yield* () async* { throw Exception(); }();
  // or:      yield* Stream.fromFuture(Future.error(Exception());
  // or:      yield* Stream.error(Exception()); // Since Dart 2.5

这将直接向生成的流发出异常，而不会在本地抛出异常并结束生成器方法体。

Answer 2

使用yield* Stream.error()而不是抛出错误。

例子：

Stream<int> getStream() async* {
  for (var i = 0; i < 5; i++) {
    yield i;
    if (i == 2 || i == 3) {
      yield* Stream.error('Custom error at index $i');
    }
  }
}

void main(List<String> arguments) {
  var stream = getStream();
  stream.listen((event) => print('Data: $event'),
      onDone: () => print('Done'), onError: (err) => print('Error: $err'));
}

输出：

Data: 0
Data: 1
Data: 2
Error: Custom error at index 2
Data: 3
Error: Custom error at index 3
Data: 4
Done