How to do unsubscribe when I got error in the pipeline?

I do subscribe to the subjectPipe pipeline outside doWork function.

Inside doWork function I throw an error (can be as part of my logic flow).

Everything works as expected but subjectPipe is log: "in subject error" and because I using take(1) I thought it should unsubscribe, But it does not happen because the complete function is not invoke.

How to make it unsubscribe when error happens?

The only thing I can think of is to create variable error$ and use takeUntil and then on error function to invoke error$. but I don't want to do it because is extra variable and the code will be messy if I do it for every case I have.

stackblitz.com

import { Subject, throwError } from 'rxjs';
import { map, switchMap, take, tap } from 'rxjs/operators';

console.clear();

const doWork = () => {
  const subject = new Subject();

  const work = () => {
    throw new Error('bla');
  };

  const subjectPipe = subject.pipe(
    tap(() => {
      console.log('in subject pipeline');
    }),
    switchMap(() => work())
  );

  subjectPipe.subscribe({
    next: () => console.log('in subjectPipe next'),
    error: () => console.log('in subjectPipe error'),
    complete: () => console.log('in subjectPipe complete'),
  });

  return { subject, subjectPipe };
};

const { subject, subjectPipe } = doWork();

subjectPipe
  .pipe(
    take(1),
    tap(() => console.log('continue the pipe...'))
  )
  .subscribe({
    next: () => console.log('in subject next'),
    error: () => console.log('in subject error'),
    complete: () => console.log('in subject complete'),
  });

subject.next(null);

CodePudding user response：

The error means completion (however, the callback associated with complete never gets called). So, there is no need to unsubscribe from the stream manually here.

In your case, if you need to call some logic in both completion and error, you can use the RxJS finalize operator, which will be called in both cases.

subjectPipe
  .pipe(
    take(1),
    tap(() => console.log('continue the pipe...')),
    finalize(() =>
      console.log(
        'You can do the final stuff here, instead of `complete` callback'
      )
    )
  )
  .subscribe({
    next: () => console.log('in subject next'),
    error: () => console.log('in subject error'),
    complete: () => console.log('in subject complete'),
  });