当有人订阅时做些什么？答案

【问题标题】：Do something when someone subscribes?当有人订阅时做些什么？
【发布时间】：2019-01-28 13:40:05
【问题描述】：

我有一个BLE设备数据流的广播流。

为了从该流中获取数据，我需要先将数据发送到设备。

Stream<Data> dataStream() {
  sendDataRequestToDevice();
  return _broadcastController.stream;
}

问题在于一切都是异步的，这意味着当返回流时，从设备发送的实际事件很可能已经消失了。我正在寻找类似的东西：

Stream<Data> dataStream() {
  return _broadcastController.stream
    .doOnSubscribe(() => sendDataRequestToDevice()); // stolen from rxjava ;)
}

在不使用 RxDart 或类似的情况下，默认 Streaming 库中是否有类似的东西。（我只是不想为了这个目的使用它......）

【问题讨论】：

标签： asynchronous dart stream

【解决方案1】：

如果您关心有多少订阅者，您可能不应该使用广播流。广播流的基本思想是它广播而不知道（或关心）谁在收听。 onListen 和 onCancel 回调最初不是广播控制器的一部分，它们稍微破坏了模型。它们确实可以让您知道没有人何时在听，但仅此而已。

在这种情况下，我会制作自己的流来记录收听和取消。

class _ListenStream<T> extends Stream<T> {
  final Stream<T> _source;
  final void Function() _onListen; 
  final void Function() _onCancel;
  _ListenStream(this._source, this._onListen, this._onCancel);
  bool get isBroadcastStream => _source.isBroadcastStream;
  StreamSubscription<T> listen(void Function(T) onData, {
    Function onError, void Function() onDone, bool cancelOnError = false}) {
    if (_onListen != null) _onListen();
    return _ListenSubscription<T>(_source.listen(onData, 
        onError: onError, onDone: onDone, cancelOnError: cancelOnError),
        _onCancel);  
  }
}
class _ListenSubscription<T> extends StreamSubscription<T> {
  final _StreamSubscription<T> _source;
  final void Function() _onCancel;

  void onData(void handleData(T data)) { _source.onData(handleData); }
  void onError(Function handleError) { _source.onError(handleError); }
  void onDone(void handleDone()) { _source.onDone(handleDone); }

  void pause([Future resumeSignal]) { _source.pause(resumeSignal); }
  void resume() { _source.resume(); }
  Future<E> asFuture<E>([E defaultValue]) => _source.asFuture<E>(defaultValue);
  bool get isPaused => _source.isPaused;

  Future cancel() {
    var future = _source.cancel();
    if (_onCancel != null) future = future.whenComplete(_onCancel);
    return future;
  }
}

然后，您不提供原始流，而是为您的客户提供原始流的 _ListenStream 包装器，以及您希望在侦听和取消时调用的回调。

【讨论】：

【解决方案2】：

看来你在找onListen

https://api.dartlang.org/stable/2.1.0/dart-async/StreamController/onListen.html

_broadcastController.onListen = () {
  _broadcastController.add('foo');
};

您也可以将处理程序传递给构造函数https://api.dartlang.org/stable/2.1.0/dart-async/StreamController/StreamController.html

【讨论】：

不，这不是我要找的。一旦第一个订阅者订阅，构造函数上的 onListen 就会被调用一次。当订阅者订阅时，我需要一个总是被调用的东西，而不仅仅是一次。
每次收听流时都应该调用它。我很确定，因为我自己使用了几次，以立即将最后一个值传递给新的接收器。如果它不起作用，可能与broadcast有关。
我知道你在说什么。不，这不是我需要的。这些方法（onListen、onCancel）用于启动和关闭广播。这意味着，如果您在该广播中有 10 个订阅者，则调用一次 onListen 方法。当我按照您的建议进行操作时（刚刚尝试过），然后我将覆盖这些广播回调，这不是我想要的。每当有人订阅广播时，我只想要一个回调。（是的，它可能与广播相关，因为如果您不在广播中，这些方法可以免费使用）
我不认为还有别的东西。我建议您在gitter.im/ReactiveX/rxdart 中提问，或者只是发布此问题的链接。这个包的维护者非常擅长使用流。

【解决方案3】：

对于寻找解决方案的其他人：

我认为这不是一个好的解决方案，但它确实有效。请告诉我有更好的解决方案！

Stream<Data> dataStream() {
  return StreamGroup.merge([
        doOnSubscribe(() => sendDataRequestToDevice()),
        _broadcastController.stream
      ]);
}

Stream doOnSubscribe(void onSubscribe()) {
  StreamController controller;
  controller = StreamController(
    onListen: () {
      onSubscribe();
      controller.close();
    },
  );
  return controller.stream;
}

很遗憾，由于其他问题，它对我的 BLE 解决方案不起作用。

【讨论】：