rxjs-consecutive-operator

Observable#consecutive

Custom RxJS 5 operator for mapping several consecutively emitted values at once.

Motivation

Say you want to compare every emitted value with previous one.

You could compose such a chain in RxJS:

function compare(a, b) {
    return a < b;
}
 
const comparisonResults = values              // Observable of a < b comparison results (booleans);
    .bufferCount(2, 1)
    .takeWhile(values => values.length === 2) // When completing bufferCount emits values 
                                              // that were left over in buffer at that 
                                              // moment - we don't want that.
    .map(([a, b]) => compare(a, b));
 

consecutive allows you to write:

const comparisons = values.consecutive(compare); // Note how we pass compare function
                                                 // directly as argument.

consecutive decides based on length of passed function (number of arguments it has in definition) how many values it should buffer. It guarantees that passed function will be always called with as many parameters as it expects.

Alternatively you can provide number of arguments that should be passed to a function (good for functions that have infinite length):

const comparisons = values.consecutive((...args) => {}, 2); // Will be always called with 2 arguments,
                                                            // even though its length is infinite.

Marbles

const values;                                             // ---1---2---3---2---|--->
 
const comparisions = values.consecutive((a, b) => a < b); // -------t---t---f---|--->
                                                          // t - true; f - false

const values;                                            // ---1---2---3----2--|--->
 
const sums = values.consecutive((a, b, c) => a + b + c); // -----------6----7--|--->

Usage

consecutive is exported as function that takes source Observable as first parameter.

You can import it directly and use with let operator like so:

import { consecutive } from 'rxjs-consecutive-operator';
 
values.let(v => consecutive(v, fn));

If this feels awkward or you want that operator always near, there is option to monkey patch Observable:

import 'rxjs-consecutive-operator/add';
 
Observable.consecutive; // Now it's defined!