CLI Dart：onPause、onResume、onDone 未按预期启动

Question

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

1. 我想知道为什么分配给流订阅的“onDone”处理程序没有启动。
2. 我想知道为什么“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!

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

• 已创建流控制器。
• 3 个事件被注入到控制器提供的流中。
• 监听器订阅控制器提供的流。
• 我们将“onDone”处理程序分配给侦听器订阅。
• 在流​​侦听器 (OnValueHandlerContainer::onValue) 中，我们暂停和恢复订阅。

流监听器按预期启动了 3 次。

但是：

• 永远不会执行“onDone”处理程序。我希望它在脚本执行结束时执行，同时控制器被销毁（因此，订阅被关闭）。
• “onPause”和“onResume”处理程序只启动一次。我预计它们会被执行 3 次。

有什么想法吗？

Answer 1

您没有收到“完成”事件的原因是您从未close 订阅流。

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

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

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

控制器报告工作已恢复，但由于没有人添加更多事件，也没有人调用close，因此不会再发生任何事情。