在 RXJS v6 中取消和处理错误的最佳方法是什么

Question

import { fromEvent, interval, Subject, Observable, of } from "rxjs";
import { takeUntil, switchMap, catchError } from "rxjs/operators";

const obs1 = p1 => {
  return new Observable(observer => {
    setTimeout(() => {
      console.log("obs1 doing");
      observer.next(p1 + "1");
      observer.complete();
    }, 1000);
  });
};
const obs2 = p2 => {
  return new Observable(observer => {
    setTimeout(() => {
      console.log("obs2 doing");
      observer.next(p2 + "2");
      observer.complete();
    }, 1000);
  });
};
const obs3 = p3 => {
  return new Observable(observer => {
    setTimeout(() => {
      console.log("obs3 doing");
      observer.next(p3 + "3");
      observer.complete();
    }, 1000);
  });
};
const obsError1 = () => console.log("obs1 is error")
const obsError2 = () => console.log("obs2 is  error")
const obsError3 = () => console.log("obs3 is error")
const cancle = new Subject();

new Observable(observer => {
  obs1(0).subscribe(
    x1 => {
      obs2(x1).subscribe(
        x2 => {
          obs3(x2).subscribe(
            x3 => {
              observer.next(x3);
              observer.complete();
            },
            obsError3
          );
        },
        obsError2
      );
    },
    obsError1
  );
}).pipe(takeUntil(cancle)).subscribe(()=>{
  console.log()
});

setTimeout(() => {
  console.log("cancle doing");
  cancle.next();
  cancle.complete();
}, 100);

当它运行cancle.next() 时，它仍然打印obs2 doing 和obs3 doing

我知道改成serial可以取消，比如：

setTimeout(() => {
  console.log("cancle doing");
  cancle.next();
  cancle.complete();
}, 100);

of(0)
  .pipe(
    switchMap(x1 => obs1(x1)),
    catchError(error => {
      obsError1()
      throw error;
    })
  ).pipe(
    switchMap(x2 => obs2(x2)),
    catchError(error => {
      obsError2()
      throw error;
    })
  )
  .pipe(
    switchMap(x3 => obs3(x3)),
    catchError(error => {
      obsError3()
      throw error;
    })
  ).pipe(
    takeUntil(cancle)
  ).subscribe(x => {
  console.log(x);
});

但是当obs1出错时，obsError2也会被执行

我不知道处理错误的最佳方法

所有代码都在https://stackblitz.com/edit/mbqhrs

Answer 1

如果你想在流中处理错误，你需要使用catchError 操作符。 它接受一个错误，如果你能够处理错误 - 只需返回新流，如果错误应该进一步传递 - 只需再次抛出它。

o$.pipe(
 catchError(error => {
   if (error.message === 'i can handle it') {
      return of('value we want to pass');
   }
   if (error.message === 'i can handle it 2') {
      return EMPTY; // simply completes the stream without an emit.
   }
   throw error; // i don't know what to do with such an error.
 }),
).subscribe(console.log); // can receive 'i can handle it'.