RxDart shareReplay不是“热”吗？

Question

在我的应用程序中，我很难成功地使用来自RxDart (0.24.1)的RxDart。我有一个Stream，我希望‘缓存’最新发出的值，以便任何‘迟到’的侦听器立即得到它(流可以空闲更长时间)。我没能做到这一点，所以我开始在我的应用程序之外进行实验，必须承认我并不完全理解正在发生的事情。

以下是一些代码：

import 'package:rxdart/rxdart.dart';

void main() async {
  final s = Stream.fromIterable([1, 2, 3]).shareReplay(maxSize: 10);
  print('First:');
  await for (final i in s) {
    print(i);
  }
  print('Second:');
  await for (final i in s) {
    print(i);
  }
}

我希望代码能打印出来

First
1
2
3
Second
1
2
3

但是它从来没有到达第二个await for，它甚至没有打印字符串Second:；但是，程序成功地完成了，所以第一个await for确实完成了(因为原始的Stream是有限的)。怎么一回事？

实际上，我所拥有的Stream可能是无限的，在事件之间有很长的时间)，而且我只对最后一个事件感兴趣，下面是一些模拟以下内容的代码：

import 'package:rxdart/rxdart.dart';

void main() async {
  final s = Stream.periodic(Duration(seconds: 1), (e) => e).shareReplay(maxSize: 1);
  // print(s.isBroadcast);
  await Future.delayed(Duration(seconds: 3));
  print('First');
  await for (final i in s.take(1)) {
    print(i);
  }
  print('Second');
  await for (final i in s.take(1)) {
    print(i);
  }
}

为了确保Stream能够完成(所以await for可以完成)，我使用take(1)。我预计产出如下：

First
2
Second
2

(也许这两个人是三个？)我预计会发生这样的事情：

周期流每秒钟开始发射(因为broadcast/hot).

• The应用程序等待3秒，因为Future.delayed.

• The第一次迟来侦听器并获得最新值，2(或者可能是3?)由于take(1).

• The第二次延迟侦听器而结束，也会获得最新的值，因为在此期间，流没有发出任何其他任何内容(即使发出，值也会大1，这是可以的)，并且由于take(1).

• The应用程序完成而结束。

然而，产出是：

First
0
Second
Unhandled exception:
type '_TakeStream<int>' is not a subtype of type 'Future<bool>'
#0      _StreamIterator._onData (dart:async/stream_impl.dart:1067:19)
#1      _RootZone.runUnaryGuarded (dart:async/zone.dart:1384:10)
#2      _BufferingStreamSubscription._sendData (dart:async/stream_impl.dart:357:11)
#3      _BufferingStreamSubscription._add (dart:async/stream_impl.dart:285:7)
#4      _ForwardingStreamSubscription._add (dart:async/stream_pipe.dart:127:11)
#5      _TakeStream._handleData (dart:async/stream_pipe.dart:318:12)
#6      _ForwardingStreamSubscription._handleData (dart:async/stream_pipe.dart:157:13)
#7      _RootZone.runUnaryGuarded (dart:async/zone.dart:1384:10)
#8      _BufferingStreamSubscription._sendData (dart:async/stream_impl.dart:357:11)
#9      _BufferingStreamSubscription._add (dart:async/stream_impl.dart:285:7)
#10     _SyncBroadcastStreamController._sendData (dart:async/broadcast_stream_controller.dart:385:25)
#11     _BroadcastStreamController.add (dart:async/broadcast_stream_controller.dart:250:5)
#12     _StartWithStreamSink._safeAddFirstEvent (package:rxdart/src/transformers/start_with.dart:56:12)
#13     _StartWithStreamSink.onListen (package:rxdart/src/transformers/start_with.dart:37:11)
#14     forwardStream.<anonymous closure>.<anonymous closure> (package:rxdart/src/utils/forwarding_stream.dart:31:37)
#15     forwardStream.runCatching (package:rxdart/src/utils/forwarding_stream.dart:24:12)
#16     forwardStream.<anonymous closure> (package:rxdart/src/utils/forwarding_stream.dart:31:16)
#17     _runGuarded (dart:async/stream_controller.dart:847:24)
#18     _BroadcastStreamController._subscribe (dart:async/broadcast_stream_controller.dart:213:7)
#19     _ControllerStream._createSubscription (dart:async/stream_controller.dart:860:19)
#20     _StreamImpl.listen (dart:async/stream_impl.dart:493:9)
#21     DeferStream.listen (package:rxdart/src/streams/defer.dart:37:18)
#22     StreamView.listen (dart:async/stream.dart:1871:20)
#23     new _ForwardingStreamSubscription (dart:async/stream_pipe.dart:118:10)
#24     new _StateStreamSubscription (dart:async/stream_pipe.dart:341:9)
#25     _TakeStream._createSubscription (dart:async/stream_pipe.dart:310:16)
#26     _ForwardingStream.listen (dart:async/stream_pipe.dart:83:12)
#27     _StreamIterator._initializeOrDone (dart:async/stream_impl.dart:1041:30)
#28     _StreamIterator.moveNext (dart:async/stream_impl.dart:1028:12)
#29     main (package:shopping_list/main.dart)
<asynchronous suspension>
#30     _startIsolate.<anonymous closure> (dart:isolate-patch/isolate_patch.dart:301:19)
#31     _RawReceivePortImpl._handleMessage (dart:isolate-patch/isolate_patch.dart:168:12)

第一个实际使其完成的await for与我预期的一样工作，但它得到的是初始值，这意味着在监听器出现之前流没有发出任何信息(不是“热”)。第二个异常失败。我遗漏了什么？

编辑：我对shareReplay的理解可能一直都是错误的，我认为它很热，因为https://pub.dev/documentation/rxdart/latest/rx/ReplaySubject-class.html说它很热。

我稍微修改了代码：

import 'package:rxdart/rxdart.dart';

void main() async {
  final s = BehaviorSubject();
  s.addStream(Stream.periodic(Duration(seconds: 1), (e) => e));
  // print(s.isBroadcast);
  await Future.delayed(Duration(seconds: 3));
  print('First');
  await for (final i in s.take(1)) {
    print(i);
  }
  print('Second');
  await for (final i in s.take(1)) {
    print(i);
  }
}

现在的输出是：

First
2
Second
2

但节目从未结束..。

如果没有take(1)，输出就永远不会到达第二个侦听器，因此它确实有一定的效果。我还是不明白。

编辑2：我想我可能知道为什么第一个shareReplay示例不能工作，下面是文档的摘录：

它将在第一次收听时自动开始发出项目，当没有侦听器存在时将自动关闭。在我的示例中，第一个侦听器/await for订阅，流启动，当循环完成时，最后(也是唯一)侦听器完成，因此从shareReplay返回的流关闭。但是，仍然无法解释为什么行print('Second');不执行.

Answer 1

您的最终节目没有终止的原因是您的广播流仍然在每秒钟发出事件。没有人在听，但这是一个广播流，所以它不在乎。它每秒钟都有一个计时器在滴答作响，即使没有人再听，这个计时器也能使程序保持活力。您也可以通过执行Timer.periodic(Duration(seconds: 1), (_) {});来保持程序的活力。只要计时器是活的，就会发生一些事情。

不过，这次坠机很有趣。它提出了StreamIterator类内部逻辑的一个问题。我会调查的。

我不能说任何关于shareReplay的保证有用的东西，我不熟悉RxDart。我的猜测是，第一个程序会出现问题，因为replayShared流没有正确终止-也就是说，当它完成时不会发送done事件。这就可以解释为什么程序会提前结束。第一个await for正在等待下一个事件，所以它不会结束并转到下一个循环，但是没有任何类型的活动定时器或计划好的事件，所以隔离器就会关闭。快速阅读BeheaviorSubject表明它像流控制器一样工作，所以您可能需要在addStream完成后关闭它。也许：

s.addStream(...).whenComplete(s.close);