Piping through to a new stream

[srcnix]

First of all, awesome library you have built!

I'm trying to wrap my head around a few things, and currently looking at being able to setup a stream, and then pipe the data from the stream into another (which may go through X number of other streams).

These streams would further filter, or do some magic, run some async code etc.

What's the best way to do this?

[amsross]

Do you have a more specific example that you could provide regarding what you're trying to do?

Barring that, here is some generic example code you might be able to extract some meaning from:

const source = _([ 1, 2, 3, 4 ])

const addOne = _.map(x => x + 1)
const filterEvens = _.filter(x => x % 2 === 0)
const asyncStuff = _.flatMap(x => _((push, next) => setTimeout(() => {
  push(null, x)
  push(null, _.nil)
}, 1000)))

const doStuff = _.pipeline(addOne, filterEvens, asyncStuff)

source
  .through(doStuff)
  .each(_.log) // this could just as well be `.pipe(someNodeStream)`