Caching API Requests with `shareReplay`

Hello! Welcome back to our module on "Real-World Architectural Patterns."

In our last lesson, we built a resilient WebSocket stream that could automatically reconnect after failures using an exponential backoff strategy. We now have a robust source of data, but what if that data is expensive to fetch or needed in multiple places in our app? Constantly re-running the source stream for every component that needs the data is inefficient.

Today, we'll solve this problem. Our learning outcome is to build an in-memory cache for API requests with time-based invalidation using shareReplay.

We will construct a pattern that not only fetches data but also shares the result among all subscribers, preventing redundant API calls. We'll then enhance it to automatically refresh the data after a certain period and, crucially, explore how to do this without introducing memory leaks—a common pitfall.

1. The Basic Caching Problem and shareReplay

Imagine you have an Observable that makes an HTTP request. By default, this Observable is cold, meaning it executes its logic (the HTTP request) for every single subscriber. If three different components in your React app subscribe to this Observable, you'll make three identical API calls.

The shareReplay operator is designed to solve this. It transforms a cold Observable into a hot one by doing two things:

1. Sharing (Multicasting): It subscribes to the source Observable only once and broadcasts the emitted values to all of its own subscribers.
2. Replaying: It keeps a buffer of the last N values and immediately sends them to any new subscriber.

For caching API calls, the most common configuration is shareReplay(1), which caches and replays only the single most recent value (the API response).

Let's see this in action.

As an alternative or supplement to the video, the blog post it's based on covers the same concept.

Here's how you might structure this in a framework-agnostic data service class:

import { ajax } from 'rxjs/ajax';
import { map, shareReplay } from 'rxjs/operators';
import { Observable } from 'rxjs';

const API_URL = 'https://api.icndb.com/jokes/random/5?limitTo=[nerdy]';

class JokeService {
  private cache$: Observable<any[]> | null = null;

  get jokes$() {
    if (!this.cache$) {
      console.log('Cache not found. Creating new cache stream.');
      this.cache$ = ajax.getJSON<any>(API_URL).pipe(
        map(response => response.value),
        tap(() => console.log('Fetched new jokes from API')),
        shareReplay(1) // bufferSize: 1
      );
    }
    return this.cache$;
  }
}

const jokeService = new JokeService();

// First subscriber triggers the API call
console.log('First subscription');
jokeService.jokes$.subscribe(jokes => console.log('Subscriber A:', jokes.length, 'jokes'));

// Second subscriber gets the cached value without a new API call
setTimeout(() => {
  console.log('\nSecond subscription');
  jokeService.jokes$.subscribe(jokes => console.log('Subscriber B:', jokes.length, 'jokes'));
}, 2000);

2. Time-Based Invalidation: Keeping Data Fresh

Our current cache is great, but it has a flaw: the data is never updated. Once fetched, it stays in the cache forever. To solve this, we need to refresh the data periodically.

A common pattern for this is to use the timer operator to create a stream that "ticks" at a specified interval. We can then use a flattening operator like switchMap to trigger a new API request on each tick. The entire chain is then piped into shareReplay.

The data flow is:

1. timer(0, REFRESH_INTERVAL) emits immediately (0), and then every REFRESH_INTERVAL milliseconds.
2. switchMap receives the tick and triggers a new ajax request. It also cancels any pending previous request, which is good practice.
3. shareReplay receives the result from the ajax request and multicasts it to all current subscribers, while also caching it for future subscribers.

The updated service logic would look like this:

import { ajax } from 'rxjs/ajax';
import { map, shareReplay, switchMap, tap } from 'rxjs/operators';
import { Observable, timer } from 'rxjs';

const REFRESH_INTERVAL_MS = 10000; // 10 seconds

class JokeServiceWithRefresh {
  private cache$: Observable<any[]> | null = null;

  get jokes$() {
    if (!this.cache$) {
      // A timer that emits every 10 seconds
      const refreshTimer$ = timer(0, REFRESH_INTERVAL_MS);

      this.cache$ = refreshTimer$.pipe(
        tap(tick => console.log(`Timer ticked (tick ${tick}), triggering refresh...`)),
        // On each tick, switch to a new HTTP request
        switchMap(() => 
          ajax.getJSON<any>(API_URL).pipe(
            map(response => response.value)
          )
        ),
        tap(() => console.log('Fetched new jokes from API')),
        // Share the result of the whole timer->switchMap chain
        shareReplay(1)
      );
    }
    return this.cache$;
  }
}

This pattern is powerful, but it hides a significant danger.

3. The shareReplay Memory Leak and refCount

By default, shareReplay maintains its subscription to the source Observable (timer in our case) forever, even if all its own subscribers unsubscribe. Since timer never completes, this creates a permanent, active subscription in memory that is never garbage collected. This is a classic memory leak.

Every time your components unmount and resubscribe, the old shareReplay instance might be garbage collected, but the underlying timer it subscribed to could live on.

The solution is to configure shareReplay to be smarter. We can pass it a configuration object instead of just a buffer size. The key property is refCount.

shareReplay({ bufferSize: 1, refCount: true })

When refCount is true, shareReplay will:

• Subscribe to the source when the first subscriber arrives.
• Unsubscribe from the source when the last subscriber leaves.

This automatically tears down the entire stream, including the timer, when it's no longer needed, preventing the memory leak.

The Corrected, Robust Cache

Our final, robust caching implementation should look like this:

// ... imports

class RobustJokeService {
  private cache$: Observable<any[]> | null = null;

  get jokes$() {
    if (!this.cache$) {
      const refreshTimer$ = timer(0, REFRESH_INTERVAL_MS);

      this.cache$ = refreshTimer$.pipe(
        switchMap(() => 
          ajax.getJSON<any>(API_URL).pipe(map(response => response.value))
        ),
        shareReplay({
          bufferSize: 1,
          refCount: true // This is crucial!
        })
      );
    }
    return this.cache$;
  }
}

Deeper Dive: How refCount Works (Optional)

For a deeper understanding of the mechanics, the same video continues to build a custom shareReplay operator from scratch. This is a fantastic exercise to see how the subscriber counting and source unsubscription are managed internally. Given your background, you may find this particularly insightful.

4. Time-based Buffer vs. Time-based Refresh

It's important to distinguish our timer pattern from another shareReplay parameter: windowTime.

• shareReplay({ bufferSize: 1, refCount: true, windowTime: 10000 })

windowTime specifies how long a value should live in the buffer. After 10 seconds, the buffered value is discarded. A new subscriber arriving after this time will not receive the stale value and will trigger a new subscription to the source.

Our timer + switchMap pattern is different: it actively refreshes the source. It ensures that even existing subscribers get a fresh value every 10 seconds. For keeping data fresh, the timer pattern is generally more robust and explicit.

5. Manual Cache Invalidation

What if the user wants to force a refresh now? We can extend our service to allow for manual cache invalidation. A clean way to do this is to use a Subject to signal when the cache should be destroyed.

Here is the implementation of that pattern:

// ... imports
import { Subject, takeUntil } from 'rxjs';

class FullFeaturedJokeService {
  private cache$: Observable<any[]> | null = null;
  private reload$ = new Subject<void>();

  get jokes$() {
    if (!this.cache$) {
      const refreshTimer$ = timer(0, REFRESH_INTERVAL_MS);

      this.cache$ = refreshTimer$.pipe(
        switchMap(() => 
          ajax.getJSON<any>(API_URL).pipe(map(response => response.value))
        ),
        // This is the key addition:
        takeUntil(this.reload$),
        shareReplay({ bufferSize: 1, refCount: true })
      );
    }
    return this.cache$;
  }

  forceReload() {
    // Emit on the reload subject to complete the current cache stream
    this.reload$.next();
    
    // Set the cache to null so that the next subscription will create a new one
    this.cache$ = null;
  }
}

When forceReload() is called, reload$.next() causes takeUntil to complete the cache$ stream. We then nullify this.cache$. The very next component that subscribes to jokes$ will find the cache is null and rebuild the entire stream from scratch, effectively resetting the timer and fetching fresh data immediately.

---

Conclusion

Today we've constructed a sophisticated, production-ready caching mechanism using RxJS. This pattern is a cornerstone of efficient data management in reactive applications.

Key Takeaways:

• shareReplay is the fundamental operator for caching, turning cold, single-execution streams into hot, shared streams.
• For periodic data refresh, combine timer with switchMap before the shareReplay operator.
• To prevent memory leaks with non-completing sources, always use shareReplay({ refCount: true }). This ensures the source stream is torn down when no longer needed.
• The windowTime parameter invalidates the buffer, while the timer pattern actively refreshes the source.
• Manual invalidation can be implemented elegantly using a Subject paired with the takeUntil operator to complete and reset the cache stream.

Next Lesson Preview:

We've now seen shareReplay used to create a shared, multicasted stream from a cold source. We also briefly used a Subject for manual invalidation. Both are tools for sharing data, but they have fundamental differences in their creation and behavior. In our next lesson, we will analyze the differences between Subjects and shareReplay for creating shared, multicasted streams, clarifying when to use each tool.