​

console.clear();

​

const { Observable, Subject } = Rx;

​

const input$ = new Subject();

​

//////////////////////////////////////////////

// Fake processing code to simulate real work

//////////////////////////////////////////////

​

const doStuffCommon = (multiplier, a) => {

  const time = Math.random() * multiplier;

  return Observable.of(a)

    .delay( time < 500 ? time * 10 : time)

    .map(d => d + '-FAKE PROCESSING');

};

​

const doStuffA = a => doStuffCommon(100, a);

const doStuffB = b => doStuffCommon(1000, b);

​

//////////////////////////////////////////////

// Custom operator

//////////////////////////////////////////////

​

const collate = (...segments) => source$ =>

  source$

    .mergeMap((obj, index) => {

      return segments.map(({ key, work }) => {

        const input = obj[key];

        const output$ = work(input);

​

        return Observable.from(output$).map(output => ({

          index,

          result: { [key]: output }

        }))

      })

    })

    .mergeAll()

    .groupBy(

      obj => obj.index,

      obj => obj.result,

      group$ => group$.skip(segments.length - 1)

    )

    .mergeMap(group$ =>

      group$.reduce(

        (obj, result) => Object.assign(obj, result),

        {}

      )

    );

​

//////////////////////////////////////////////

// Using the custom operator

//////////////////////////////////////////////

​

const result$ = input$.let(

  collate({

    key: 'a',

    work: doStuffA

  }, {

    key: 'b',

    work: doStuffB

  })

);

​

result$.subscribe(result => console.log(result));

​

//////////////////////////////////////////////

// Feeding fake input data

//////////////////////////////////////////////

​

let i = 0;

setInterval(() => {

  i++;

  input$.next({ a: `a ${i}`, b: `b ${i}` });

}, 1000);

​