Better way to cache an expensive Observable source?

[thomasnield]

I have this code below that demonstrates a caching replayable Observable that updates every 2 seconds. It also will shut down when there are no subscribers, but will rebuild once there are new subscribers.

Is there a better way to do this? Or is there an existing operator I'm not using?

import {
  of,
  interval,
  shareReplay,
  tap,
  toArray,
  switchMap,
  take,
  delay,
  mergeMap,
} from 'rxjs';

// refresh cache every 2 seconds 
const source = interval(2000)
  .pipe(
    switchMap(_ => of('A', 'B', 'C', 'D', 'E').pipe(
          toArray(),
          tap((_) => console.log('Data refreshed'))
        )
    )
  )
  .pipe(shareReplay({ bufferSize: 1, refCount: true }), take(1));

// These subscribers should use the same cache
source.subscribe((x) => console.log(`SUBSCRIBER 1: ${x}`));

of('Start')
  .pipe(
    delay(500),
    mergeMap((_) => source)
  )
  .subscribe((x) => console.log(`SUBSCRIBER 2: ${x}`));


  of('Start')
  .pipe(
    delay(1500),
    mergeMap((_) => source)
  )
  .subscribe((x) => console.log(`SUBSCRIBER 3: ${x}`));

// These subscribers should trigger a rebuild and reuse of the next cache

of('Start')
  .pipe(
    delay(5000),
    mergeMap((_) => source)
  )
  .subscribe((x) => console.log(`SUBSCRIBER 4: ${x}`));

  
of('Start')
.pipe(
  delay(6500),
  mergeMap((_) => source)
)
.subscribe((x) => console.log(`SUBSCRIBER 5: ${x}`));

[alexanderharding]

Sure, it looks like you want a cache that implements the stale-while-revalidate pattern but only while there are subscribers.

// The resource you want to cache doesn't totally matter as long as it eventually completes
// so the repeat operator can kick in. This is the reason I've broken it out into a separate 
// constant for this example.
const resource = of('A', 'B', 'C', 'D', 'E').pipe(toArray());

const cached = resource.pipe(
     // This is the actual refresh mechanism. It supports quite a few different arguments to meet
     // your needs. In this case I am keeping it in relation to your example. 
     // 
     // It's worth noting that the delay will not start until the source observable (resource) completes. This is equivalent
     // to your example because resource is synchronous but may need to be adjusted to account for
     // asynchronous source observables. 
     //
     // I like this operator because in this example it starts the delay at the moment that the value from the source is actually cached.
     repeat({ delay: 2000 }),

     // This is the actual caching mechanism. As in your example, it will replay the cached value to new subscribers
     // but in this one it will also push refreshed values as they replace the previously cached value.
     shareReplay({ bufferSize: 1, refCount: true }),

     // No need to add a take(1) here because we want subscribers to continue to receive values 
     // when the cache is refreshed.
);

I have more examples of other caching strategies that leverage other operators if you'd like to see.

[thomasnield]

Clever! And exactly what I’m looking for. I have to rely on caching patterns a lot so if you have other handy ones to share I’d be much obliged. I’m getting back into Rx after a hiatus from RxJava since 2019, so I’m happy to see this library is going strong as I get into full stack for a project. Thank you!

[alexanderharding]

@thomasnield Absolutely! I have a handful to share but I'll start with a real world problem I had to solve recently. shareReplay cache invalidation

// This could change for many reasons (e.g. on place order);
const cartId$$ = new BehaviorSubject('123456');

const noCart = Symbol(`Flag indicating that there's no cart (likely because it's in the process of loading)`);
const cartService = new CartService();

const cart$: Observable<ICart> = cartId$$.pipe(
     // Note two things:
     // 1. We start with 'noCart' to emulate a 'shareReplay' cache bust.  In reality, it's still caching 'noCart', but we don't let it get past the 'filter(...)' below.
     // 2. The 'startWith' has to be attached to the inner Observable so that it emits every time the 'cartId' emits a new value.
     switchMap((id) => cartService.get(id).pipe(startWith(noCart))),
     // The actual caching mechanism.
     shareReplay(1),
     // This ensures that new subscribers don't get a stale cached cart while a new cart is loading.
     filter((cart) => cart !== noCart),
);

We had problems in our ecommerce site where the customer would do something to generate a new cart (place order, logout, ect.) and a side effect of that was code immediately subscribing to get the new cart. However, the new cart wasn't available yet, and that logic was acting on an old now stale cart.