Powtórz żądanie (Angular2 - http.get) n sekund po zakończeniu

Question

Grałem z kącikiem 2 i utknąłem po chwili.

Korzystanie http.get działa dobrze dla pojedynczego żądania, ale chcę sondować live-danych co 4 sekundy, po majstrować na dłuższą chwilę i przeczytaniu wielu reactivex rzeczy skończyło się z:

Observable.timer(0,4000) .flatMap( () => this._http.get(this._url) .share() .map(this.extractData) .catch(this.handleError) ) .share();

Czy istnieje prosty sposób, aby rozpocząć interwał (4 sekundy) po tym, jak http.get -observable wyemitował wynik żądania? (Albo skończę w piekle obserwowalnym-?)

Timeline chcę:

Time(s): 0 - - - - - 1 - - - - - 2 - - - - - 3 - - - - - 4 - - - - - 5 - - - - - 6 Action: Request - - Response - - - - - - - - - - - - - - - - - - - -Request-... Wait: | wait for 4 seconds -------------------------> |

Answer 1

Używam RxJS 5 i nie jestem pewien, co to są operatory równoważne RxJS 4. W każdym razie tutaj jest moje RxJS 5 rozwiązanie, nadzieję, że to pomaga:

var pollData = this._http.get(this._url) 
      .map(this.extractData) 
      .catch(this.handleError); 
pollData.expand(
() => Observable.timer(4000).concatMap(() => pollData) 
).subscribe(); 

Rozwinięcie operator wyda dane i rekurencyjnie rozpocząć nowy Obserwowalne z każdej emisji

Answer 2

Można spróbować użyć odstępu, jeżeli jest to bardziej wygodne. Wywołanie subscribe daje Subscription, która pozwala anulować polling po pewnym czasie.

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(); 

Answer 3

udało mi się zrobić to sam, z jedynym minusem beeing że http.get nie można powtórzyć łatwiej.

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(); } 
    ); 
}; 

---

Oto dlaczego nie można użyć więcej prostą subskrypcji przodu:

var subscribeToNewRequestObservable =() => { 
    this._pollLiveData() 
     .subscribe(pollSubject); 
    }; 

Zakończenie http.get -observable również zakończyć temat i uniemożliwić dalsze elementy emitujące .

---

To wciąż zimno obserwowalne, więc jeśli nie zapisać się do niej żadne wnioski zostaną wykonane.

this._pollService.pollData().subscribe(
    (res) => { this.count = res.count; } 
); 

Answer 4

A minor przerobienie na answer od Can Nguyen, w Przypadek, dla którego opóźnienie odpytywania ma zależeć od poprzedniego stanu zakończenia żądania.

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(); 

Plunk.

Lub alternatywnie:

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 

Plunk.