`MultiStreamController.onResume` is never called?

[sgrekhov]

Why `onResume callback in the code below in not called?

import "dart:async";
import "../../../Utils/expect.dart";

main() {
  asyncStart(2);
  var stream = Stream<int>.multi((controller) {
    Expect.isNull(controller.onResume);
    controller.onPause = () {
      print("onPause"); // prints
    };
    controller.onResume = () {
      print("onResume"); // is not printed
    };
    controller.add(1);
    controller.add(2);
    controller.add(3);
    controller.close();
  });
  listen(stream);
  listen(stream);
}

void listen(Stream<int> stream) {
  late StreamSubscription ss;
  int i = 0;
  ss = stream.listen((v) {
    Expect.equals(++i, v);
    if (i == 1) {
      ss.pause();
      Future.delayed(Duration(milliseconds: 100), () {
        ss.resume();
      });
    }
  }, onDone: asyncEnd);
}

The output is

unittest-suite-wait-for-done
onPause
onPause
unittest-suite-success

cc @lrhn

Dart SDK version: 3.7.0-27.0.dev (dev) (Tue Oct 15 17:02:51 2024 -0700) on "windows_x64"

[dart-github-bot]

Summary: The MultiStreamController.onResume callback is not being called when a paused StreamSubscription is resumed. This is unexpected behavior, as the documentation suggests it should be called.

[lrhn]

Same reason that isPaused stays false. All the events are synchronously enqueued during the onListen. When the subscription is paused and then resumed, it doesn't report the resume to the controller until it has delivered all enqueued events. Since that includeds the done event, which unsubscribes the subscription and calls onCancel, it's too late to report the resume after that.

So, in short, onResume is delayed until existing events have been delivered, and it gets overtaken by a cancel.

Shouldn't be unique to MultiStreamController, normal non-broadcast stream controllers should behave the same way.