- Motivation
- Basic usage
- Reference
- merge
stream1.merge(stream2) - collect
stream.collect() - filter
stream.filter(function) - chunk
stream.chunk(length) - map
stream.map(function) - omit
stream.omit(field1, field2, ...) - pick
stream.pick(field1, field2, ...) - pluck
stream.pluck(field); - uniq
stream.uniq(field) - weakSort
stream.weakSort(sortFunction, [bufferMinSize=75], [bufferMaxSize=100]) - onSeries
stream.onSeries(async function, [concurrency=1])
- merge
This package simplifies transformations and filtering of NodeJS object streams. Think about it as Underscore.js for streams.
The basic use case I faced many times was a transformation of a large number of JSON objects that are finally stored in some database. Transformation is the quick part but then you have to then chunk data in size allowed by your database to limit the number of queries and control the flow of the whole stream based on how fast you are able to save the transformed data.
const { Readable } = require('stream');
const Flowage = require('flowage');
// Let's have some stream that will output a series of objects { n: 0 }, { n: 1 }, { n: 2 }, { n: 3 }, ...
const readable = new Readable({ objectMode: true });
let n = 0;
setInterval(() => readable.push({ n: n++ }), 1000);
// Pipe it thru Flowage() to get stream extended by helper methods.
const flowage = readable.pipe(new Flowage());
// Split the stream into a stream of odd objects and even objects and extend them with some field is='odd' or is='even'.
const oddStream = flowage
.filter(obj => obj.n % 2)
.map(obj => Object.assign({}, obj, { is: 'odd' }));
const evenStream = flowage
.filter(obj => obj.n % 2 === 0)
.map(obj => Object.assign({}, obj, { is: 'even' }));
// Then merge them back.
const mergedStream = oddStream.merge(evenStream);
// Chunk them by 100 records.
const chunkedStream = mergedStream.chunk(100);
// Save them to MongoDB in batches of 100 items with concurrency 2.
// This also corks the stream everytime the period when max concurrency is reached.
chunkedStream.onSeries(async (arrayOf100Items) => {
await datase.collection('test').insert(arrayOf100Items);
}, { concurrency: 2 });Returns stream containing values merged from 2 given streams. Merged stream ends when both streams ends.
const mergedStream = stream1.merge(stream2);Returns Promise that gets resolved when stream ends to an array of all the values.
const data = await stream.collect();Returns stream containing filtered values.
// Filter out even items from stream.
const filteredStream = stream.filter(val => val.index % 2 === 0);Returns stream where each item is an array given number of items from original stream.
// Chunk values into arrays of 10 items.
const chunkedStream = stream.chunk(10);Returns stream where original items are transformed using given function.
// Extend each object in the stream with `.foo = 'bar'` field.
const mappedStream = stream.map(val => Object.assign({}, val, { foo: 'bar' }));Returns stream where given fields where omitted.
// Omit field1 and field2 from stream objects.
const resultingStream = stream.omit('field1', 'field2');Returns stream where each item contains only the given fields.
// Pick only field1 and field2 from stream objects.
const resultingStream = stream.pick('field1', 'field2');Returns stream with given field picked from each item.
// Pick only field1 and field2 from stream objects.
const resultingStream = stream.pluck('field1');Returns stream containing only unique items based on given field. You need enough memory to keep a set of all unique values hashed using sha256.
// Filter unique items based on id field.
const uniquesStream = stream.uniq('id');Returns stream containing values sorted using given function and floating buffer of a given size.
This method is helpful when only a few neighboring items may have the wrong order. This may happen for example when a client is pushing data into the storage via API with concurrency higher than 1 and the quests reach the server in the wrong order. Or the API has multiple redundant instances that may process the incoming requests with different speed.
This method uses a buffer for streamed items. Every time the buffer reaches bufferMaxSize gets
sorted and bufferMaxSize - bufferMinSize items are outputted to the stream.
const sortFunction = (a, b) => a.index < b.index ? -1 : 1;
const sortedStream = stream.sort(sortFunction, 75, 100);Returns a promise that gets resolved when given function gets finished for the last item of the stream.
Everytime the given concurrency is reached it pauses the stream.
// Store items in MongoDB with concurrency 10.
await stream.onSeries(async (item) => {
await database.collection('items').insert(item);
}, 10);