bufferUntil

bufferUntil.dart

extension BufferUntilStreamExtension<T> on Stream<T> {
  Stream<T> bufferUntil(Stream<void> trigger) {
    final controller = StreamController<T>(sync: true);
    final bag = CompositeSubscription();

    // send source events to buffer controller.
    final bufferController = StreamController<T>(sync: true);
    listen(
      bufferController.add,
      onError: bufferController.addError,
      onDone: bufferController.close,
    ).addTo(bag);

    controller.onListen = () {
      // when trigger fires, send buffer controller events to output controller.
      final subscription =
          trigger.take(1).exhaustMap((_) => bufferController.stream).listen(
                controller.add,
                onError: controller.addError,
                onDone: controller.close,
              )..addTo(bag);

      if (!isBroadcast) {
        controller.onPause = subscription.pause;
        controller.onResume = subscription.resume;
      }
    };
    controller.onCancel = bag.cancel;

    return controller.stream;
  }
}