combineLatest2<A, B, R> static method

Stream<R> combineLatest2<A, B, R>(Stream<A> streamA, Stream<B> streamB, R combiner(A a, B b))

Implementation

static Stream<R> combineLatest2<A, B, R>(
  Stream<A> streamA,
  Stream<B> streamB,
  R Function(A a, B b) combiner,
) {
  late StreamController<R> controller;
  StreamSubscription<A>? subA;
  StreamSubscription<B>? subB;
  A? latestA;
  B? latestB;
  var hasA = false;
  var hasB = false;
  var doneA = false;
  var doneB = false;

  void emitIfReady() {
    if (hasA && hasB) controller.add(combiner(latestA as A, latestB as B));
  }

  void closeIfDone() {
    if (doneA && doneB) controller.close();
  }

  controller = StreamController<R>(
    sync: true,
    onListen: () {
      subA = streamA.listen(
        (event) {
          latestA = event;
          hasA = true;
          emitIfReady();
        },
        onError: controller.addError,
        onDone: () {
          doneA = true;
          closeIfDone();
        },
      );
      subB = streamB.listen(
        (event) {
          latestB = event;
          hasB = true;
          emitIfReady();
        },
        onError: controller.addError,
        onDone: () {
          doneB = true;
          closeIfDone();
        },
      );
    },
    onPause: () {
      subA?.pause();
      subB?.pause();
    },
    onResume: () {
      subA?.resume();
      subB?.resume();
    },
    onCancel: () async {
      await subA?.cancel();
      await subB?.cancel();
    },
  );

  return controller.stream;
}