I want to know if it is necessary to unsubscribe from Observables made with forkJoin([..]) using a set of HttpClient Observables piped by shareReplay(..) operator?

I read about forkJoin(..) that it unsubscribe automatically, but i don't know if it's the same case when piped with shareReplay(..) operator, because i read that this one can leads to a Memory Leaks problems if it's misused.

My following example is used to cache Http Requests results and here is my chunk of codes:

Service:

@Injectable()
export class HttpcatcherService {
  codeNafsObservable$: Observable<any>;
  constructor(private positionservice: PositionserviceService) {}

  getAllEntities(refresh?: boolean): Observable<any> {
    if (!this.codeNafsObservable$ || refresh) {
      this.codeNafsObservable$ = forkJoin([
        this.positionservice.getPosts(), // returns Observable
        this.positionservice.getComments(), // returns Observable
      ]).pipe(
        map((result) => {
          return {
            secondNaf: result[0],
            thirdNaf: result[1],
          };
        }),
        shareReplay(1)
      );
    }
    return this.codeNafsObservable$;
  }
}

My component (knowing that i use the same way/logic in different components also):

@Component({
  selector: 'hello',
  templateUrl: './hello.component.html',
  styles: [`h1 { font-family: Lato; }`],
})
export class HelloComponent implements OnInit {
  constructor(private httpCat: HttpcatcherService) {}
  ngOnInit(): void {
    this.httpCat
    .getAllEntities()
    .subscribe(
     (result) => {
      console.log(result);
    });
  }
}

CodePudding user response：

I think that the use of forkJoin together with shareReplay does not cause any memory leak issue. The reason is the following:

• forkJoin notifies when all the Observables passed in as parameters complete. In this case, since the Observable passed to forkJoin are related to http calls, they complete just after the http call returns. therefore as soon as both http calls return, the Observable created by forkJoin notifies and then completes.
• The effect of shareReplay is to create an internal cache that makes sure that any subscription will get the last value notified by upstream. In this case, the last value notified by upstream is the only value notified by the Observable created by forkJoin

The risk of memory leaks or, at least, the risk of uncontrolled execution of code with shareRepley comes when the source Observable is an infinite stream (for instance a stream created with interval). In this case, shareReplay creates an Observable which never completes, even if there are no subscribers, unless the refCount property of ShareReplayConfig is set to true.

This behavior is explained in details in the documentation.

In any case this is not a risk you run, since the source Observable you have will emit only once and then complete.

CodePudding user response：

As of RxJS 7, shareReplay(1) is a wrapper around share with the following parameters:

  share({
    connector: () => new ReplaySubject(1),
    resetOnComplete: false,
    resetOnError: false,
    resetOnRefCountZero: false
  })

If you want finer control over how shareReplay works, use share instead and set the configuration options as you need.

For example, if you set resetOnRefCountZero to true, then when there are no longer any subscribers the inner ReplaySubject will be unsubscribed.