在 rxjs 中“管道”多次观察到答案

【问题标题】：'piping' an observable multiple times in rxjs在 rxjs 中“管道”多次观察到
【发布时间】：2020-02-01 16:33:30
【问题描述】：

这是我的情况：我正在通过 Angular 的 httpClient 发出 HTTP 请求。

如果发生错误，我想通过管道传递 observable 以捕获该错误并将自定义错误返回给订阅者。即：


let observable = null;

if (sourceType === Source.HTTP) {
  observable = this.http.get("url", options);

  observable.pipe(
    catchError((err: HttpErrorResponse) => {
      const newError = new Error();
      ...
      return throwError(newError);
    })
  );
}
...

但是，我还想为请求设置一个超时，所以稍后，我打开另一个管道：

// ... code above...

observable.pipe(timeout(2500));

然后，当执行时，第一个管道不处理错误，所以我的问题是：第二个管道是否覆盖第一个管道？在那种情况下，我认为这不应该被称为“管道”......

【问题讨论】：

标签： angular rxjs

【解决方案1】：

这不会做任何事情，除非你把它归还或者用它做点什么：

observable.pipe(
    catchError((err: HttpErrorResponse) => {
      const newError = new Error();
      ...
      return throwError(newError);
    })
  );

你缺少平等。我相信这也是你的意图。 Observable 是不可变的，无论从管道返回什么，您都应该存储、返回或订阅它以使其工作。

observable = observable.pipe(
    catchError((err: HttpErrorResponse) => {
      const newError = new Error();
      ...
      return throwError(newError);
    })
  );

【讨论】：

【解决方案2】：

每次调用pipe，都会返回新的 Observable。执行顺序仅在您 subscribe 到 observable 之后才重要。例子：

const firstObservable$ = of(1);

// firstObservable$.subscribe(console.log) -> '1'


const secondObservable$ = firstObservable$.pipe(delay(3000));

// secondObservable$.subscribe(console.log) -> '1' with a delay of 3000


const thirdObservable$ = firstObservable$.pipe(
  withLatestFrom(of(3)),
  map(data => data[0] + data[1])
);

// thirdObservable$.subscribe(console.log) -> '4' without the delay


const fourth$ = firstObservable$.pipe(
  map(a => a + 2)
).pipe(
  filter(f => f > 1)
).pipe(
  flatMap(num => this.http.get('url' + num))
);

// fourth$.subscribe(console.log) -> The result of http call to 'url3'

【讨论】：