Объединить два потока RxJS (на основе синхронизации)

Question

У меня есть две темы RxJS, скажем a и b, что мне нужно как-то соединить.

someComboOfAandB.subscribe({aVal, bVal} => console.log("value:", aVal, bVal)); 

Я хочу, чтобы объединить их таким образом, что если a и b обновляются синхронно, значения поставляются вместе:

a.next(1); 
// some code 
b.next(2) 

// at end of synchronous code/frame: 
// value: 1 2 

Однако, если только одно значение обновляется, обновление будет еще толкания в то же время будет перенесено обновление с двумя новыми значениями:

a.next(5) 

// at end of synchronous code/frame: 
// value: 5 2 

Возможно ли это? Если да, то как? Даже если это возможно, это то, чего следует избегать?

Answer 1

Вы должны быть в состоянии использовать Scheduler, чтобы осуществить поведение, которое вы хотите:

import "rxjs/add/observable/combineLatest"; 
import "rxjs/add/operator/map"; 

import { BehaviorSubject } from "rxjs/BehaviorSubject"; 
import { Observable } from "rxjs/Observable"; 
import { asap } from "rxjs/scheduler/asap"; 

let a = new BehaviorSubject(1); 
let b = new BehaviorSubject(2); 
let combined = Observable 
    .combineLatest(a, b, asap) 
    .map((values) => ({ aVal: values[0], bVal: values[1] })); 

combined.subscribe(
    ({ aVal, bVal }) => { console.log("value:", aVal, bVal); } 
); 

a.next(3); 
b.next(4); 

Вышеприведенный код выведет следующее:

value: 3 4 

Если asapScheduler не указан, вывод будет:

value: 1 2 
value: 3 2 
value: 3 4 

Репо RxJS GitHub содержит Scheduler documentation.