Queuing observable requests

Question

I have a service which, after some validation, makes a call to my backend server

get(systems?: string[], options?: {}): Observable<System[]> {
    if (
        // cache check
    ) {
        // validate options
        return this.api.get('/systems', options)
            .map(response => {
                // proccess results, store to cache
            });
    } else {
        return Observable.of(this.systems);
    }
}

This request is called when the service is initialized, to get basic data (passing the options {basic: true}. On one of my pages, the same service is called on component initialization, without the basic option. However, when the site is loaded to that page, of course both HTTP requests fire, as neither has returned and populated the cache.

I can't seem to figure out how to best avoid this situation. In an ideal solution, the requests would queue, and I would use the cache logic to determine if another HTTP request was necessary or I could return cached data. However, that logic doesn't seem to fit the observable/observer pattern. I'd love some advice on how to address this.

Answer 1

In order to avoid race conditions, you need to cache Observables instead of caching values directly.

The issue with caching values directly is that you could have multiple calls made to your service back-to-back, and your service will check the cache for values that will not exist until your request completes (this creates the race condition -- where your request is racing against the logic that checks the cache). Your service will end up making multiple requests and overriding the cache of the same value multiple times.

By caching Observables, you place them in a cache immediately after they are created, and you return that Observable when any subscriber needs it (regardless of whether or not that Observable has resolved to a value), so you won't have multiple requests being made.

To make things easier, we can cache ReplaySubject objects, which will retain the last value that was called using next and returns that value automatically when new subscribers subscribe. This is essentially a Subject that caches previous values and returns them to new subscribers.

See this simple example, where cache is an object representing a cache of ReplaySubjects:

let cache = {};

getCached('key1').subscribe(val => console.log("Sub1: ", val)); // first subscriber
getCached('key1').subscribe(val => console.log("Sub2: ", val)); // second subscriber immediately after first

function getCached(key) {

  if(cache[key]) return cache[key]; // does cached Observable exist? If so, return it right away
  
  console.log("Not cached. Fetching new value.");
  
  cache[key] = new Rx.ReplaySubject(1); // set cache equal to new ReplaySubject
  
  setTimeout(() => cache[key].next(500), 1000); // fetch new value and alert our ReplaySubject subscribers
  
  return cache[key]; // return ReplaySubject immediately
}

<script src="https://unpkg.com/@reactivex/rxjs@5.0.0-rc.4/dist/global/Rx.js"></script>

The key thing to note about the example above is that getCached('key1') is called back-to-back before the cached value has had the chance to call next (which is delayed by 500 milliseconds) and the ReplaySubject is cached only once (notice the console.log only prints once, the first time that it sets the cache).

---

Caching Options and Systems

Because you need a new cache entry for every combination of systems and options, you can set your key equal to the string representation of systems + options by using JSON.stringify on options:

let cache = {};

getCached('key1', {basic: true}).subscribe(val => console.log("Sub1: ", val)); // first subscriber
getCached('key1', {basic: true}).subscribe(val => console.log("Sub2: ", val)); // second subscriber immediately after first

getCached('key1', {basic: false}).subscribe(val => console.log("Sub3: ", val));

function getCached(systems, options) {

  const key = systems + JSON.stringify(options);
  
  console.log("key is: ", key);

  if(cache[key]) return cache[key]; // does cached Observable exist? If so, return it right away
  
  console.log("Not cached. Fetching new value.");
  
  cache[key] = new Rx.ReplaySubject(1); // set cache equal to new ReplaySubject
  
  setTimeout(() => cache[key].next(500), 1000); // fetch new value and alert our ReplaySubject subscribers
  
  return cache[key]; // return ReplaySubject immediately
}

<script src="https://unpkg.com/@reactivex/rxjs@5.0.0-rc.4/dist/global/Rx.js"></script>

Notice that the cache is updated only when a new combination of systems and options is used. In the example above, the unique keys are the following strings:

• 'key1{"basic":true}'
• 'key1{"basic":false}'