自定义 RxJs 运算符仅在第一次触发

Question

我正在尝试创建一个自定义 RxJs 运算符，用于检查 API 响应是否包含错误消息，如果存在，则会引发该消息的错误。然后我想在我的 Ticket-feedback.service.ts 中处理该错误。

BaseServerResponse> 类型：

class BaseServerResponse<T> {
    data?: T;
    error?: string;
    message?: string;
}

自定义 RxJs 运算符：

export const rxThrowCustomServerError = <T>() => {
  return function (source: Observable<BaseServerResponse<T>>): Observable<BaseServerResponse<T>> {
    return new Observable((subscriber) => {
      const subscription = source.subscribe({
        next(x) {
          if (x.error) {
            subscriber.error(x.error);
          }
          subscriber.next(x);
        },
        error(error) {
          subscriber.error(error);
        },
        complete() {
          subscriber.complete();
        },
      });

      return () => {
        subscription.unsubscribe();
      };
    });
  };
};

ticket-feedback.component.ts：

@Component({
  selector: 'ticket-feedback-list',
  standalone: true,
  imports: [...ANGULAR_MODULES, ...ANGULAR_MATERIAL_MODULES],
  providers: [],
  templateUrl: './ticket-feedback-list.component.html',
  styleUrls: ['./ticket-feedback-list.component.scss'],
  changeDetection: ChangeDetectionStrategy.OnPush,
})
export default class TicketFeedbackListComponent {
  private _service = inject(TicketFeedbackService);
  form = this._service.getFeedbackForm;
  feedbackResponse = toSignal(this._service.feedbackResponse$);
  public forSubmit() {
    this._service.feedbackRequested$.next({
      fromDate: this.form.controls.fromDate.value,
      toDate: this.form.controls.toDate.value,
      retailerMobileNumber: this.form.controls.retailerMobileNumber.value,
      pageNumber: 1,
      pageSize: 10,
    });
  }
}

票务反馈.service.ts：

  feedbackResponse$ = this.feedbackRequested$.pipe(
    filter((feedbackRequest) => feedbackRequest && !isEmpty(feedbackRequest) && this.getFeedbackForm.valid),
    switchMap((request) =>
      this._http.get<BaseServerResponse<Feedback[]>>(appApiResources.feedback, {
        params: {
          pageNumber: request.pageNumber,
          pageSize: request.pageSize,
          fromDate: request.fromDate,
          toDate: request.toDate,
          'api-version': 1,
        },
      })
    ),
    rxThrowCustomServerError(),
    catchError(() => of(new BaseServerResponse<Feedback[]>([] as Feedback[], 0, 0, 0, '', '', 'Unable to fetch feedbacks'))),
    finalize(() => this.feedbackRequested$.next({} as IGetFeedbackRequest))
  );

在 HTML 中，我只是将反馈响应信号读取为：

feedbackResponse() | json

但是，这是有效的（获取API数据，从自定义RxJs运算符抛出错误，在服务中处理catchError中的错误。）仅在第一次当我单击表单提交按钮时，但从下一次开始什么也没有发生甚至连 API 调用都没有。如果我删除 rxThrowCustomServerError()，则每次单击按钮时 API 调用都会起作用。

这里出了什么问题？

Answer 1

当可观察对象发生错误时，它就会完成。因此，当

feedbackResponse$ = this.feedbackRequested$.pipe(
    filter((feedbackRequest) => feedbackRequest && !isEmpty(feedbackRequest) && this.getFeedbackForm.valid),
    switchMap((request) =>
      this._http.get<BaseServerResponse<Feedback[]>>(appApiResources.feedback, {
        params: {
          pageNumber: request.pageNumber,
          pageSize: request.pageSize,
          fromDate: request.fromDate,
          toDate: request.toDate,
          'api-version': 1,
        },
      })
    ),
    rxThrowCustomServerError()
  );

发出一个错误，这是它发出的最后一个错误，然后它关闭，幸福地没有意识到它立即通过管道传输到

catchError()

运算符。