How should I write my single run RxJs code so that I can test against subscription leaks?

Question

with this code:

select(): void {
        this.initialObservable$
            .pipe(
                first(),
                switchMap(() => this.service.getData())
            ).subscribe(// do stuff);
}

I just experienced a memory leak bug. the service call emitted again later and I realised that the subscription was never cleaned up because the subscribe content ran again.

Obviously this is my fix:

select(): void {
        this.initialObservable$
            .pipe(
               switchMap(() => this.service.getData()),
                first()
            ).subscribe(// do stuff);
}

and that now works fine - moving the first() operator call to the end of the pipe.

I have been looking a little at jasmine marble tests and I believe I would need a reference to the observable stream to test it, which I dont have. So regression testing this will be quite messy and that makes me think what I am doing must be bad practice. What is the correct way to write such code in a unit testable way please? Preferable with marble tests.

Answer 1

If you return that observable here and subscribe in the caller, then this would get easily testable.

scheduler.run(({ cold, expectObservable }) => {
  const source$ = cold('      -n--y--', { n: false, y: true });
  const expectedMarble = '    ----x';
  const result$ = select();
  
  expectObservable(result$).toBe(expectedMarble, { x: true });
});