关闭接收无限数据流的主题的最佳方法是什么？

Question

我有一个关于如何正确制作无限数据流以停止添加到主题的问题。像下面的例子：

void main() async {
  var infinite = Stream.periodic(0.5.seconds, (it) => it);

  var subject = BehaviorSubject<int>();
  subject.addStream(infinite);

  subject.listen(print);
  await Future.delayed(Duration(milliseconds: 1000));
  await subject.close(); //It will not allow me to close the subject
}

infite 是一个无限的数据源。它永远不会发出close 事件。那么，如何正确关闭subject，因为它永远不会完成接收数据？

编辑：Bad state: You cannot close the subject while items are being added from addStream in flutter 上提出的解决方案不起作用仅仅是因为drain 永远不会返回。

Answer 1

不要使用addStream。

您需要一种停止接收事件的方法，而addStream 方法不提供任何在流完成之前停止的方法，而永远不会。

如果你提前知道你想要多少个元素，你可以通过subject.addStream(infinite.take(1000)) 只获取前 1000 个事件。

如果不是，也许就像您的示例一样，您将其基于时间，那么我会选择以下内容：

var subscription = infinite.listen(subject.add);
subject.listen(print);
await Future.delayed(Duration(milliseconds: 100));
await subscription.cancel();
subject.close();

Stream.listen 返回的StreamSubscription 为您提供了一种停止再次收听的方法。或者暂停并稍后恢复事件，如果这就是您所需要的。

如果有其他内容告诉您何时停止，例如用户在其他地方单击按钮，您要么需要保持订阅可访问，以便稍后从知道它的代码中调用 cancel是时候停止了，或者在主题被取消时停止：

var subscription = infinite.listen(subject.add);
subject.onCancel = subscription.cancel;
subject.listen(print);