onValueChange method

Stream<ValueUpdate> onValueChange(String path, [ int qos = 0 ])

Implementation

Stream<ValueUpdate> onValueChange(String path, [int qos = 0]) {
  ReqSubscribeListener? listener;
  late StreamController<ValueUpdate> controller;
  var subs = 0;
  controller = StreamController<ValueUpdate>.broadcast(
    onListen: () {
      subs++;
      listener ??= subscribe(path, (ValueUpdate update) {
        controller.add(update);
      }, qos);
    },
    onCancel: () {
      subs--;
      if (subs == 0) {
        listener?.cancel();
        listener = null;
      }
    },
  );
  return controller.stream;
}