cancelOnInterrupt<T> function

Stream<T> cancelOnInterrupt<T>(Stream<T> source, Stream<void> interrupt)

Returns a stream that forwards events from source until interrupt emits, then cancels source and completes with StreamInterruptedException.

interrupt must be a broadcast stream if the caller wraps more than one source with the same interrupt stream.

Implementation

Stream<T> cancelOnInterrupt<T>(
  final Stream<T> source,
  final Stream<void> interrupt,
) {
  StreamSubscription<T>? sourceSubscription;
  StreamSubscription<void>? interruptSubscription;
  final controller = StreamController<T>();
  var interrupted = false;

  controller.onListen = () {
    sourceSubscription = source.listen(
      controller.add,
      onError: controller.addError,
      onDone: () {
        if (!interrupted && !controller.isClosed) {
          controller.close();
        }
      },
    );

    interruptSubscription = interrupt.listen((_) async {
      if (interrupted) {
        return;
      }
      interrupted = true;
      await sourceSubscription?.cancel();
      if (!controller.isClosed) {
        controller.addError(StreamInterruptedException());
        await controller.close();
      }
    });
  };

  controller.onCancel = () async {
    await sourceSubscription?.cancel();
    await interruptSubscription?.cancel();
  };

  return controller.stream;
}