events - CLI Dart : onPause, onResume，onDone 未按预期启动

Question

我正在试验 Dart，我无法解释两个观察结果。

我想知道为什么分配给流订阅的“onDone”处理程序没有启动。

我想知道为什么“onPause”和“onResume”处理程序只启动一次。

编码:

import 'dart:async';
import 'dart:io';

/// This class encapsulates all the necessary data used by the "onValue" event
/// handler (the construct avoids using global variables).
class OnValueHandlerContainer {
  static StreamSubscription<int> _streamSubscriber;

  static setStreamSubscriber(StreamSubscription<int> stream) {
    _streamSubscriber = stream;
  }

  // This static method is the handler executed when a event is received through
  // the stream.
  //
  // WARNING: you have absolutely no idea when this handler will be executed.
  // Do not assume that it will be executed right after the execution of the code
  // that emits an event. It may be executed several lines (of codes) below the
  // line that emits the event. It may well be executed after the end of the
  // script.
  static void onValue(int value) {
    // At this point: the state of the subscription is (inevitably) "active".
    print("onValue: An event has been raised. The associated value is ${value}!");
    print("         Pause the subscription. Wait for 1 second. Resume the subscription");

    // Note 1: once a Dart function starts executing, it continues executing until
    //         it exits. When managing interrupts in C, it is necessary to protect
    //         interrupt handlers from being interrupted. This is not the case in
    //         Dart : a function (and, thus, an event handler) cannot be interrupted
    //         by the occurrence of another event.
    //         => The code below has no sense, other than experimentation.
    // Note 2: while paused, the subscription will not fire any events. If it receives
    //         events from its source, they will be buffered until the subscription
    //         is resumed.
    _streamSubscriber.pause();
    sleep(Duration(seconds: 1));
    _streamSubscriber.resume();

    // At this point: the state of the subscription is "active".
  }
}

main() async {

  // Create a controller.
  // A StreamController gives you a new stream and a way to add events to the stream
  // at any point, and from anywhere. The stream has all the logic necessary to handle
  // listeners and pausing. You return the stream and keep the controller to yourself.
  StreamController<int> sc = StreamController<int>(
      onListen: () => print("Controller: the stream has been assigned a listener!"),
      onCancel: () => print("Controller: the stream has been canceled!"),
      // As you may notice, the event handlers are not executed every time the
      // subscription gets paused or resumed.
      //
      // This behaviour comes from these facts:
      // - Dart is single-threaded.
      // - An event handler cannot be interrupted: once a Dart function starts
      //   executing, it continues executing until it exits. In other words, Dart
      //   functions can’t be interrupted by other Dart code.
      //   See https://webdev.dartlang.org/articles/performance/event-loop
      // - A stream is a FIFO.
      onPause:  () => print("Controller: the stream has been paused!"),
      onResume: () => print("Controller: the stream has been resumed!")
  );

  // Get the stream created by the stream controller.
  // Right now, this stream has no assigned listener.
  Stream<int> stream = sc.stream;
  print("Does the stream provided by the controller have a listener ? ${sc.hasListener ? 'yes' : 'no'} - the answer should be no.");

  // Push values into the stream controlled by the stream controller.
  // Because no listener subscribed to the stream, these values are just stored
  // into the stream.
  for(int i=0; i<3; i++) {
    print("Send the value ${i} into the stream.");
    sc.add(i);
  }

  // Add a listener to the stream.
  // Now the stream has an assigned listener.
  StreamSubscription<int> subscriber = stream.listen(OnValueHandlerContainer.onValue);
  OnValueHandlerContainer.setStreamSubscriber(subscriber);
  subscriber.onDone(() => print("The subscription is done!"));
  print("Does the stream provided by the controller have a listener ? ${sc.hasListener ? 'yes' : 'no'} - the answer should be yes.");

  // Wait for 10 seconds.
  print("Start waiting for 10 seconds");
  Future.delayed(Duration(seconds: 10)).then((var v) => print("10 seconds ellapsed!"));
  print("End of script");
}

结果:

Does the stream provided by the controller have a listener ? no - the answer should be no.
Send the value 0 into the stream.
Send the value 1 into the stream.
Send the value 2 into the stream.
Controller: the stream has been assigned a listener!
Does the stream provided by the controller have a listener ? yes - the answer should be yes.
Start waiting for 10 seconds
End of script
onValue: An event has been raised. The associated value is 0!
         Pause the subscription. Wait for 1 second. Resume the subscription
Controller: the stream has been paused!
onValue: An event has been raised. The associated value is 1!
         Pause the subscription. Wait for 1 second. Resume the subscription
onValue: An event has been raised. The associated value is 2!
         Pause the subscription. Wait for 1 second. Resume the subscription
Controller: the stream has been resumed!
10 seconds ellapsed!

基本上，提供的代码执行以下操作:

创建一个流 Controller 。

3 个事件被注入(inject)到 Controller 提供的流中。

监听器订阅 Controller 提供的流。

我们为监听器订阅分配一个“onDone”处理程序。

在流监听器 (OnValueHandlerContainer::onValue) 中，我们暂停和恢复订阅。

正如预期的那样，流监听器启动了 3 次。

然而:

“onDone ”处理程序永远不会执行。我希望它在脚本执行结束时执行，而 Controller 正在被销毁(因此，订阅被关闭)。

“onPause ”和“onResume ”处理程序只启动一次。我希望他们被执行 3 次。

任何的想法 ？

Answer 1

你没有得到“完成”事件的原因是你从来没有close流订阅。

您没有收到更多“暂停”事件的原因是流订阅很聪明。

您要做的第一件事是在任何人收听流之前添加很多事件。你不应该在实际代码中这样做，而是只在 onListen 时才开始添加事件。被调用，并在 onPause 时再次停止被调用，直到订阅恢复。

在这里，流订阅被许多事件填满，然后它传递一个事件，然后订阅被暂停。
订阅会尽职尽责地将其报告给 Controller 。
然后订阅会得到一份简历。这就是它变得聪明的地方。由于它已经有要传递的事件，它不会将简历报告回 Controller 。它现在实际上并不想要更多的事件，有很多要交付的。因此，它以一秒的间隔传递缓冲的事件，直到缓冲区为空。此时，它将恢复报告给 Controller 。

Controller 报告工作已恢复，但由于没有人添加更多事件，也没有人调用 close , 不会有任何进一步的事情发生。