Security News
tea.xyz Spam Plagues npm and RubyGems Package Registries
Tea.xyz, a crypto project aimed at rewarding open source contributions, is once again facing backlash due to an influx of spam packages flooding public package registries.
clusterstream
Advanced tools
Readme
Clusterstream is a duplex stream that distributes any incoming data to a pool of workers and pipes down the output of the workers. Clusterstream uses the stdin/stdout to manage backpressure: when a worker buffer is full the worker is no longer considered available until it emits a drain
event. When no workers are available the Clusterstream will pause until any worker becomes available again.
Clusterstream([options],fn)
Each worker is initialized with a supplied function and executes it on the incoming stream. The first argument of the function will be the incoming stream (sharing all data passed to that worker) and the function itself has to return a stream as well. The returned stream is then piped back to the parent.
The function argument can either be a pure javascript function or a full path to a module containing a function. The function will be executed in the context of the child process, so it will not have access to any variables on the parent scope. So if you need any required packages you will have to explicitly require them inside the function, even if they are available in parent. The function will receive a stream as the first argument and will have to return a stream (that is piped back to the parent).
Options are optional (can be omitted completely):
By default, Clusterstream creates as many workers as there are cpus. If options is a number (not an object), then that number will determine the number of workers created.
Here is a quick and dirty example where each worker waits 500ms before returning each object:
const Stream = require('stream');
const Clusterstream = require('../index');
const feeder = Stream.PassThrough({objectMode:true});
[...Array(100)].forEach( (d,i) => feeder.write({data:i}));
feeder.end();
feeder.pipe(Clusterstream(incoming => {
const Stream = require('stream');
return incoming.pipe(Stream.Transform({
objectMode: true,
transform: (d,e,cb) => setTimeout(() =>
cb(null,{data:d.data,worker:incoming.workerId}),
500
)
}));
}))
.on('error',console.log)
.pipe(Stream.Transform({
objectMode: true,
transform: (d,e,cb) => {
console.log('received',d);
cb();
}
}));
Same example using etl
instead of Stream.Transform
:
const etl = require('etl');
const Clusterstream = require('../index');
etl.toStream([...Array(100)].map( (d,i) => ({data:i})))
.pipe(Clusterstream(incoming => incoming
.pipe(require('etl').map(d =>
require('bluebird').delay(500,{data: d.data,worker:incoming.workerId})
))
))
.pipe(etl.map(d => console.log('received',d)));
Here is an example of how global variables can be passed into 10 workers:
const etl = require('etl');
const Clusterstream = require('../index');
const multiplier = 2;
const options = {
children: 10,
global: {multiplier},
require: ['etl','bluebird']
};
etl.toStream([...Array(100)].map( (d,i) => ({data:i})))
.pipe(Clusterstream(options, incoming => incoming
.pipe(etl.map(d =>
bluebird.delay(500,{data: d.data* multiplier,worker:incoming.workerId})
))
))
.pipe(etl.map(d => console.log('received',d)));
Clusterstream.map([options],fn)
Clusterstream.map
is a simple wrapper where you supply a transform function
that receives each packet as an argument and returns the transformed values
(instead of stream input/output)
FAQs
Piping streams across a cluster of Workers
The npm package clusterstream receives a total of 9 weekly downloads. As such, clusterstream popularity was classified as not popular.
We found that clusterstream demonstrated a not healthy version release cadence and project activity because the last version was released a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Tea.xyz, a crypto project aimed at rewarding open source contributions, is once again facing backlash due to an influx of spam packages flooding public package registries.
Security News
As cyber threats become more autonomous, AI-powered defenses are crucial for businesses to stay ahead of attackers who can exploit software vulnerabilities at scale.
Security News
UnitedHealth Group disclosed that the ransomware attack on Change Healthcare compromised protected health information for millions in the U.S., with estimated costs to the company expected to reach $1 billion.