[Help needed] Broadcast Streams Cache Last Result about async HOT 5 CLOSED

import 'dart:async';

class _ReplayLatestStream<T> extends StreamView<T> {
  T _latest;

  _ReplayLatestStream(this._latest, Stream<T> stream) : super(stream) {
    if (!stream.isBroadcast) {
      throw ArgumentError.value(stream, "stream", "Must be a broadcast stream");
    }
    stream.listen((e) {
      _latest = e;
    });
  }

  @override
  StreamSubscription<T> listen(void Function(T) onData,
      {Function onError, void Function() onDone, bool cancelOnError}) {
    Future(() {
      onData(_latest);
    });
    return super.listen(onData,
        onError: onError, onDone: onDone, cancelOnError: cancelOnError);
  }
}

extension ReplayLatest<T> on Stream<T> {
  Stream<T> replayLatest(T startWith) =>
      _ReplayLatestStream(startWith, isBroadcast ? this : asBroadcastStream());
}

var sub = stream.listen(null);
sub.onData((v) { .... });

Stream<T> repeatLast<T extends Object>(Stream<T> source) {
  var listeners = <StreamControlller>{};
  T? last = null;
  bool done = false;
  source.listen((event) {
     last = event;
     for (var listener in [...listeners]) listener.add(event);
   }, onError: (Object e, StackTrace s) {
     for (var listener in [...listeners]) listener.addError(e, s);
   }, onDone: () {
      done = true;
      last = null;
      for (var listener in listeners) listener.close();
   });
  return Stream.multi((StreamController c)  {
    if (done) {
      c.close();
    } else {
      var lastEvent = last;
      if (lastEvent != null) c.add(lastEvent);
      listeners.add(c);
    }
    c.onCancel = () {
      listeners.remove(c);
    };
  });
}

// Copyright 2020 Google LLC.
// SPDX-License-Identifier: BSD-3-Clause
Stream<T> repeatLast<T>(Stream<T> source) {
  List<MultiStreamController<T>> controllers = [];
  void forEachController(void action(MultiStreamController)) {
    var i = 0, j = 0;
    while (i < controllers.length) {
      var controller = controllers[i++];
      if (controller.hasListener) {
        action(controller);
        if (controller.hasListener) {
          controllers[j++] = controller;
        }
      }
    }
    if (j < controllers.length) controllers.length = j;
  }

  bool done = false; // Set to true when [source] closes.
  /*late*/ T mostRecent; // Most recent event value.
  bool hasMostRecent = false; // Whether mostRecent is available.

  StreamSubscription<T> /*?*/ sub;
  void ensureSubscribed() {
    sub ??= source.listen((value) {
      mostRecent = value;
      hasMostRecent = true;
      forEachController((c) {
        c.addSync(value);
      });
    }, onError: (e, s) {
      forEachController((c) {
        c.addErrorSync(e, s);
      });
    }, onDone: () {
      done = true;
      forEachController((c) {
        c.closeSync();
      });
    });
  }

  // Subscribe early if it's a broadcast stream.
  if (source.isBroadcast) ensureSubscribed();
  return Stream<T>.multi((controller) {
    if (hasMostRecent) controller.add(mostRecent /*as T*/);
    if (done) {
      controller.close();
      return;
    }
    controllers.add(controller);
    ensureSubscribed();
  }, isBroadcast: source.isBroadcast);
}