Sharing HTTP requests between multiple concurrent observers with RxJS

Question

I'm implementing functionality in an Angular app where multiple concurrent fetches for the same data should result in exactly one HTTP request, and the request is shared between all concurrent subscribers. The first subscriber should immediately send the request, and while that request is outstanding, new subscribers should subscribe to the same request. So similarly to this question, but without caching. I.e. once a request has completed, subsequent fetches will trigger a new HTTP request, allowing to fetch fresh data (e.g. if the cache has expired). Caching is managed elsewhere in the app.

My current implementation is this (simplified version):

class DataLoader<T> {

    private readonly requestInitiator = new Subject<void>();
    private readonly requests$: Observable<T>;

    constructor(requestCreator: () => Observable<T>) {
        this.requests$ = this.requestInitiator
            .pipe(
                mergeMap(requestCreator, 1), // concurrent = 1
                share(),
            );
    }

    load(): Observable<T> {
        return new Observable<T>(observer => {
            this.requests$.subscribe(observer);
            this.requestInitiator.next();
        });
    }
}

This loader can be used in an Angular singleton service like this:

@Injectable({ providedIn: 'root' })
export class SomeService {

    private readonly loader = new DataLoader<Something>(
        () => this.getHttp().get('/api/something'),
    );

    getSomething(): Observable<Something> {
        if (this.cache.isSomethingCached()) {
            return this.cache.getSomething();
        }

        return this.loader.load()
            .pipe(/* Cache if necessary */);
    }
}

If multiple components call SomeService.getSomething() simultaneously, they will all share the same HTTP request. This seems to work nicely. However, I don't really understand why. My main point of confusion is why mergeMap() works in this scenario. I initially implemented the loader using exhaustMap(), and it also worked most of the time, but in specific scenarios some observers ended up with an observable which didn't emit any values and didn't complete. I.e. effectively got stuck. Also, as far as I've understood, mergeMap() with a concurrency of 1 should work like concatMap(), i.e. in this case should cause the loader to execute 1 HTTP request for each new observer (since each observer sends a value to requestInitiator). But this does not seem to be the case here, new observers only cause a new request to be sent if there is no existing outstanding request. I.e. exactly what I want.

Apparently share() changes the behavior of the observable in a way I just don't understand. Can someone explain why mergeMap() with a concurrency on 1 works here, and why exhaustMap() does not?

Answer 1

If I understand the problem right, you have multiple Angular components which issue a request to a remote server to fetch some data. If a call is on-flight, you want subsequent calls not to hit the server but still whichever component has issued the subsequent call should be notified of the result produced by the call on-flight as soon as it completes with a result.

After completion of the first call, if a subsequent call to remote server is issued, then it should hit the server.

If my understanding is right, then I would suggest the following approach.

1. The remote service (i.e. the Class that implements it on the Angular side) should expose a method that executes the remote call and an Observable that emits the result of the remote call once this result received.
2. The remote service should have also a private boolean flag that is turned on before calling the remote service and is turned off once the response is received

The code of the service could look like this

class SomeService {
  private resultSubject = new Subject<any>();
  public resultObservableApi = this.resultSubject.asObservable();

  private callOnFlight = false;

  getSomething(input: any) {
    if (this.callOnFlight) {
      return;
    }
    this.callOnFlight = true
    remoteService(input).pipe(
      tap(() => this.callOnFlight = false)
    ).subscribe(res => this.resultSubject.next(res));
  }
}

This stackblitz simulates this scenario.

Coming to your specific question about the share operator, this operator under the hood introduces a Subject into the pipeline of transformations. Such subject is the one that notifies the values received from upstream. It is a concept similar to the one suggested above for our data service.

I can confirm that mergeMap() with a concurrency of 1 is the same of concatMap(). To be honest, I do not understand why mergeMap works while exaustMap sometimes does not work (actually I would have expected the contrary).