Javascript 当我在 rxjs 的管道中遇到错误时如何执行取消订阅

当我在管道中遇到错误时，怎么做unsubscribe？

我确实订阅了doWork函数外的subjectPipe管道.

在doWork函数中，我抛出一个错误(可以作为逻辑流的一部分).

一切正常，但subjectPipe是log:"in subject error"，因为我使用take(1)，所以我认为应该取消订阅，但它没有发生，因为完整的函数不是invoke.

当错误发生时，如何使其达到unsubscribe？

我唯一能想到的就是创建变量error$，使用takeUntil，然后在错误函数上调用error$.但我不想这样做，因为这是一个额外的变量，如果我对每一个 case 都这样做，代码会很混乱.

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);