Array functions and parallelism for streams
npm install parallel-streamsAdds the functionality available to arrays to streams and allows parallelism.
There are two complementary aspects to the parallel-streams module.
The ParallelStream class supports a set of options to allow control of
processing each item in the stream in parallel.
This is particularly useful when doing disk or network IO of variable length so
that slower items don't hold up faster ones when order is unimportant.
A set of utility functions allow using much of the functionality of Arrays on streams.
It uses similar syntax and semantics to hopefully make code more readable.
const ParallelStream = require('parallel-streams.js');ParallelStream.from([0,1,2,3], {name: "Munching"}) // Create a stream called "Munching" that will output the items 0,1,2,3
.log(m=>[m], {name:"Stream 0..3"}); // 0 1 2 3 Log each item in the stream before passing it on.
// Now add a map function that squares each item in the stream
ParallelStream.from([0,1,2,3]) .map(m=>m**2) .log(m=>[m], {name:"Stream 0,1,4,9"})
// If our stream has arrays, then you can flatten them into a stream of their elements
ParallelStream.from([[0,"a"],[1,"b"]]) .log(m=>[m], {name:"Stream [0,a],[1,b]"}) .flatten() .log(m=>[m], {name:"Stream 0,a,1,b"})
// And you can filter
ParallelStream.from([0,1,2,3,4]) .filter(m=>m>1 && m<4) .log(m=>[m], {name:"Stream filter 2,3"})
// Or select uniq items
ParallelStream.from([0,1,2,2,1]) .uniq() .log(m=>[m], {name:"Stream uniq 0,1,2"})
// Or slice out a range
ParallelStream.from([0,1,2,3,4]) .slice(2,4) .log(m=>[m], {name:"Stream slice 2,3"})
// A little more complexity allows forking a stream and running two or more sets of processing on it.
let ss = ParallelStream.from([0,1]) .fork(2).streams;
ss[0].log(m=>[m], {name: "ForkA 0,1"});
ss[0].log(m=>[m], {name: "ForkB 0,1"});
// Reduce works, but note that you have to use the 'function' syntax instead of (a,b)=>(a+b) if you want to use "this" for debugging.
// The result here should be 110 as 0 is used as the initial value
ParallelStream.from([10,20,30,40]) .reduce(function(acc,d,i) { return (acc + i + d+1) }, 0, function(res) {this.debug("SUM=", res)}, { name: "Sum=110" });
// The result here should be 110 as it runs reduce 3 times, with the 10 is used as initial value.
ParallelStream.from([10,20,30,40]) .reduce(function(acc,d,i) { return (acc + i + d+1) }, undefined, function(res) {this.debug("SUM=", res)}, { name: "Sum=109" });
// Reduce with no arguments is useful at the end of a chain of streams to avoid the last stream pushing back when it can't write.
ParallelStream.from([10,20,30,40]) .reduce();
Summary
* paralleloptions - {limit, retryms, silentwait}
* options = {name, paralleloptions, parallel(data,encoding,cb), init(), flush(cb), highWaterMark, verbose, just ReportError, async}
* ParallelStream(options) -> stream: create new ParallelStream
* ps.log(f(data)=>string, options): Output debugging
* ps.map(f(data)=>obj, options); (esp options async: true); stream of modified objects
* ps.flatten(options); stream of streams to concatenated stream
* ps.filter(f(data)=>boolean, options); stream of objects where f(obj)
* ps.slice(begin, end, options) => subset of s
* ps.fork(f(ps)..., options) => Fork stream into other functions
* ps.uniq(f(data)=>string, options) => stream containing only uniq members (optional f(data) provides a uniqueness funciton)
* ps.from(arr, options) => stream from array - often first step of a pipeline
* ps.reduce(f(acc, data, index) => acc, initialAcc, cb(data), options); reduce a stream to a single value
API
####ParallelStream(options) - Create a new Parallel Stream
options = {
name Set to a name to use in debugging (this.debug will be active on parallel-streams:
paralleloptions {
limit: maximum number of threads to run in parallel
retryms: How long to wait before retrying if thread count exceeded,
silentwait: Set to true to remove debugging when waiting
},
parallel(data, encoding, cb), Function like transform(), including how to use push() and cb(err, data)
init() Called at initialization
//Inherited from TransformStream:
flush(cb) runs at completion before the stream is closed, should call cb when complete.
highWaterMark int Sets how many data items can be queued up
verbose True to get some debugging, especially around callbacks
justReportError Normally an error gets sent downstream, theoretically causing a (clean) terminate. Set this if want errors ignored.
async In .map the function is asynchronous
}
The main differences with TransformStream are:* do not pass transform() unless you intentionally are replacing the parallel behavior,
* objectMode is set to true by default.
Other functionality of TransformStream should work, but might not have been tested.
$3
Each of the following functions (unless stated) is intended to be chained,
i.e. the call returns a stream which can itself be chained. Each function takes options which are passed to the ParallelStreams constructor.
Each function defaults to having a name which is the name of the function,
but can be overwritten by setting the option
#### Piping Readable Streams into ParallelStreams
Each function (except
from())can be called either as a function on an existing ParallelStream
or as a static function e.g. if ps is a parallelstream and rs is a readable
ps.log(m=>m) or rs.pipe(ParallelStream.log(m=>m)).
This is intended to allow smooth integration with Readable, Writable & TransformStreams.Note that a common mistake is:
rs=ParallelStream.log(m=>m).reduce(); ps.pipe(rs).
This won't work because rs will be the reduce and ps will be piped there rather than to the log. #### ParallelStream.prototype.log(f(data)=>string, options={})
Log output using debug("parallel-streams:"),
f(data) should return an array suitable for passing to debug(),
i.e. the first parameter can contain formatting like %s %d %o (see npm:debug for details)Passes input to the next stream unchanged (unless f(input) has side-effects)
e.g. .log(data => ["Handling %o", data])
#### ParallelStream.prototype.map(f(data)=>obj, options={}) or (f(data,cb)=>obj, {async:true,...})
`
async If true, then cb(err,data) is called by the function when it is 'complete' rather than returning a value.
Transform input data to output data like e.g.
Or if function is async something like or .map((data, cb)=> f(data, (err,data)=>{dosomethingtodata; cb(err,newdata)})#### ParallelStream.prototype.flatten(options={})
input: stream of arrays of x
output: stream of x
Flatten a stream of arrays into a stream of items in those arrays,
useful for example where a previous map call returns a list, each element of which requires independent processing.TODO could add options as to whether should handle single objs as well as arrays and whether to ignore undefined
#### ParallelStream.prototype.filter(f(data) => boolean, options={})
f(data) => boolean Filter function that returns true for items to outputinput stream: objects
output stream: objects where f(data) returns true
Usage example: #### ParallelStream.prototype.slice(begin, end, options={})
begin: first item to pass,
end: one after last item
input stream: objects
output stream: objects[begin...end-1]
#### ParallelStream.prototype.fork(cb1 ... cbn, options={})
Fork a stream into multiple streams,
cb1..cbn f(parallelstream)
returns parallelstream
Usage of fork is slightly different
let ss = parallelstream
.fork(s=>s.log(m=m).reduce())
.filter etc
Warning all streams need to properly end e.g. with .reduce() or pushback on one fork could effect all of them.#### ParallelStream.prototype.uniq(f(data)=>string, options={}) {
Uniq allows cleaning a stream of duplicates, an optional function allows generating an id to use for duplicate checking.
f(data) => string: optional function to return a string that can be used to compare uniqueness (for example an id)
options { uniq: optional array to use for checking uniqueness (allows testing against existing list)
#### static ParallelStream.from(arr, options={})
Create a new ParallelStream from an array, usually this will be the start of any pipeline of streams.
arr Array of any kind of object
output: Elements of arr, in order.
#### ParallelStream.reduce(f(acc, data, index)=>acc, initialAcc, cb(data), options={})
Behaves like
f(acc, data, index): acc = result so far, data = next item from stream, index = index in stream
initialAcc: Initial value to acc
cb(data): Function to call after the last item is processed.
Note, as for Array.prototype.reduce(), if no initialAcc is provided, then the first item in the stream
will be used as the initial value of acc, and the reduction function will get called for the first time
using the index=1 and the second item from the stream.Note that reduce() is the only one of the above functions (other than fork) that doesnt return a ParallelStream.
This makes it suitable for ending a chain of streams to avoid the last stream pushing back. Expect to see