stream-operators
A library for creating Readable, Writable and Transform stream which adopt two major tools in async world - Stream and Promise - with declarative programming approach.
Node.js native stream is very powerfully tool for creating high performance application. This library designed for using the power of stream for read, transform and write streams in object mode. the API is very declarative and inspired by RxJs. another design decision is combining streams with promise for better async handling; Most of the helpers could return promise instead of absolute data.
Installation
$ npm install stream-operators
Read stream creators
counter
counter:Readable
Creates a read stream that start a counter from zero to limit - the limit default value is Infinity.
sample
; counter5 .pipeprocess.stdout // console 0 1 2 3 4
from
from:Readable
Creates a read stream from a list and push items in sequence.
sample
; from .pipeprocess.stdout // console: 1 2 3 4
read
read:Readable
Creates a read stream from a read function, each time consumer send a read signal the reader function is called and the return value push to the stream. The return value could be a promise and in this case the resolved value push to the stream.
sample
readfetchProductspage.thenlist.mapp.name.join"\n" .pipeprocess.stdout // console apple\n orange\n banana
Write stream creators
write
write:Writable
Creates a write stream from a write function, each time a chunk is prepared this function is called and when the function execution finishes ( in case of promise if the promise resolved) the read signal send to top reader stream.
sample
counter5 .pipewriteconsole.log // console 0 1 2 3 4 readFileLineStream .pipewritewriteToDbmapToProductline
fork
fork:Writable
forks N read stream from a main read stream. if the callback function return a write stream the finish event fire after branches finish event.
sample
req .pipeforkstream.piperes, // echo request to responsestream.pipewriteconsole.log, // log request to console
Transform stream creators
map
map:Transform
Creates a transform stream from a map function. for each chunk the the result of map function ( the resolved value in case of promise) push to stream.
sample
; counter5 .pipemapitem * 2 .pipelogger // console: 0 2 4 6 8 readProductCsvFile .pipemapmapToJson .pipestartWith"[" .pipeendWith"]" .piperes // response [ {id:1, name:"apple"}, {id:2, name:"orange"}, ... ]
filter
filter:Transform
Creates a transform stream from a map function. for each chunk if the result of filter be true (or it resolve to truthy value in case of promise) the chunk push to stream.
sample
;; counter5 .pipefilterisEven .pipelogger // console: 0 2 4
delay
delay:Transform
Creates a transform stream that made delay on the input stream. the time unit is ms
.
sample
;; counter5 .pipedelay1000 .pipefilterisEven .pipelogger // console: 0 2 4
buffer
buffer:Transform
Creates a transform stream that buffer N chunk and pushes this buffer to output stream. the bufferSize default value is Infinity - so the output stream push only one chunk when the input stream ends.
sample
; counter5 .pipebuffer2 .pipelogger // console: [0,1] [2,3] [4] // writing a json stream data to db - each 1000 items insert in one bulk insert query for better performancereadEnormousJson .pipemapmapToStandardProduct .pipebuffer1000 .pipewriteProductListToDB
startWith
startWith:Transform
Creates a transform stream that push a chunk with input value as first chunk.
sample
; counter5 .pipestartWith"start" .pipelogger // console: start 0 1 2 3 4 // read product list from DB and convert it to CSV first csv line is "id,name,price" and send stream as responsereadProductFromDb .pipemapproductToCSVLine .pipestartWith"id,name,price\n" .piperes
endWith
endWith:Transform
Creates a transform stream that push a chunk with input value as last chunk.
sample
; counter5 .pipestartWith"start" .pipeendWith"end" .pipelogger // console: start 0 1 2 3 4 end readProductCsvFile .pipemapmapToJson .pipestartWith"[" .pipeendWith"]" .piperes // response [ {id:1, name:"apple"}, {id:2, name:"orange"}, ... ]
extract
extract:Transform
Create a transform stream that push item of a chunk (if chunk is am array or has forEach function) to output stream.
; from .pipeextract .pipelogger // console: 1 2 3 4 5 6 7 // read each 1000 records in one query from DB then push array to stream the extract function extract the list for normalizing data and the rest of stream chain work with simple data instead of a list.readreadProductFromDB .pipeextract .pipetoJsonStream .piperes
reduce
reduce:Transform
Create a transform stream that reduce N chunk to one chunk. the N - bufferSize - default value is Infinity, so it reduce all chunk to one chunk by default. the initial value is reset value (or the output of reset if the reset is a function - the reset function give a times (chunk index) and return reset value ).
sample
; counter5 .pipereduceacc + sum , 0 .pipelogger // console 10 counter5 .pipereduceacc + sum , 0, 3 .pipelogger // console 3 7 // implementing buffer transform stream with reduce helper.
inspect
inspect:Transform
Create a transform stream that log each chunk to console without any change to stream. this function is useful in development and debug purpose.
counter5 .pipeinspect .pipewritevoid 0 // console: 0\n 1\n 2\n 3\n 4\n