我正在尝试轮询 REST API 以更新使用以下代码正常工作的数据表:
pollData(url, interval) {
return Rx.Observable.interval(interval)
.mergeMap(() => this.http.get(url));
}
// get data
this.dataService.pollData(this.url, this.updateInterval)
.subscribe(
data => console.log(data),
err => console.log(err),
() => console.log('done'));
问题是错误和完成永远不会被调用。任何与onError和onCompete一起工作的建议将不胜感激。谢谢!
关于对观察者的onComplete
调用,它只会在源可观察量完成时生效。这意味着 pollData
返回的可观察量何时完成。由于您当前在没有退出条件的情况下进行轮询,因此您的可观察量自然永远不会完成。
要完成这个可观察性,您需要提出一个退出条件:
- 超时(例如,轮询 X 秒,然后停止轮询(
- 投票次数
- 基于轮询数据的条件(例如,如果在 X 连续轮询后未检测到更改(
- 外部完成信号
- 对您的用例有意义的任何其他条件
所有这些条件都很容易通过 RxJS 实现,它们需要您更新 pollData
函数的代码。
例如,对于外部完成信号,您可以编写:
// defining somewhere the subject for signalling end of polling
stopPollingS = new Rx.Subject();
// somehow pass this subject as a parameter of the polling function
pollData(url, interval, stopPollingS) {
return Rx.Observable
.interval(interval)
.mergeMap(() => this.http.get(url))
.takeUntil(stopPollingS);
}
// somewhere in your code when you want to stop polling
stopPollingS.onNext(true);
关于观察者的onError
电话,我不确定我是否了解正在发生的事情。您是否尝试过引发错误并检查观察者的onError
处理程序是否确实被调用?如果没有错误,很明显不会调用onError
。
以防万一有人想知道我是如何解决这个问题并实现所需功能的。基本上,我只需要将可观察量包装在另一个中并将错误作为数据返回。
initiatePolling(url, interval) {
var http = this.http;
return Rx.Observable.create(function (observer) {
// initial request (no delay)
requestData();
var timerId = setInterval(requestData, interval);
function requestData() {
var subscription = http.get(url).timeout(20000)
.subscribe(
result => {
observer.next(result);
subscription.unsubscribe();
},
err => {
observer.next(err);
subscription.unsubscribe();
},
() => {
subscription.unsubscribe();
});
}
return function () {
observer.complete();
window.clearInterval(timerId);
}
});
}