event-stream

EventStream Streams are node's best and most misunderstood idea, and EventStream is a toolkit to make creating and working with streams easy. Normally, streams are only used for IO, but in event stream we send all kinds of objects down the pipe. If your application's input and output are streams, shouldn't the throughput be a stream too? The EventStream functions resemble the array functions, because Streams are like Arrays, but laid out in time, rather than in memory. All the event-stream functions return instances of Stream. event-stream creates 0.8 streams, which are compatible with 0.10 streams. NOTE: I shall use the term "through stream" to refer to a stream that is writable and readable. NOTE for Gulp users: Merge will not work for gulp 4. merge-stream should be used. simple example: //pretty.js if(!module.parent) { var es = require('event-stream') var inspect = require('util').inspect process.stdin //connect streams together with `pipe` .pipe(es.split()) //split stream to break on newlines .pipe(es.map(function (data, cb) { //turn this async function into a stream cb(null , inspect(JSON.parse(data))) //render it nicely })) .pipe(process.stdout) // pipe it to stdout ! } run it ... curl -sS registry.npmjs.org/event-stream | node pretty.js node Stream documentation through (write?, end?) Re-emits data synchronously. Easy way to create synchronous through streams. Pass in optional write and end methods. They will be called in the context of the stream. Use this.pause() and this.resume() to manage flow. Check this.paused to see current flow state. (write always returns !this.paused) this function is the basis for most of the synchronous streams in event-stream. es.through(function write(data) { this.emit('data', data) //this.pause() }, function end () { //optional this.emit('end') }) map (asyncFunction) Create a through stream from an asynchronous function. var es = require('event-stream') es.map(function (data, callback) { //transform data // ... callback(null, data) }) Each map MUST call the callback. It may callback with data, with an error or with no arguments, callback() drop this data. this makes the map work like filter, note:callback(null,null) is not the same, and will emit null callback(null, newData) turn data into newData callback(error) emit an error for this item. Note: if a callback is not called, map will think that it is still being processed, every call must be answered or the stream will not know when to end. Also, if the callback is called more than once, every call but the first will be ignored. mapSync (syncFunction) Same as map, but the callback is called synchronously. Based on es.through flatmapSync (syncFunction) Map elements nested. var es = require('event-stream') es.flatmapSync(function (data) { //transform data // ... return data }) filterSync (syncFunction) Filter elements. var es = require('event-stream') es.filterSync(function (data) { return data > 0 }) split (matcher) Break up a stream and reassemble it so that each line is a chunk. matcher may be a String, or a RegExp Example, read every line in a file ... fs.createReadStream(file, {flags: 'r'}) .pipe(es.split()) .pipe(es.map(function (line, cb) { //do something with the line cb(null, line) })) split takes the same arguments as string.split except it defaults to '\n' instead of ',', and the optional limit parameter is ignored. String#split NOTE - Maintaining Line Breaks If you want to process each line of the stream, transform the data, reassemble, and KEEP the line breaks the example will look like this: fs.createReadStream(file, {flags: 'r'}) .pipe(es.split(/(\r?\n)/)) .pipe(es.map(function (line, cb) { //do something with the line cb(null, line) })) This technique is mentioned in the underlying documentation for the split npm package. join (separator) Create a through stream that emits separator between each chunk, just like Array#join. (for legacy reasons, if you pass a callback instead of a string, join is a synonym for es.wait) merge (stream1,...,streamN) or merge (streamArray) concat → merge Merges streams into one and returns it. Incoming data will be emitted as soon it comes into - no ordering will be applied (for example: data1 data1 data2 data1 data2 - where data1 and data2 is data from two streams). Counts how many streams were passed to it and emits end only when all streams emitted end. es.merge( process.stdout, process.stderr ).pipe(fs.createWriteStream('output.log')); It can also take an Array of streams as input like this: es.merge([ fs.createReadStream('input1.txt'), fs.createReadStream('input2.txt') ]).pipe(fs.createWriteStream('output.log')); replace (from, to) Replace all occurrences of from with to. from may be a String or a RegExp. Works just like string.split(from).join(to), but streaming. parse Convenience function for parsing JSON chunks. For newline separated JSON, use with es.split. By default it logs parsing errors by console.error; for another behaviour, transforms created by es.parse({error: true}) will emit error events for exceptions thrown from JSON.parse, unmodified. fs.createReadStream(filename) .pipe(es.split()) //defaults to lines. .pipe(es.parse()) stringify convert javascript objects into lines of text. The text will have whitespace escaped and have a \n appended, so it will be compatible with es.parse objectStream .pipe(es.stringify()) .pipe(fs.createWriteStream(filename)) readable (asyncFunction) create a readable stream (that respects pause) from an async function. while the stream is not paused, the function will be polled with (count, callback), and this will be the readable stream. es.readable(function (count, callback) { if(streamHasEnded) return this.emit('end') //... this.emit('data', data) //use this way to emit multiple chunks per call