如何缓冲流事件？

Question

我有一个订阅流的Web组件.

由于Web组件每次显示时都会重新创建,因此我必须清理订阅者并重做它.

现在我将所有订阅者添加到列表中,并在removed()生命周期方法中我正在做:

subscriptions.forEach((sub) => sub.cancel());

现在,问题是:当没有显示Web组件时,没有人监听流.问题是组件在未显示时缺少数据/事件.

我需要的是缓冲.在注册监听器时,需要缓冲事件并立即发送事件.根据文档,缓冲发生直到监听器注册:

控制器将缓冲所有传入事件,直到订户注册.

这样可行,但问题是侦听器会在某些时候删除并重新注册,并且看起来这不会触发缓冲.

看起来缓冲只是最初发生,即使所有听众都不在,也不会在以后发生.

所以问题是:在这种听众可能会消失的情况下,如何缓冲？

Answer 1

注意:通常您不能重新订阅已关闭的Stream.这似乎是我们忘记解决的错误.

我不熟悉网络组件,但我希望我通过以下建议解决您的问题.

一种方式(当然也有许多方法)是为每个订阅者(如html-events)创建一个暂停原始流的新流.

Say origin是最初的Stream.然后实现一个streamgetter,它返回一个链接到的新Stream origin:

未经测试的代码.

Stream origin;
var _subscription;
final _listeners = new Set<StreamController>();

_addListener(controller) {
  _listeners.add(controller);
  if (_subscription == null) {
    _subscription = origin.listen((event) {
      // When we emit the event we want listeners to be able to unsubscribe
      // or add new listeners. In order to avoid ConcurrentModificationErrors
      // we need to make sure that the _listeners set is not modified while
      // we are iterating over it with forEach. Here we just create a copy with
      // toList().
      // Alternatively (more efficient) we could also queue subscription
      // modification requests and do them after the forEach.
      _listeners.toList().forEach((c) => c.add(event));
    });
  }
  _subscription.resume();  // Just in case it was paused.
}
_removeListener(controller) {
  _listeners.remove(controller);
  if (_listeners.isEmpty) _subscription.pause();
}

Stream get stream {
  var controller;
  controller = new StreamController(
      onListen: () => _addListener(controller),
      onCancel: () => _removeListener(controller));
  return controller.stream;
}

如果您需要立即缓冲事件,则需要立即开始订阅,而不是像示例代码那样懒惰.