forkJoin.subscribe с ReplaySubject не вызывается

Question

Я хотел бы создать Observable, который запускается, если оба экземпляра ReplaySubject содержат значение.Я попробовал это так:

let v1 = new Rx.ReplaySubject();
let v2 = new Rx.ReplaySubject();

v1.next('X');

const combo = Rx.Observable.forkJoin(v1, v2)

combo.subscribe( arr => console.log(arr) )

v1.subscribe( s => console.log('v1', s) );
v2.subscribe( s => console.log('v2', s) );

v2.next('Y');

Обе v1.subscribe и v2.subscribe называются, но не combo.subscribe.Что я не прав?Как можно решить эту проблему?

Answer 1

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

let v1 = new rxjs.ReplaySubject();
let v2 = new rxjs.ReplaySubject();

v1.next('X');
v2.next('Y');
v2.next('Z'); // This additional value should demonstrate the idea of forkJoin

const combo = rxjs.forkJoin(v1, v2)
combo.subscribe( arr => console.log(arr) )

v1.complete();
v2.complete();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.js"></script>

Возможно, вы ищете combineLatest вместо этого, если вы не хотите, чтобы завершались внутренние наблюдаемые:

let v1 = new rxjs.ReplaySubject();
let v2 = new rxjs.ReplaySubject();

v1.next('X');
v2.next('Y');

const combo = rxjs.combineLatest(v1, v2)
combo.subscribe( arr => console.log(arr) )

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.js"></script>

Answer 2

forkJoin ожидает завершения входных наблюдаемых, затем испускает последний испущенный элемент.Вы должны закончить оба предмета.