node pipe vs rx.js

rx-and-pipe.js

// Node streams
import through from 'through2'
function doSomethingToStream (config) {
  
  return through((chunk, enc, next) {
    
    next(/* do somethig here */)
  })
}

// VS

// Rx.js
function doSomethingToObservable (input$) {
  
  return input$.map(/* do somethign */)
}

Maybe something such as this where you process each chunk and transform it? Chances are each source may be an Observable so it will just naturally work.

function processInput(data) {
  return Rx.Observable.create(observer => {
    // Do any data retrieval, etc
    const newData = process(data);
    observer.onNext(newData)
  });
}

var transformed = source.flatMap(data => {
  return processInput(data);
});