我玩了angular2并且在一段时间后卡住了。
使用http.get
可以正常处理单个请求,但是我想每4秒轮询一次实时数据,经过一段时间的修补并阅读了很多反应性的东西,我最终得到了:
Observable.timer(0,4000)
.flatMap(
() => this._http.get(this._url)
.share()
.map(this.extractData)
.catch(this.handleError)
)
.share();
在http.get
-observable发出请求结果后,是否有一种简单的方法可以启动(4秒)间隔? (或者我最终会在可观察到的地狱?)
我想要的时间表:
Time(s): 0 - - - - - 1 - - - - - 2 - - - - - 3 - - - - - 4 - - - - - 5 - - - - - 6
Action: Request - - Response - - - - - - - - - - - - - - - - - - - -Request-...
Wait: | wait for 4 seconds -------------------------> |
更新到RxJS 6
import { timer } from 'rxjs';
import { concatMap, map, expand, catchError } from 'rxjs/operators';
pollData$ = this._http.get(this._url)
.pipe(
map(this.extractData),
catchError(this.handleError)
);
pollData$.pipe(
expand(_ => timer(4000).pipe(concatMap(_ => pollData$)))
).subscribe();
我正在使用RxJS 5,我不确定RxJS 4等效运算符是什么。无论如何这里是我的RxJS 5解决方案,希望它有所帮助:
var pollData = this._http.get(this._url)
.map(this.extractData)
.catch(this.handleError);
pollData.expand(
() => Observable.timer(4000).concatMap(() => pollData)
).subscribe();
扩展运算符将发出数据并以每次发射递归地启动一个新的Observable
我设法自己做,唯一的缺点是http.get
不能更容易重复。
pollData(): Observable<any> {
//Creating a subject
var pollSubject = new Subject<any>();
//Define the Function which subscribes our pollSubject to a new http.get observable (see _pollLiveData() below)
var subscribeToNewRequestObservable = () => {
this._pollLiveData()
.subscribe(
(res) => { pollSubject.next(res) }
);
};
//Subscribe our "subscription-function" to custom subject (observable) with 4000ms of delay added
pollSubject.delay(4000).subscribe(subscribeToNewRequestObservable);
//Call the "subscription-function" to execute the first request
subscribeToNewRequestObservable();
//Return observable of our subject
return pollSubject.asObservable();
}
private _pollLiveData() {
var url = 'http://localhost:4711/poll/';
return this._http.get(url)
.map(
(res) => { return res.json(); }
);
};
这就是为什么你不能使用更直接的订阅:
var subscribeToNewRequestObservable = () => {
this._pollLiveData()
.subscribe(pollSubject);
};
完成http.get
-observable也将完成您的主题并防止它发出更多的项目。
这仍然是一个冷的可观察,所以除非你订阅它,否则不会提出任何要求。
this._pollService.pollData().subscribe(
(res) => { this.count = res.count; }
);
如果您希望轮询延迟取决于先前的请求完成状态,则可以对Can Nguyen中的answer进行一次小的返工。
var pollData = () => request() // make request
.do(handler, errorHandler) // handle response data or error
.ignoreElements() // ignore request progress notifications
.materialize(); // wrap error/complete notif-ns into Notification
pollData() // get our Observable<Notification>...
.expand( // ...and recursively map...
(n) => Rx.Observable // ...each Notification object...
.timer(n.error ? 1000 : 5000) // ...(with delay depending on previous completion status)...
.concatMap(() => pollData())) // ...to new Observable<Notification>
.subscribe();
或者:
var pollData = () => request() // make request
.last() // take last progress value
.catch(() => Rx.Observable.of(null)); // replace error with null-value
pollData()
.expand(
(data) => Rx.Observable
.timer(data ? 5000 : 1000) // delay depends on a value
.concatMap(() => pollData()))
.subscribe((d) => {console.log(d);}); // can subscribe to the value stream at the end
如果更方便,您可以尝试使用间隔。调用subscribe
会给你Subscription
,让你在一段时间后取消投票。
let observer = Observable.interval(1000 * 4);
let subscription = observer.subsscribe(x => {
this._http.get(this._url)
.share()
.map(this.extractData)
.catch(this.handleError)
});
....
// if you don't require to poll anymore..
subscription.unsubscribe();