GroupBy -> transform -> merge

groupByTransformMerge

var groupTransform = (stream$, f, keySelector, selector) => {

  return Rx.Observable.create(observer => {

    var disposable = new Rx.CompositeDisposable()

    disposable.add(stream$.groupBy(keySelector, selector)
      .subscribe(groupObservable => {
        disposable.add(f(groupObservable, groupObservable.key).subscribe(observer))
      }))

    return disposable
  })

}

var input$ = Rx.Observable.fromArray([
  {
    key: 1,
    value: 10
  }, {
    key: 2,
    value: 20
  }, {
    key: 2,
    value: 200
  }, {
    key: 1,
    value: 10
  }, {
    key: 3,
    value: 3
  }
])

var comp$ = groupTransform(
  input$,
  (s$, key) => s$.map(v => v * 2),
  x => x.key,
  x => x.value
)

var d = comp$.subscribe(x => console.log(x))