Can exceptions thrown in dart streams be handled by subscribers without closing the stream?

Question

Short example of what I'm having trouble understanding:

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

With usage:

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

This will stop at 3 with the output:

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

From the documentation (https://dart.dev/tutorials/language/streams) and (https://api.dart.dev/stable/2.9.1/dart-async/Stream/handleError.html), this is as expected, as exceptions thrown will automatically close the stream.

1. Does this mean that the correct way to handle exceptions in a stream, so that subscriptions can be long-lived in such an event, is to handle the exception inside the stream itself? That it is not possible to do so from the outside?
2. Is this the same for broadcast streams?
3. If I'm thinking about this in the wrong way, what are some pointers to start thinking right?

I'm currently thinking of streams as being a source of asynchronous data events that occasionally might be error events. From the documentation and examples, it all looks neat, but I'm thinking that wanting to handle errors and otherwise continue observing the data stream is a normal use case. I'm having a hard time writing the code to do so. But, I might be going about this wrong. Any insights will be much appreciated.

---

Edit: I can add that I've tried various things like using a stream transformer, with the same result:

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');
});

Also, listen() has an optional argument cancelOnError that looks promising, but it defaults to false, so no cigar here.

Answer 1

The generator method

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

will terminate when you throw an exception. The throw works normally, it doesn't directly add the exception to the stream. So, it propagates out through the loop and the method body, until the entire method body ends with the thrown exception. At that point the unhandled exception is added to the stream, and then the stream is closed because the body has ended.

So, the problem is not with the handling, but with the generation of the stream. You must indeed handle the error locally to avoid it ending the stream generating body.

You can't add more than one error to a stream using throw in an async* method, and the error will be the last thing that stream does.

The availabe hack to actually emit more than one error is to yield the exception:

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

That will emit an exception directly into the generated stream without throwing it locally and ending the generator method body.