The CombineLatest operator behaves in a similar way to Zip, but while Zip emits items only when each of the zipped source Observables have emitted a previously unzipped item, CombineLatest emits an item whenever any of the source Observables emits an item (so long as each of the source Observables has emitted at least one item). When any of the source Observables emits an item, CombineLatest combines the most recently emitted items from each of the other source Observables, using a function you provide, and emits the return value from that function.
ReactiveXcombineLatestwithLatestFromTBD
RxGroovy implements this operator as combineLatest. It may take between two and
nine Observables (as well as the combining function) as parameters, or a single
List of Observables (as well as the combining function). It does not by default
operate on any particular Scheduler.
combineLatest(List,FuncN)combineLatest(Observable,Observable,Func2) (there are also versions that take up to nine Observables)
Under development, but not part of the 1.0 release, is the withLatestFrom operator. It is
similar to combineLatest, but only emits items when the single source Observable emits an
item (not when any of the Observables that are passed to the operator do, as
combineLatest does).
RxJava implements this operator as combineLatest. It may take between two and
nine Observables (as well as the combining function) as parameters, or a single
List of Observables (as well as the combining function). It does not by default
operate on any particular Scheduler.
combineLatest(List,FuncN)combineLatest(Observable,Observable,Func2) (there are also versions that take up to nine Observables)
Under development, but not part of the 1.0 release, is the withLatestFrom operator. It is
similar to combineLatest, but only emits items when the single source Observable emits an
item (not when any of the Observables that are passed to the operator do, as
combineLatest does).
RxJS implements this operator as combineLatest. It may take a variable number
of individual Observables (as well as the combining function) as parameters, or a single
Array of Observables (as well as the combining function).
/* Have staggering intervals */
var source1 = Rx.Observable.interval(100)
.map(function (i) { return 'First: ' + i; });
var source2 = Rx.Observable.interval(150)
.map(function (i) { return 'Second: ' + i; });
// Combine latest of source1 and source2 whenever either gives a value
var source = source1.combineLatest(
source2,
function (s1, s2) { return s1 + ', ' + s2; }
).take(4);
var subscription = source.subscribe(
function (x) {
console.log('Next: ' + x.toString());
},
function (err) {
console.log('Error: ' + err);
},
function () {
console.log('Completed');
});Next: First: 0, Second: 0 Next: First: 1, Second: 0 Next: First: 1, Second: 1 Next: First: 2, Second: 1 Completed
RxJS also has a withLatestFrom operator. It is similar to combineLatest, but
only emits items when the single source Observable emits an item (not when any of the
Observables that are passed to the operator do, as combineLatest does).
/* Have staggering intervals */
var source1 = Rx.Observable.interval(140)
.map(function (i) { return 'First: ' + i; });
var source2 = Rx.Observable.interval(50)
.map(function (i) { return 'Second: ' + i; });
// When source1 emits a value, combine it with the latest emission from source2.
var source = source1.withLatestFrom(
source2,
function (s1, s2) { return s1 + ', ' + s2; }
).take(4);
var subscription = source.subscribe(
function (x) {
console.log('Next: ' + x.toString());
},
function (err) {
console.log('Error: ' + err);
},
function () {
console.log('Completed');
});Next: First: 0, Second: 1 Next: First: 1, Second: 4 Next: First: 2, Second: 7 Next: First: 3, Second: 10 Completed
These two operators are both available in each of the following distributions:
rx.jsrx.compat.jsrx.lite.jsrx.lite.compat.js
RxPHP implements this operator as combineLatest.
Merges the specified observable sequences into one observable sequence by using the selector function whenever any of the observable sequences produces an element. Observables need to be an array. If the result selector is omitted, a list with the elements will be yielded.
//from https://github.com/ReactiveX/RxPHP/blob/master/demo/combineLatest/combineLatest.php
/* Have staggering intervals */
$source1 = \Rx\Observable::interval(100);
$source2 = \Rx\Observable::interval(120);
$source = $source1->combineLatest([$source2], function ($value1, $value2) {
return "First: {$value1}, Second: {$value2}";
})->take(4);
$subscription = $source->subscribe($stdoutObserver);
Next value: First: 0, Second: 0
Next value: First: 1, Second: 0
Next value: First: 1, Second: 1
Next value: First: 2, Second: 1
Complete!
RxPHP also has an operator withLatestFrom.
Merges the specified observable sequences into one observable sequence by using the selector function only when the (first) source observable sequence produces an element.
//from https://github.com/ReactiveX/RxPHP/blob/master/demo/withLatestFrom/withLatestFrom.php
/* Have staggering intervals */
$source1 = \Rx\Observable::interval(140)
->map(function ($i) {
return 'First: ' . $i;
});
$source2 = \Rx\Observable::interval(50)
->map(function ($i) {
return 'Second: ' . $i;
});
$source3 = \Rx\Observable::interval(100)
->map(function ($i) {
return 'Third: ' . $i;
});
$source = $source1->withLatestFrom([$source2, $source3], function ($value1, $value2, $value3) {
return $value1 . ', ' . $value2 . ', ' . $value3;
})->take(4);
$source->subscribe($stdoutObserver);
Next value: First: 0, Second: 1, Third: 0
Next value: First: 1, Second: 4, Third: 1
Next value: First: 2, Second: 7, Third: 3
Next value: First: 3, Second: 10, Third: 4
Complete!