RxJS cacheMap

cacheMap.ts

function cacheMap<T, Result>(
  cache$: Observable<{[key:string]: Observable<Result>}>,
  selector: (value:T) => string,
  projector: (key:string) => Observable<Result>) {
 	return function cmFunc(observable) {
    return observable.pipe(
      withLatestFrom(cache$),
      switchMap(([value, cache]) => {
        var key = selector(value);
        if (cache[key]) return cache[key];
        observable = projector(key).pipe(shareReplay());
        cache[key] = observable;
        return observable;
      })
    );
  };
}

Rx Vis demo.js

//use at https://rxviz.com/examples/custom
//there's some extra debug code in to see if the value came from the cache or not

const { interval, of, combineLatest, timer } = Rx;
const { map, switchMap, delay, shareReplay, share, groupBy, withLatestFrom } = RxOperators;
 

function cacheMap(cache$, selector, projector) {
 	return function cmFunc(observable) {
    return observable.pipe(
      withLatestFrom(cache$),
      switchMap(([value, cache]) => {
        var key = selector(value);
        if (cache[key]) return cache[key].pipe(map(x => 'c' + x));
        observable = projector(key).pipe(shareReplay());
        cache[key] = observable;
        return observable.pipe(map(x => 'p' + x));
      })
    );
  };
}

var cache$ = timer(0, 6500).pipe(map(() => ({})));
function fetch(key) {
  return of(key * 10);
}

timer(0, 1000).pipe(
	map(x => x % 3),
  cacheMap(cache$, x=>x, (key) => fetch(key)),
  groupBy(x => x.includes('c'))
)