One observable, multiple Subscribers Rxjs/Angular share not working

Question

I a using angular 4 for one of my projects and i am making a http call to get the data. I cache the data when the data is returned from the backend so that for the subsequent requests i dont have to make the rest calls again and again.

getApplication(): Observable<Array<Application>> {
        return new Observable<Array<Application>>((observer) => {
            if (this.app&& this.app.length > 0) {
                observer.next(this.app)
            } else {
                this._http
                    .get<Application[]>(this.url)
                    .map((app) => this.mapToApp(app))
                    .subscribe((result: Array<Application>) => {
                        observer.next(result);
                    }, () => {
                        observer.next([]);
                    })
            }
        })
    }

This works perfectly fine. But if two calls go at the same time, i see there are two requests going in the network tab instead of one. So i tried share() to make satisfy 'One observable, multiple subscribers'.

getApplication(): Observable<Array<Application>> {
            return new Observable<Array<Application>>((observer) => {
                if (this.app&& this.app.length > 0) {
                    observer.next(this.app)
                } else {
                    this._http
                        .get<Application[]>(this.url)
                        .map((app) => this.mapToApp(app))
                        .subscribe((result: Array<Application>) => {
                            observer.next(result);
                        }, () => {
                            observer.next([]);
                        })
                }
            }).share()
        }

But this also, i see there are multiple requests going in the network tab instead of one. What has to be done to make only one call and share the result with all the subscribers ? Please help.

Hi I'd use a Behavior Subject which publishes its changes when next is called: github.com/Reactive-Extensions/RxJS/blob/master/doc/api/… — t3__rry
Share sounds so simple doesn't it! It uses Subject internally which only multicasts values to an observer if they were already subscribed at the time. So if something subscribes, takes a value and then something else subscribes it won't get that same value. Essentially share allows the first one to steal! ShareReplay(1) fixes this. I've found share most useful when you have a connectable observable that doesn't start until you've made sure all the subscribers have subscribed - then you can share all you want in that case. — Simon_Weaver

Picci Picci · Accepted Answer · 2018-05-07T20:51:25

My first suggestion is to consider a bit of simplification in the code along these lines

getApplication(): Observable<Array<Application>> {
      if (this.app && this.app.length > 0) {
         return Observable.of(this.app);
      }
      return this._http
                    .get<Application[]>(this.url)
                    .map((app) => this.mapToApp(app))
                    .do((result: Array<Application>) => {
                        this.app = result;
                    }
}

With this logic you can cache using any logic you want. You have decided to check if this.app is not null and if it has some elements. But you could use any other logic, maybe dependent on the input parameters.

A different story is the problem of 2 calls issued at the same time. In this case you probably want both to share the Observable so that just one subscription is shared and replay the last result recorded so that you can return a cached value.

One way of doing this could be the following

private cache$: Observable<Array<Application>>;

requestApplication(): Observable<Array<Application>> {
          return this._http
                        .get<Application[]>(this.url)
                        .map((app) => this.mapToApp(app))
                        .do((result: Array<Application>) => {
                            this.app = result;
                        }
}

getApplication() {
    if(!cache$) {
       cache$ = this.requestApplication().shareReplay(1);
    }
    return cache$;
}

The trick here is performed by the shareReplay operator. The share part of it makes sure that the subscription is shared. The replay part of it, with a buffer size of 1, makes sure that the last result stored is always returned.

Then you may face the problem of clearing the cache once in a while. But here the story gets more complex and I suggest you to look at this very good article https://blog.thoughtram.io/angular/2018/03/05/advanced-caching-with-rxjs.html

One observable, multiple Subscribers Rxjs/Angular share not working

2 Answers