observeStream<T> function

Stream<T> observeStream<T>(
  1. T read(), {
  2. bool emitInitial = true,
})

Creates a single-subscription stream of values derived by read.

Implementation

Stream<T> observeStream<T>(T Function() read, {bool emitInitial = true}) {
  ObservationSubscription<T>? subscription;
  late final StreamController<T> controller;
  controller = StreamController<T>(
    onListen: () {
      try {
        subscription = observe(
          read,
          onChange: controller.add,
          onError: (error, stackTrace) {
            controller.addError(error, stackTrace);
            controller.close();
          },
          fireImmediately: emitInitial,
        );
      } catch (error, stackTrace) {
        controller.addError(error, stackTrace);
        controller.close();
      }
    },
    onCancel: () {
      subscription?.dispose();
      subscription = null;
    },
  );
  return controller.stream;
}