class DumbCache {

  cache = {};

​

  get(key) {

    return new Promise((resolve, reject) => {

      var value = this.cache[key];

      resolve(value);

    });

  }

​

  set(key, value) {

    return new Promise((resolve, reject) => {

      this.cache[key] = value;

      resolve();

    });

  }

}

​

class CacheValueObservable {

  private cache: DumbCache;

​

  constructor(private key: string) {

    this.cache = new DumbCache();

  }

​

  /*

   * meta observer to handle promises from observables with all results and errors

   * thanks to ReplaySubject(1) current value is available immediately after subscribing 

   */

  private _valueSource$$ = new Rx.ReplaySubject(1);

  private _value$ = this._valueSource$$.mergeAll(1);

​

  public value$() { return this._value$; }

​

  public updateValue(value) {

    this._valueSource$$.next(

      // encapsulate cache promise (with both result and errors) to observable

      Rx.Observable.defer(() =>

        this.cache.set(this.key, value)

        .then(() => this.cache.get(this.key))

      )

    );

  }

​

  complete() {

    this._valueSource$$.complete();

  }

}

​

console.log('starting...');

​

let cacheValueObservable = new CacheValueObservable("TEST_KEY");

​

cacheValueObservable.updateValue('VALUE 0');

​

cacheValueObservable.value$().subscribe(

    val => {

      console.log('val:' + val);

    },

    val => console.log('err', val.stack),

    () => (console.log('complete'))

);

​

cacheValueObservable.updateValue('VALUE 1');

cacheValueObservable.updateValue('VALUE 2');

​

console.log('completing...');

​

//cacheValueObservable.complete();

​

console.log('end');

​

​

​