Bug iphone：订阅一个可观察的返回给我它发出的第一个值，

Question

你能告诉我问题是否出在这段代码中。我正在尝试设置数据轮询。对第一个请求的响应之后的响应不会传输到后端，并且我返回的 observables 都具有与第一个响应相同的值。即使后端的状态发生变化也是如此。

示例如果第一个 getStatus 返回打开，则不会传输以下请求并且将打开可观察的响应

在我的服务中，我有以下代码

getIntervalStatus(sessionId: string): Observable<HttpResponse<string>> {

    return this.getStatus(sessionId).pipe(
      map(
        (result: HttpResponse<string>) => {
          this.statusOpened = result.body.toLowerCase();
          this.utils.alertLog(this.getStatus, result.body.toLowerCase());
          this.statusEmit.next(this.statusOpened);
          if (this.statusOpened === 'opened' || this.statusOpened === 'closed') {
            this.getIntervalStatus(sessionId).subscribe( );
          }
          return result;
        }
      ),
      catchError((err) => {
        this.statusOpened = 'error';
        this.utils.alertLog(this.getStatus, JSON.stringify(err));
        this.statusEmit.error(err);
        throw err;
      })
    );

  }

getStatus(sessionId: string): Observable<HttpResponse<string>> {
    const httpParams = new HttpParams().set('sessionId', sessionId);
    return this.http.post<string>(this.fullWsUrl + '/Payment/GetPaymentStatus', null,
      { observe: 'response', params: httpParams }).pipe(
         map((value: HttpResponse<string>) => {
          this.utils.alertLog(this.getStatus, value.status.toString());
          return value;
        })
      );
  } 

在我的组件中，我有以下代码

      this.openSessionSubscription = this.device.openSession(amount, this.printerVMService.getSessionId()).subscribe(
        data => {
          this.utils.alertLog(this.device.getIntervalStatus, 'Session is opened');
          this.sessionIsOpened = true;
          // subject for information session is opened
          this.device.statusOpened = 'opened';

          this.device.statusEmit = new Subject<string>();
          this.statusEmitSubscriber = this.device.statusEmit.subscribe(
            status => {

              this.status = status;
              console.log(this.constructor.name + 'payment status', this.status);
              switch (this.status) {
                case 'opened':
                  break;
                case 'payed':
                  this.utils.alertLog(this.device.getIntervalStatus, 'statut payed');
                  this.localStorageService.storePaymentIsMade(true);
                  this.dbService.addOrderPaymentResult(ProcessResult.done, this.printerVMService.getSessionId()).subscribe();
                  this.router.navigate(['welcome/paying/paying_accepted']);
                  break;
                case 'closed':
                  this.utils.alertLog(this.device.getIntervalStatus, 'statut closed');
                  // this.paymentSessionIsClosed = true;
                  // this.dbService.addOrderPaymentResult(ProcessResult.error, this.printerVMService.getSessionId()).subscribe();
                  // this.utils.deleteSectionAction(this.printerVMService.getPrinterVM());
                  break;
                case 'used':
                  this.utils.alertLog(this.device.getIntervalStatus, 'status used');
                  // this.utils.deleteSectionAction(this.printerVMService.getPrinterVM());
                  break;
                default:
                  console.error('status don\'t exist');
                  this.utils.alertLog(this.device.getIntervalStatus, 'statut not exist');
                  this.utils.deleteSectionAction(this.printerVMService.getPrinterVM());
              }
            },
            error => {
              this.alert.success('statut error');
              console.error(this.translate.instant('PAYMENT.DEVICE.GETSTATUSERRORMSG'));
              this.utils.deleteSectionAction(this.printerVMService.getPrinterVM());
            }
          );

          this.utils.alertLog(this.device.getIntervalStatus, 'getIntervalStatusSubscription is call');
          this.getIntervalStatusSubscription = this.device.getIntervalStatus(this.printerVMService.getSessionId()).subscribe();

        },
        error => {
          // emit session not open
          console.error(this.translate.instant('PAYMENT.DEVICE.OPENERRORMSG'));
          this.utils.alertLog(this.device.openSession, 'open error');
          this.utils.deleteSectionAction(this.printerVMService.getPrinterVM());
        }
      );

Answer 1

问题

您希望轮询您的 API，直到返回某些状态。如果返回状态opened 或closed，您想重复请求。如果返回状态 payed、used 或其他内容，则您希望您的组件执行操作。

你的实现问题

您正在创建大量嵌套订阅，这通常是个坏主意。应该有一个顶级订阅者 - 其他所有内容都可以链接为可观察对象。

我的设计

我会首先寻求删除所有嵌套订阅。然后我想使用expand 运算符递归地链接一个可观察对象——这就是轮询的处理。

我要为强类型创建一个Status 类型。

getStatus 函数将返回 Observable<Status>，因为它不需要返回完整的响应。

为了回答的目的，我将简化您正在做的事情。这可能意味着我删除了您想要的功能，但我将主要关注轮询方面。然后，您应该能够重新添加任何额外的功能。

我的实现

状态

export type Status = 'open' | 'closed' | 'payed' | 'used' | 'error';

服务

private continuePollingStatuses: Status[] = [
  'open', 'closed'
];

getIntervalStatus(sessionId: string): Observable<Status> {
  return this.getStatus(sessionId).pipe(
    expand((status: Status) => {
      if (this.continuePollingStatuses.includes(status)) {
        // repeat the request
        return this.getStatus(sessionId);
      }     
      // do not repeat the request
      return empty();
    }),
    catchError((err) => of('error'))
  );
}

private getStatus(sessionId: string): Observable<Status> {
  // cache-busting query param
  const timestamp = new Date().getTime();
  const url = `${this.fullWsUrl}/Payment/GetPaymentStatus?t=${timestamp}`;
  const httpParams = new HttpParams().set('sessionId', sessionId);
  const options = { observe: 'response', params: httpParams };
  return this.http.post<string>(url, null, options).pipe(
    map(response => response.body.toLowerCase())
  );
} 

组件

ngOnInit() {
  const sessionId = this.printerVMService.getSessionId();
  this.service.getIntervalStatus(sessionId).pipe(
    switchMap((status: Status) => this.performAction(status))
  ).subscribe(status => {
    this.status = status;    

    switch (this.status) {
      case 'opened':        
      case 'closed':      
      case 'used': 
        break;
      case 'payed':        
        this.localStorageService.storePaymentIsMade(true);
        this.router.navigate(['welcome/paying/paying_accepted']);
        break;
      case 'error':
        console.error(this.translate.instant('PAYMENT.DEVICE.GETSTATUSERRORMSG'));
        this.utils.deleteSectionAction(this.printerVMService.getPrinterVM());
        break;
      default:
        console.error('status don\'t exist');
        this.utils.deleteSectionAction(this.printerVMService.getPrinterVM());
     }
   });
}

private performAction(status: Status): Observable<Status> {
  switch (status) {
    case 'payed':
      return this.dbService.addOrderPaymentResult(ProcessResult.done, 
        this.printerVMService.getSessionId()).pipe(
        map(() => of(status))
      );
  }

  return of(status);
}

演示：https://stackblitz.com/edit/angular-7gkixg