RxJS cache

cache.js

const Rx = require('rxjs/Rx');

const PERFORM = 1;
const INVALIDATE_CACHE = 2;

function asyncOp() {
  return Promise.resolve(Date().toString());
}

const clearCacheInterval$ = Rx.Observable.interval(5000);
const interval$ = Rx.Observable.interval(1000);

function wrap(value) {
  return { value };
}

function unwrap({ value }) {
  return value;
}

interval$
  .mapTo(PERFORM)
  .merge(clearCacheInterval$.mapTo(INVALIDATE_CACHE))
  .mergeScan((acc, value) => {
    if (value === INVALIDATE_CACHE) {
      return Rx.Observable.of(undefined);
    }

    if (acc !== undefined) {
      return Rx.Observable.of(acc);
    }

    const promise = asyncOp();
    return Rx.Observable.from(promise).map(wrap);
  }, undefined)
  .filter(omission => omission !== undefined)
  .map(unwrap)
  .subscribe(console.log);

window.js

const Rx = require('rxjs/Rx');
const process = require('process');
const stdin = process.stdin;

// ceremony to have keypress events in node

stdin.setRawMode(true);
stdin.setEncoding('utf8');
stdin.resume();

// split the keypress observable into ctrl-c and c observables.

const keyPressed$ = Rx.Observable.fromEvent(stdin, 'data').share();
const ctrlCPressed$ = keyPressed$.filter(code => code === '\u0003');
const cPressed$ = keyPressed$.filter(code => code === 'c');

ctrlCPressed$.subscribe(() => process.exit());

function asyncOp() {
  return Promise.resolve(Date().toString());
}

const invalidateCache$ = Rx.Observable.interval(5000).merge(cPressed$);
const interval$ = Rx.Observable.interval(1000);

interval$
  .windowWhen(() => invalidateCache$)
  .map(win => win.mergeScan((acc, value) => {
    if (acc === undefined) {
      const promise = asyncOp();
      return Rx.Observable.from(promise);
    }

    return Rx.Observable.of(acc);
  }, undefined))
  .mergeAll()
  .subscribe(console.log);