Is there any way to handle stream error without stop stream, when we await for another stream

I want to listen some stream via another stream.
When we use stream which uses another stream, we need to use await for,
however I can't find out the way to catch another stream's error and continue stream data.

Here is my program.

Future<void> main() async {
  // I want to listen stream1 via stream2, and want to handle error without stop
  // stream.
  stream2().listen(print, onError:print);
  /* 
   * Result:
   * 
   * 0
   * 1
   * 2
   * Exception: Error at 3
   * 
   */
  
  
  // If we directly listen stream1, we can handling error and continue
  // to get stream.
  // stream1.listen(print, onError:print);
  /* 
   * Result: this is the result i want.
   * 
   * 0
   * 1
   * 2
   * Exception: Error at 3
   * 4
   * 5
   * 
   */
}

// This Stream is not modifiable.
final stream1 = Stream<int>.periodic(Duration(seconds: 1), (i) {
  if (i == 3) {
    throw Exception("Error at $i");
  }
  return i;
}).take(5);


Stream<int> stream2() async* {
  await for (final item in stream1) {
    // Is there a way we handle error without stop stream2?
    try {
      yield item;
    } catch (e) {
      yield* Stream.error(e);
    }
  }
}

You can use the handleError method on your stream.

Future<void> main() async {
  stream2().listen(print, onError: print);
}

final stream1 = Stream<int>.periodic(Duration(seconds: 1), (i) {
  if (i == 3) {
    throw Exception("Error at $i");
  }
  return i;
}).take(5);

Stream<int> stream2() async* {
  await for (final item in stream1.handleError(print)) {
    yield item;
  }
}

Output:

0
1
2
Exception: Error at 3
4
5

If you need to use yield or yield* inside the error handling part, then .handleError isn't going to work, since you can't yield from the outer function from within another function.

I don't think await-for with try-catch will work either. The try needs to be on the outside of the await-for loop in order to catch the error, but then you are no longer in the loop once the error is caught.

One thing you could do is use a StreamQueue from async library, which you will have to add to your pubspec.yaml as a dependency.

import 'package:async/async.dart';

...

Stream<int> stream2() async* {
  StreamQueue<int> queue = StreamQueue<int>(stream1);
  while (await queue.hasNext) {
    try {
      final item = await queue.next;
      yield item;
    } catch (e) {
      print(e);
      // can do a yield or yield* here
    }
  }
}

The method without external packages also worked. I'll leave the code for reference.

Stream<int> stream2() async* {
  final store = StreamTemporaryStore<int>(stream1);
  while (store.hasNext) {
    final val = await store.next();
    if (val != null) {
      if (val.error == null) {
        yield val.data!;
      } else {
        yield* Stream.error(val.error!);
      }
    }
  }
}

class StreamTemporaryStore<T> {
  StreamTemporaryStore(this.stream) {
    stream.listen((T data) {
      list = [(data: data, error: null), ...list];
    }, onError: (Object error) {
      list = [(data: null, error: error), ...list];
    }, onDone: () {
      hasNext = false;
    });
  }
  final Stream<T> stream;
  List<({T? data, Object? error})> list = [];
  bool hasNext = true;

  Future<({T? data, Object? error})?> next() async {
    await Future(() {});
    if (list.isEmpty) return null;
    final head = list[list.length - 1];
    list.removeAt(list.length - 1);
    return head;
  }
}