RXJS Подождите все наблюдаемые в массиве для завершения (или ошибка)

Question

я выдвигаю наблюдаемые в массив, как, например ...

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

Я хочу наблюдаемый, который излучает, когда все задачи $ завершены. Имейте в виду, что на практике задачи $ не имеют известного числа наблюдаемых.

Я пробовал Observable.zip(tasks$).subscribe(), но это кажется неудачным в случае, если есть только одна задача, и заставляет меня полагать, что для ZIP требуется четное количество элементов, чтобы работать так, как я ожидал.

Я пробовал Observable.concat(tasks$).subscribe(), но результат оператора concat просто кажется массивом наблюдаемых ... например. в основном то же, что и вход. Вы даже не можете подписаться на него.

в C# это будет сродни Task.WhenAll(). в ES6 обещал, что это будет похоже на Promise.all().

Я столкнулся с рядом вопросов, но все они, похоже, имеют дело с ожиданием на известном количестве потоков (например, их сопоставление).

Answer 1

Если вы хотите, чтобы составить наблюдаемый, который излучает, когда все источник наблюдаемым завершено, вы можете использовать forkJoin:

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

Для меня это sample был лучшим решением.

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

Так .. только тогда, когда второй (пример) испускают - вы увидите последнее emited значение первого (источник).

В моей задаче я жду подтверждения формы и другого события DOM.