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

【问题标题】：Stop RxJS HTTP Polling after HTTP success responseHTTP 成功响应后停止 RxJS HTTP 轮询
【发布时间】：2023-03-17 09:05:04
【问题描述】：

我有以下模仿 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 请求（在此演示版本中它记录“轮询...”）。出于某种原因，即使在 10000 毫秒超时后它也会继续发送请求。我该怎么做？

【问题讨论】：

也许使用retry而不是repeat？
恐怕不行。请参阅此处的示例 2 learnrxjs.io/learn-rxjs/recipes/http-polling
timer(10000) 将在 10 秒后发出 0，这是一个虚假值。所以takeUntil 不会停止管道。您可以使用 timer(10000,1) 的重载使其真实，或者使用 timeout 运算符将其与您可以在订阅中触发的主题合并。
@Eldar 无关紧要takeUntil 下的流是否发出虚假值，它发出就足够了。即使是Subject<void> 也可以。
@Eldar 感谢您的回答，但 timer(10000,1) 没有帮助。你能用timeout 运算符为你的另一个理论写一个代码sn-p 吗？

标签： angular rxjs observable polling rxjs-observables

【解决方案1】：

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

超时（10 秒）后
当响应满足您的条件时（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 将在目标 observable 发出一个值时停止。 timer 将在 10 秒后发出一个值。 race 将发出一个来自 stopper 或来自 timer 的值，哪个先出现。

Stackblitz

【讨论】：

【解决方案2】：

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

您可以尝试以下操作，而不是重复：

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

【讨论】：