Flow.io

Collection of Node.js streams.

Installation

$ npm install flow.io

Usage

To use flow,

varflow=require('flow.io');

The flow library is comprised of several smaller modules. If you want to roll your own flow, follow the links and import the individual modules.

The flow library includes the following stream factories...

Files

flow.read()

File readStream factory.

varstream=flow.read().path('path/to/file').stream(clbk);stream.pipe(process.stdout);

flow.write()

File writeStream factory.

varstream=flow.write().path('path/to/destination').stream(clbk);readStream.pipe(stream);

JSON

flow.parse()

Transform stream factory to parse JSON. Wraps JSONStream's parse stream.

varrStream=flow.read().path('path/to/file.json').stream();varpStream=flow.parse().stream();rStream.pipe(pStream).pipe(/* writable stream*/);

flow.stringify()

Transform stream factory to stringify JSON. Wraps JSONStream's stringify stream.

varrStream=flow.read().path('path/to/file.json').stream();varpStream=flow.parse().stream();varsStream=flow.stringify().stream();varwStream=flow.write().path('path/to/destination.json').stream();rStream.pipe(pStream).pipe(sStream).pipe(wStream);

Arrays

flow.array()

Transform stream factory to convert an array into an element-by-element readable stream. Similar to event-stream readArray, except a transform stream. Useful when using a sink stream which generates an array and where downstream streams in a stream pipeline require individual data elements.

flow.chunkify()

Transform stream factory to chunk streamed data values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([1,1,1,2,2,2,3,3,3]);varstream=flow.chunkify().numValues(3).stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.sinkandstream()

Transform stream factory to sink a specified number of streamed data values and then stream new data values as they arrive.

varreadArray=require('event-stream').readArray;varreadStream=readArray([1,1,1,2,3,4,5]);varstream=flow.sinkandstream().numValues(3).stream();readStream.pipe(stream).pipe(/* writable stream */);

Map

flow.map()

Transform stream factory to map a data value to another data value via a transformation function. Wraps event-stream's map stream.

Reduce

flow.reduce()

Transform stream factory to perform a data reduction.

Mathematics

flow.abs()

Transform stream factory to map numeric data stream values to their absolute values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([-1,-1,0,1,1]);varstream=flow.abs().stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.round()

Transform stream factory to round numeric data stream values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([3.245,-0.9845,-20.446,0.5,1.0,10]);varstream=flow.round().stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.floor()

Transform stream factory to floor numeric data stream values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([3.245,-0.9845,-20.446,0.5,1.0,10,5.9999]);varstream=flow.floor().stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.ceil()

Transform stream factory to round numeric data stream values toward positive infinity.

varreadArray=require('event-stream').readArray;varreadStream=readArray([3.245,-0.9845,-20.446,0.5,1.0,10,5.9999]);varstream=flow.ceil().stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.add()

Transform stream factory to increment streamed numeric data values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([1,0,1,1,0]);varstream=flow.add().add(100).stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.subtract()

Transform stream factory to decrement streamed numeric data values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([1,0,1,1,0]);varstream=flow.subtract().subtract(1).stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.multiply()

Transform stream factory to perform scalar multiplication on streamed numeric data values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([12,8,34,512,72]);varstream=flow.multiply().scalar(10).stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.divide()

Transform stream factory to perform scalar division on streamed numeric data values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([12,8,34,512,72]);varstream=flow.divide().divisor(10).stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.pow()

Transform stream factory to exponentiate numeric data stream values according to a specified power.

varreadArray=require('event-stream').readArray;varreadStream=readArray([2,4,3,5,7]);varstream=flow.pow().exponent(3).stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.exp()

Transform stream factory to evaluate an exponential function for each numeric data value.

varreadArray=require('event-stream').readArray;varreadStream=readArray([1.453,0,9.504,102.2,1]);varstream=flow.exp().stream();readStream.pipe(stream).pipe(/* writable stream */);

flow.diff()

Transform stream factory to calculate the difference between successive streamed data values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([10,8,22,100,1]);varstream=flow.diff().stream();readStream.pipe(stream).pipe(/* writable stream */);

Statistics

flow.count()

Reduce stream factory to count the number of streamed data elements and stream the result.

flow.min()

Reduce stream factory to find a numeric data stream's minimum value.

flow.mmin()

Transform stream factory to find sliding-window minimum values (moving min) over a numeric data stream.

varreadArray=require('event-stream').readArray;varreadStream=readArray([1,0,5,2,3,6,8,1,0]);varstream=flow.mmin().window(3).stream();readStream.pipe(stream).pipe(/* writable stream*/);

flow.max()

Reduce stream factory to find a numeric data stream's maximum value.

flow.mmax()

Transform stream factory to find sliding-window maximum values (moving max) over a numeric data stream.

varreadArray=require('event-stream').readArray;varreadStream=readArray([1,0,5,2,3,6,8,1,0]);varstream=flow.mmax().window(3).stream();readStream.pipe(stream).pipe(/* writable stream*/);

flow.sum()

Reduce stream factory to calculate a numeric data stream's sum.

flow.msum()

Transform stream factory to compute a sliding-window sum over a numeric data stream.

flow.csum()

Transform stream factory to compute the cumulative sum over a numeric data stream.

flow.median()

Reduce stream factory to find a numeric data stream's median value.

flow.quantiles()

Reduce stream factory to compute quantiles from a numeric data stream.

flow.iqr()

Reduce stream factory to compute the inter-quartile range from a numeric data.

flow.histc()

Reduce stream factory to compute a histogram over a numeric data stream.

flow.mean()

Reduce stream factory to calculate a numeric data stream's mean.

flow.mmean()

Transform stream factory to compute a sliding-window average over a numeric data stream.

flow.cmean()

Transform stream factory to calculate arithmetic means for streamed data arrays (chunks).

varreadArray=require('event-stream').readArray;varreadStream=readArray([1,2,1,2,2,1,4,3,5.5]);varcStream=flow.chunkify().numValues(3).stream();varmStream=flow.cmean().stream();readStream.pipe(cStream).pipe(mStream).pipe(/* writable stream */);

flow.variance()

Reduce stream factory to calculate a numeric data stream's variance.

flow.covariance()

Reduce stream factory to calculate the covariance between data elements in a numeric data stream.

flow.pcc()

Reduce stream factory to compute the Pearson product-moment correlation coefficient between data elements in a numeric data stream.

flow.skewness()

Reduce stream factory to calculate the sample skewness of streamed data values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([82,34,45,56,56,71]);varsStream=flow.skewness().stream();readStream.pipe(sStream).pipe(/* writable stream */);

flow.kurtosis()

Reduce stream factory to calculate the sample excess kurtosis of streamed data values.

varreadArray=require('event-stream').readArray;varreadStream=readArray([82,34,45,56,56,71]);varkStream=flow.kurtosis().stream();readStream.pipe(kStream).pipe(/* writable stream */);

Filters

flow.find()

Filter stream factory which finds stream values matching user-defined criteria.

flow.nans()

Filter stream factory which removes any stream values which are not numeric.

Mock

flow.mockWrite()

Provides a mock source for writable streams.

Note: implemented as a [classic stream](classic stream).

vareventStream=require('event-stream');// Simulate some data:vardata=newArray(100);for(vari=0;i<data.length;i++){data[i]=Math.random()*100;}// Create a writable stream:varwritable=eventStream.map(function(d,clbk){clbk(null,d.toString()+'\n');});// Pipe to standard out:writable.pipe(process.stdout);// Start streaming...mock(data,writable);

flow.mockRead()

Provides a mock stream sink for readable streams.

Note: implemented as a [classic stream](classic stream).

vareventStream=require('event-stream');// Simulate some data:vardata=newArray(100);for(vari=0;i<data.length;i++){data[i]=Math.random()*100;}// Create a readable stream:varreadStream=eventStream.readArray(data);// Start streaming...mock(readStream,onEnd);functiononEnd(error,data){console.log(JSON.stringify(data));}

---

Tests

Unit

Unit tests use the Mocha test framework with Chai assertions. To run the tests, execute the following command in the top-level application directory:

$ make test

All new feature development should have corresponding unit tests to validate correct functionality.

Test Coverage

This repository uses Istanbul as its code coverage tool. To generate a test coverage report, execute the following command in the top-level application directory:

$ make test-cov

Istanbul creates a ./reports/coverage directory. To access an HTML version of the report,

$ open reports/coverage/lcov-report/index.html

---

License

MIT license.

---

Copyright

Copyright © 2014. Athan Reines.