HTTP成功响应后停止RxJS HTTP轮询

问题描述

我有以下代码模仿HTTP请求轮询。

  timeout:Observable<number> = timer(10000);

  startPollingStackblitz(arnId: string) {
    const poll:Observable<BuyingData[]> = of({}).pipe(
        mergeMap(_ => {
          console.log('polling...' + arnId);
          return of([]);
          // return this.service.getData(arnId);
        }),takeUntil(this.timeout),tap(_ => console.info('---waiting 2 secs to restart polling')),delay(2000),repeat(),tap(_ => console.info('---restarted polling')),);

    this.subscription = poll.subscribe((data) => {
      console.log('subscribe...')
      if (data.length > 0) {
        console.log('timeout...');
        console.log(this.timeout);// I want to stop polling immediately before timer will elapse
      }
    });
  }

当服务器响应 data.length> 0 时，我希望轮询停止发送HTTP请求（在此演示版本中记录为“ polling ...”）。由于某些原因，即使在10000ms超时后，它仍会继续发送请求。我该怎么办？

解决方法

Repeat返回一个Observable，它将在源流完成时重新订阅源流。在您的情况下，尽管源Observable完成了（由于takeUntil），但使用repeat将重复订阅源流

您可以尝试以下操作来代替重复操作：

const poll :Observable<BuyingData[]> = interval(2000).pipe(
  exhaustMap(() => this.service.getData())
  takeUntil(this.timeout),takeWhile(data => data.length > 0),);

,

嗯，据我了解，您有两个停止条件：

1. 超时（10秒）后
2. 当响应满足您的条件时（data.length> 0）

您可以通过将takeUntil，race和timer运算符与以下主题组合来实现此目的。

const stopper = new Subject(); // to stop emitting
const poll = of({}).pipe(
  mergeMap(_ =>
    fakeDelayedRequest().pipe(
      catchError(e => {
        console.error(e);
        return of(false);
      })
    )
  ),tap(write),tap(_ => console.info("---waiting 3 secs to restart polling")),delay(3000),tap(_ => console.info("---restarted polling")),repeat(),takeUntil(stopper.pipe(race(timer(10000)))) // this should be the last in the pipe
  // else `repeat` operator will be repeating without a condition.
);

poll.subscribe(_ => {
  const rnd = Math.random();
  if (rnd> 0.3) { // random stop condition
    console.log("closing !",rnd);
    stopper.next(); // emit the stop
  }
});

takeUntil将在目标可观察对象发出一个值时停止。 timer将在10秒后发出一个值。 race将从stopper或首先出现的timer发出一个值。

Stackblitz