watchRunChanges method

Implementation

@override
Stream<void> watchRunChanges(String runId) => Stream<void>.multi((listener) {
  final controller = _runChangeControllers.putIfAbsent(
    runId,
    StreamController<void>.broadcast,
  );
  final subscription = controller.stream.listen(
    listener.add,
    onError: listener.addError,
    onDone: () => unawaited(listener.close()),
  );
  listener.onCancel = () async {
    await subscription.cancel();
    if (!controller.hasListener &&
        identical(_runChangeControllers[runId], controller)) {
      _runChangeControllers.remove(runId);
      await controller.close();
    }
  };
});