Dynamically Chain a Sequence of Filters

rxjs-chain-filters.js

import Rx from 'rx/dist/rx.all';

/**
 * Filters accept an input, and return an observable that will yield a single value (the filtered input).
 *
 * @param x
 * @returns {Observable<T>}
 */
function f1( x ) {
	return Rx.Observable.create( function( observer ) {
		Rx.Scheduler.default.scheduleWithRelativeAndState(x, 5000, function (_, state) {
			observer.onNext(state);
			observer.onCompleted();
		});
	});
}

function f2( x ) {
	return Rx.Observable.return( ( 10 * x ) + 2 );
}

function f3( x ) {
	return Rx.Observable.return( ( 10 * x ) + 3 );
}

/**
 * The `demo` function demonstrates, by manually chaining filters, what I would like to accomplish dynamically.
 *
 * @param input {} A value to be processed by a chain of filters.
 * @returns {Observable} The output after processing `input` through the chained filters.
 */
function demo( input ) {
	return Rx.Observable.return( input ).concatMap( f1 ).concatMap( f2 ).concatMap( f3 );
}

demo( 0 ).subscribe( function( result ) {
	console.log( "Demo:", result );
});

/**
 * I'd like to identify I way reach the same results dynamically, given an array (or sequence) of filters.
 *
 * @param input {} A value to be processed by a chain of filters.
 * @param filters {Array} An array of filters through which to process the input.
 * @returns {Observable} The output after processing `input` through the chained filters.
 */
function chain( input, filters ) {
	let seq = Rx.Observable.fromArray( filters );

	return seq.reduce( ( chain, filter ) => chain.concatMap( filter ), Rx.Observable.return( input ) ).concatMap( value => value );
}

let arr = [f1, f2, f3];

chain( 0, arr ).subscribe( function( result ) {
	console.log( "Chain:", result );
});