TheCombineLatest operator behaves in a similar way toZip, but whileZip emits items only wheneach of the zipped source Observables have emitted a previously unzipped item,CombineLatest emits an item wheneverany 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 ascombineLatest. It may take between two and nine Observables (as well as the combining function) as parameters, or a singleList of Observables (as well as the combining function). It does not by default operate on any particularScheduler.
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 thewithLatestFrom operator. It is similar tocombineLatest, but only emits items when the single source Observable emits an item (not whenany of the Observables that are passed to the operator do, ascombineLatest does).
RxJava implements this operator ascombineLatest. It may take between two and nine Observables (as well as the combining function) as parameters, or a singleList of Observables (as well as the combining function). It does not by default operate on any particularScheduler.
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 thewithLatestFrom operator. It is similar tocombineLatest, but only emits items when the single source Observable emits an item (not whenany of the Observables that are passed to the operator do, ascombineLatest does).
RxJS implements this operator ascombineLatest. It may take a variable number of individual Observables (as well as the combining function) as parameters, or a singleArray 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 valuevar 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: 0Next: First: 1, Second: 0Next: First: 1, Second: 1Next: First: 2, Second: 1Completed
RxJS also has awithLatestFrom operator. It is similar tocombineLatest, but only emits items when the single source Observable emits an item (not whenany of the Observables that are passed to the operator do, ascombineLatest 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: 1Next: First: 1, Second: 4Next: First: 2, Second: 7Next: First: 3, Second: 10Completed
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 ascombineLatest.
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: 0Next value: First: 1, Second: 0Next value: First: 1, Second: 1Next value: First: 2, Second: 1Complete!
RxPHP also has an operatorwithLatestFrom.
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: 0Next value: First: 1, Second: 4, Third: 1Next value: First: 2, Second: 7, Third: 3Next value: First: 3, Second: 10, Third: 4Complete!