RXJS für alle Observablen in einem Array Warten Sie abzuschließen (oder Fehler)

Question

Ich Observablen in ein Array wie solche drängen ...

var tasks$ = []; 
tasks$.push(Observable.timer(1000)); 
tasks$.push(Observable.timer(3000)); 
tasks$.push(Observable.timer(10000)); 

Ich möchte eine beobachtbare dass aussendet, wenn alle Aufgaben abgeschlossen $. Denken Sie daran, dass Aufgaben in der Praxis keine bekannte Anzahl von Observablen haben.

Ich habe versucht Observable.zip(tasks$).subscribe(), aber das scheint zu scheitern, wenn es nur 1 Aufgabe gibt, und führt mich zu der Annahme, dass ZIP eine gerade Anzahl von Elementen erfordert, um so zu arbeiten, wie ich es erwarten würde.

Ich habe versucht, Observable.concat(tasks$).subscribe(), aber das Ergebnis der concat-Operator scheint nur eine Reihe von Observablen ... z. im Grunde das gleiche wie die Eingabe. Sie können nicht einmal bei ihm abonnieren.

in C# wäre dies Task.WhenAll() ähnlich. in ES6 versprechen es wäre Promise.all().

Ich bin auf eine Reihe von SO Fragen gestoßen, aber sie alle scheinen sich damit zu befassen, auf eine bekannte Anzahl von Streams zu warten (z. B. sie zusammen zu kartieren).

Answer 1

Wenn Sie eine beobachtbare komponieren möchten, dass emittiert, wenn alle von der Quelle vollständig Observablen, können Sie forkJoin verwenden:

import { Observable } from 'rxjs/Observable'; 
import 'rxjs/add/observable/forkJoin'; 
import 'rxjs/add/operator/first'; 

var tasks$ = []; 
tasks$.push(Observable.timer(1000).first()); 
tasks$.push(Observable.timer(3000).first()); 
tasks$.push(Observable.timer(10000).first()); 
Observable.forkJoin(...tasks$).subscribe(results => { console.log(results); }); 

Answer 2

Für mich sample war beste Lösung.

const source = Observable.interval(500); 
const example = source.sample(Observable.interval(2000)); 
const subscribe = example.subscribe(val => console.log('sample', val)); 

Also .. nur wenn zweite (Beispiel) emittieren - Sie sehen zuletzt ausgegebenen Wert der ersten (Quelle).

In meiner Aufgabe warte ich Formvalidierung und andere DOM-Ereignis.