Combine two RxJS streams (based on synchrony)

Question

I have two RxJS subjects, say a and b that I need to combine somehow.

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

I want to combine them such that if a and b are updated synchronously, the values are delivered together:

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

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

However, if just one value is updated, an update will still be pushed at the same time an update with two new values would be pushed:

a.next(5)

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

Is this possible? If it is, how so? Even if it is possible, is it something that should be avoided?

Answer 1

You should be able use a Scheduler to effect the behavior you want:

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);

The above code will output the following:

value: 3 4

If the asap Scheduler is not specified, the output would be:

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

The RxJS GitHub repo contains some Scheduler documentation.