Angular中Observable出错后，如何点击按钮继续刷新该Observable？

问题描述 #

开发Angular应用时，通过自定义state管道和async管道从API获取数据展示，页面设有刷新按钮。但API请求出错后，Observable停止发射值，导致无法再次刷新。且不想使用catchError(()=>of([]))的处理方式，需要在UI上展示错误信息。

问题代码 #

state.pipe.ts #

import { Pipe, PipeTransform } from '@angular/core';
import { Observable, catchError, map, of, startWith } from 'rxjs';

type ObsState<T> = Observable<ObsStateAsync<T>>;

export type ObsStateAsync<T> = { loading: boolean; result?: T; error?: any };
@Pipe({
  name: 'state',
})
export class StatePipe implements PipeTransform {
  transform<T>(val: Observable<T>): ObsState<T> {
    return val.pipe(
      map((value: any) => ({ loading: false, result: value })),
      startWith({ loading: true }),
      catchError((error) => of({ loading: false, error }))
    );
  }
}

组件代码 #

@Component({
  selector: 'app-root',
  template: `
    <button (click)="onRefresh()">Refresh</button>
    @let breedState = breeds$ | state | async;

    @if(breedState?.loading){
      Fetching ...
    }

    @if(breedState?.error){
      Error
    }

    @if(breedState?.result){
      @for(breed of breedState?.result;track $index){
        <div> 
          {{ breed.attributes.name }}
        </div>
      }
    }
  `,
  imports: [StatePipe, AsyncPipe],
})
export class App {
  private readonly httpClient = inject(HttpClient);

  refresh = new BehaviorSubject<void>(undefined);

  getBreed$ = this.httpClient
    .get('https://dogapi.dog/api/v2/breeds') // 修改接口可触发错误
    .pipe(map((res: any) => res.data));

  breeds$ = this.refresh.asObservable().pipe(switchMap(() => this.getBreed$));

  onRefresh(): void {
    this.refresh.next();
  }
}

问题原因 #

当API请求出错时，getBreed$的错误会冒泡到breeds$主流，导致整个breeds$流终止。后续点击刷新按钮时，refresh Subject虽发送新值，但已终止的breeds$流无法响应，无法发起新请求。state管道的catchError虽能捕获错误，但它是在breeds$流之后应用的，无法阻止主流终止。

解决方案 #

需要在switchMap内部为每个请求单独处理错误，确保错误不会冒泡到breeds$主流，让主流保持活跃，每次刷新都能发起新请求。

方式一：在switchMap内嵌入状态处理逻辑 #

修改breeds$的定义，将状态处理（加载、成功、错误）移到switchMap内部，确保每个请求的错误被单独捕获：

breeds$ = this.refresh.asObservable().pipe(
  switchMap(() => 
    this.getBreed$.pipe(
      map(value => ({ loading: false, result: value })),
      startWith({ loading: true }),
      catchError(error => of({ loading: false, error }))
    )
  )
);

此时模板可简化，直接使用breeds$ | async，无需state管道：

@let breedState = breeds$ | async;

方式二：在getBreed$内部处理状态和错误 #

将状态处理逻辑整合到getBreed$中，让每个请求自身处理错误：

getBreed$ = this.httpClient
  .get('https://dogapi.dog/api/v2/breeds')
  .pipe(
    map((res: any) => ({ loading: false, result: res.data })),
    startWith({ loading: true }),
    catchError(error => of({ loading: false, error }))
  );

breeds$ = this.refresh.asObservable().pipe(switchMap(() => this.getBreed$));

模板同样直接使用breeds$ | async即可。

额外修正：state管道的操作符顺序（可选） #

如果仍想保留state管道，需修正操作符顺序，确保startWith先发出加载状态，再处理响应和错误：

transform<T>(val: Observable<T>): ObsState<T> {
  return val.pipe(
    map(value => ({ loading: false, result: value })),
    catchError(error => of({ loading: false, error })),
    startWith({ loading: true })
  );
}

但核心仍需确保在switchMap内部处理每个请求的错误，避免主流终止。

内容的提问来源于stack exchange，提问作者Nhut Truong