AsyncObservable.hx

package tink.state.internal;

class AsyncObservable<T> implements ObservableObject<Promised<T>> extends Dispatcher implements Observer {

  final source:ObservableObject<Promise<T>>;

  var promise:Promise<T> = null;
  var hot = false;
  var link:CallbackLink = null;
  public function new(source:ObservableObject<Promise<T>> #if tink_state.debug , ?toString, ?pos #end) {
    super(
      active ->
        if (hot = active) source.subscribe(this)
        else {
          link.cancel();
          source.unsubscribe(this);
        }
      #if tink_state.debug , toString, pos #end
    );
    this.source = source;
  }

  public function getValue():Promised<T> {
    if (promise == null)
      promise = source.getValue();
    if (hot)
      link = promise.handle(_ -> trigger());

    return switch promise.status {
      case Ready(l) if (l.computed):
        switch l.get() {
          case Success(v): Done(v);
          case Failure(e): Failed(e);
        }
      default: Loading;
    }
  }

  public function isValid():Bool
    return promise != null;

  public function getComparator():Comparator<Promised<T>>
    return null;

  #if tink_state.debug
  public function getDependencies():Iterator<Observable<Any>>
    return [(cast source:Observable<Any>)].iterator();
  #end

  inline function trigger()
    if (inline isValid()) {
      promise = null;
      fire(this);
    }

  public function notify<R>(from:ObservableObject<R>) {
    link.cancel();
    trigger();
  }
}