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

Question

我有以下模仿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
}
});
}

我希望我的轮询停止发送 HTTP 请求(它记录"轮询..."在此演示版本中)，当服务器响应data.length> 0时。出于某种原因，即使在 10000 毫秒超时后，它也会继续发送请求。我该怎么做？

Answer 1

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

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发出一个值。

斯塔克闪电战

Answer 2

Repeat

返回一个 Observable，该 Observable 将在源流完成时重新订阅源流，在您的情况下，尽管源 Observable 完成(感谢takeUntil)，使用 repeat 将重复重新订阅到源流

与其重复，不如尝试以下操作：

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