RxJs Beobachtbares Intervall bis zum Erreichen des gewünschten Werts

Question

Ich möchte nach Änderungen suchen und wenn ein gewünschter Wert erreicht ist, sollte das Observable beendet werden (oder bis zum Timeout warten). Im Moment benutze ich den Filter, der gut funktioniert, bis der gewünschte Wert erreicht ist. Aber ich möchte, dass das Observable Ereignisse pusht, während es auf diesen Wert wartet.

Zum Beispiel warte ich auf den Status 'Erfolg' und bis der Status auf 'Erfolg' wechselt, wird der Status 'Testen' von meinem Dienst zurückgegeben. Aber da der Filter auf "Erfolg" wartet, kommt "Testen" nie zurück.

jetzt Mein Code:

return Observable 
    .interval(this.POLL_TIMEOUT) 
    .flatMap(() => this.getSingleProjectStatus(projectId, repoName)) 
    .filter(data => this.finishedStatus(data.status)) 
    .take(1) 
    .timeout(this.MAX_TIMEOUT, Observable.throw(new Error('Timeout'))); 

Answer 1

Sie wollen wahrscheinlich takeWhile statt filter.

return Observable 
    .interval(this.POLL_TIMEOUT) 
    .flatMap(() => this.getSingleProjectStatus(projectId, repoName)) 
    .takeWhile(data => this.finishedStatus(data.status)) 
    .timeout(this.MAX_TIMEOUT, Observable.throw(new Error('Timeout')); 

Hinweis nimmt die vor allem außer das letzte Ereignis, wenn Sie das letzte Ereignis wollen auch Sie ein wenig schwieriger sein müssen.

const source = Observable.interval(this.POLL_TIMEOUT) 
    .flatMap(() => this.getSingleProjectStatus(projectId, repoName)) 
    .share(); 

    source 
    .takeUntil(source.filter(data => this.finishedStatus(data.status))) 
    .timeout(this.MAX_TIMEOUT, Observable.throw(new Error('Timeout')); 

In diesem Fall Sie alle Ergebnisse bis ein anderes Observable aussendet einnehmen, ist die andere Observable in diesem Fall einfach die source so gefiltert, dass sie emittiert nur Erfolgsereignisse.

JsBin: http://jsbin.com/sojosuhune/edit?html,js,console,output