有没有办法在等待另一个流时处理流错误而不停止流？

huangapple
go评论106阅读模式
英文:

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

问题

我想通过另一个流来监听一些流。当我们使用使用另一个流的流时，我们需要使用await关键字，但我找不到捕获另一个流的错误并继续流数据的方法。

这是我的程序。

  1. Future<void> main() async {
  2.   // 我想通过stream2来监听stream1，并希望在不停止流的情况下处理错误。
  3.   stream2().listen(print, onError: print);
  4.   /* 
  5.    * 结果:
  6.    * 
  7.    * 0
  8.    * 1
  9.    * 2
  10.    * Exception: Error at 3
  11.    * 
  12.    */
  13.   
  14.   
  15.   // 如果直接监听stream1，我们可以处理错误并继续获取流。
  16.   // stream1.listen(print, onError: print);
  17.   /* 
  18.    * 结果: 这是我想要的结果。
  19.    * 
  20.    * 0
  21.    * 1
  22.    * 2
  23.    * Exception: Error at 3
  24.    * 4
  25.    * 5
  26.    * 
  27.    */
  28. }
  30. // 这个流是不可修改的。
  31. final stream1 = Stream<int>.periodic(Duration(seconds: 1), (i) {
  32.   if (i == 3) {
  33.     throw Exception("Error at $i");
  34.   }
  35.   return i;
  36. }).take(5);
  39. Stream<int> stream2() async* {
  40.   await for (final item in stream1) {
  41.     // 有没有一种方法可以在不停止stream2的情况下处理错误？
  42.     try {
  43.       yield item;
  44.     } catch (e) {
  45.       yield* Stream.error(e);
  46.     }
  47.   }
  48. }
英文:

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.

  1. Future&lt;void&gt; main() async {
  2.   // I want to listen stream1 via stream2, and want to handle error without stop
  3.   // stream.
  4.   stream2().listen(print, onError:print);
  5.   /* 
  6.    * Result:
  7.    * 
  8.    * 0
  9.    * 1
  10.    * 2
  11.    * Exception: Error at 3
  12.    * 
  13.    */
  14.   
  15.   
  16.   // If we directly listen stream1, we can handling error and continue
  17.   // to get stream.
  18.   // stream1.listen(print, onError:print);
  19.   /* 
  20.    * Result: this is the result i want.
  21.    * 
  22.    * 0
  23.    * 1
  24.    * 2
  25.    * Exception: Error at 3
  26.    * 4
  27.    * 5
  28.    * 
  29.    */
  30. }
  32. // This Stream is not modifiable.
  33. final stream1 = Stream&lt;int&gt;.periodic(Duration(seconds: 1), (i) {
  34.   if (i == 3) {
  35.     throw Exception(&quot;Error at $i&quot;);
  36.   }
  37.   return i;
  38. }).take(5);
  41. Stream&lt;int&gt; stream2() async* {
  42.   await for (final item in stream1) {
  43.     // Is there a way we handle error without stop stream2?
  44.     try {
  45.       yield item;
  46.     } catch (e) {
  47.       yield* Stream.error(e);
  48.     }
  49.   }
  50. }

答案1

得分: 0

你可以在流上使用handleError方法。

  1. Future<void> main() async {
  2.   stream2().listen(print, onError: print);
  3. }
  5. final stream1 = Stream<int>.periodic(Duration(seconds: 1), (i) {
  6.   if (i == 3) {
  7.     throw Exception("Error at $i");
  8.   }
  9.   return i;
  10. }).take(5);
  12. Stream<int> stream2() async* {
  13.   await for (final item in stream1.handleError(print)) {
  14.     yield item;
  15.   }
  16. }

输出：

  1. 0
  2. 1
  3. 2
  4. Exception: Error at 3
  5. 4
  6. 5

如果你需要在错误处理部分使用yield或yield*，那么.handleError方法将无法工作，因为你无法在另一个函数内部从外部函数中使用yield。

我认为await-for和try-catch也不会起作用。try需要在await-for循环的外部才能捕获错误，但一旦捕获到错误，你就不再在循环中了。

你可以使用async库中的StreamQueue，你需要将其添加到pubspec.yaml文件中作为依赖项。

  1. import 'package:async/async.dart';
  3. ...
  5. Stream<int> stream2() async* {
  6.   StreamQueue<int> queue = StreamQueue<int>(stream1);
  7.   while (await queue.hasNext) {
  8.     try {
  9.       final item = await queue.next;
  10.       yield item;
  11.     } catch (e) {
  12.       print(e);
  13.       // 在这里可以使用yield或yield*
  14.     }
  15.   }
  16. }
英文:

You can use the handleError method on your stream.

  1. Future&lt;void&gt; main() async {
  2.   stream2().listen(print, onError: print);
  3. }
  5. final stream1 = Stream&lt;int&gt;.periodic(Duration(seconds: 1), (i) {
  6.   if (i == 3) {
  7.     throw Exception(&quot;Error at $i&quot;);
  8.   }
  9.   return i;
  10. }).take(5);
  12. Stream&lt;int&gt; stream2() async* {
  13.   await for (final item in stream1.handleError(print)) {
  14.     yield item;
  15.   }
  16. }

Output:

  1. 0
  2. 1
  3. 2
  4. Exception: Error at 3
  5. 4
  6. 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.

  1. import &#39;package:async/async.dart&#39;;
  3. ...
  5. Stream&lt;int&gt; stream2() async* {
  6.   StreamQueue&lt;int&gt; queue = StreamQueue&lt;int&gt;(stream1);
  7.   while (await queue.hasNext) {
  8.     try {
  9.       final item = await queue.next;
  10.       yield item;
  11.     } catch (e) {
  12.       print(e);
  13.       // can do a yield or yield* here
  14.     }
  15.   }
  16. }

答案2

得分: 0

不带外部包的方法也起作用。我会留下代码供参考。

  1. Stream<int> stream2() async* {
  2.   final store = StreamTemporaryStore<int>(stream1);
  3.   while (store.hasNext) {
  4.     final val = await store.next();
  5.     if (val != null) {
  6.       if (val.error == null) {
  7.         yield val.data!;
  8.       } else {
  9.         yield* Stream.error(val.error!);
  10.       }
  11.     }
  12.   }
  13. }
  15. class StreamTemporaryStore<T> {
  16.   StreamTemporaryStore(this.stream) {
  17.     stream.listen((T data) {
  18.       list = [(data: data, error: null), ...list];
  19.     }, onError: (Object error) {
  20.       list = [(data: null, error: error), ...list];
  21.     }, onDone: () {
  22.       hasNext = false;
  23.     });
  24.   }
  25.   final Stream<T> stream;
  26.   List<({T? data, Object? error})> list = [];
  27.   bool hasNext = true;
  29.   Future<({T? data, Object? error})?> next() async {
  30.     await Future(() {});
  31.     if (list.isEmpty) return null;
  32.     final head = list[list.length - 1];
  33.     list.removeAt(list.length - 1);
  34.     return head;
  35.   }
  36. }

希望对你有帮助！

英文:

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

  1. Stream&lt;int&gt; stream2() async* {
  2.   final store = StreamTemporaryStore&lt;int&gt;(stream1);
  3.   while (store.hasNext) {
  4.     final val = await store.next();
  5.     if (val != null) {
  6.       if (val.error == null) {
  7.         yield val.data!;
  8.       } else {
  9.         yield* Stream.error(val.error!);
  10.       }
  11.     }
  12.   }
  13. }
  15. class StreamTemporaryStore&lt;T&gt; {
  16.   StreamTemporaryStore(this.stream) {
  17.     stream.listen((T data) {
  18.       list = [(data: data, error: null), ...list];
  19.     }, onError: (Object error) {
  20.       list = [(data: null, error: error), ...list];
  21.     }, onDone: () {
  22.       hasNext = false;
  23.     });
  24.   }
  25.   final Stream&lt;T&gt; stream;
  26.   List&lt;({T? data, Object? error})&gt; list = [];
  27.   bool hasNext = true;
  29.   Future&lt;({T? data, Object? error})?&gt; next() async {
  30.     await Future(() {});
  31.     if (list.isEmpty) return null;
  32.     final head = list[list.length - 1];
  33.     list.removeAt(list.length - 1);
  34.     return head;
  35.   }
  36. }

huangapple
  • 本文由 发表于 2023年8月9日 09:07:14
  • 转载请务必保留本文链接：https://go.coder-hub.com/76863964.html
  • async-await
  • dart
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定