Streams

James Bunt @unkillbob

Orion Health

Streams?

"We should have some ways of connecting programs like garden hose--screw in another segment when it becomes necessary to massage data in another way. This is the way of IO also."

Doug McIlroy. October 11, 1964

Wikipedia says:

"...a sequence of data elements made available over time."

"...a conveyor belt that allows items to be processed one at a time..."

Production Line

But why?

Using streams encourages...

Breaking it down

small, generic, single purpose modules

Performance

var http = require('http'),
    fs = require('fs');

var pirateBay = http.createServer(function(req, res) {
    fs.readFile('photoshop.dmg', function(err, data) {
        res.end(data);
    });
});

pirateBay.listen(8080);

Parallelism

var http = require('http'),
    fs = require('fs');

var pirateBay = http.createServer(function(req, res) {
    fs.createReadStream('photoshop.dmg').pipe(res);
});

pirateBay.listen(8080);

require('stream')

Introduced in v0.4

Great! But...

data: coming ready or not!

can't consume X bytes or put data back

hard to implement, not DRY

"Streams2" added in v0.10

5 types

Readable, Writable, Transform, Duplex, "Classic"

Readable Streams

Can be piped from but not to

process.stdin

fs.createReadStream('/path/to/file.txt', {
    encoding: 'utf8'
})

Writable Streams

Can be piped to but not from

process.stdout

fs.createWriteStream('/path/to/file.txt', {
    encoding: 'utf8'
})

Transform Streams

Transform input and produce output

var zlib = require('zlib');
var gzip = zlib.createGzip(); // readable & writable

Duplex Streams

Bi-directional

var net = require('net');
var client = net.connect({ port: 8124 }, function() {
    // ...
});

Classic Streams

Old interface from Node 0.4

process.stdin.on('data', function(buf) {
    console.log(buf);
});

Triggered whenever a stream has a data listener registered

How do I use streams?

Reading

readableStream.on('readable', function() {
    var data = readable.read();
});

read() returns null when data is finished

Writing

writableStream.write(someData);
writableStream.write(someMoreData);

writableStream.end(/*data*/);

Pipes!

pipe()

Readable ==> Writable

Manages back-pressure for you

readable.pipe(writable) // returns writable

readable.pipe(transform1)
    .pipe(transform2)
    // ...
    .pipe(writable);

Modules!

from

Easily create readable streams

var from = require('from');

var readData = from(function getChunk(count, next) {
    this.emit('data', someData);

    if (finished) { this.emit('end'); }

    next(); // or just `return true;` if sync
});

readData.pipe(writable);

through

Synchronous transform streams

var through = require('through');

var transform = through(function write(data) {
    this.queue(data);
}, function end() {
    this.queue(null);
});

readable.pipe(transform).pipe(writable);

map-stream

Asynchronous transform streams

var map = require('map-stream');

var transform = map(function(data, callback) {
    // transform
    callback(null, newData);
    // emit 'error' event
    callback(err);
    // drop data (filter)
    callback();
});

readable.pipe(transform).pipe(writable);

concat-stream

Concatenate stream data

var concat = require('concat-stream');

var gatherData = concat(function(allTheDatas) {
    // concatenated Buffer/String or an array
});

readable.pipe(gatherData);

Highland.js

_ + async for streams

each, map, filter, series, parallel, etc

http://highlandjs.org/

When would I use streams?

Web

var http = require('http');

http.createServer(function(req, res) {
    // req: Readable stream
    // res: Writable stream
});

var net = require('net');
var client = net.connect({ port: 8124 }, function() {
    // ...
});

I/O

Databasing

Caching

Load balancing

Web servicing

Large Files

fs.createReadStream('log.txt')
    .pipe(split())
    .pipe(processEachLineOfLog());

Control Flow

@caolan

Anything!

gulp.js

The streaming build system

So?

Fast!

In memory transforms

Code over configuration

Plugins!

gulpjs.com/plugins/

Setting it all up

$ npm install -g gulp

$ npm install --save-dev gulp gulp-util

$ atom gulpfile.js

var gulp = require('gulp');
gulp.task('default', function() {
    // build all the things
});

$ gulp

gulpfile.js

var gulp = require('gulp'),
    jshint = require('gulp-jshint'),
    uglify = require('gulp-uglify'),
    concat = require('gulp-concat'),

    through = require('through'),
    map = require('map-stream'),

    growl = require('growl'),
    // ...

gulpfile.js cont'd

gulp.task('scripts', ['lint'], function(done) {
    // return a stream, promise or call done

    return gulp.src('./scripts/**/*.js') // readable

        .pipe(uglify())
        .pipe(concat('all.js'))

        .pipe(gulp.dest('./dist')); // through stream
});

gulp.task('default', ['lint', 'test'], function() {
    gulp.watch('{scripts,tests}/**', ['lint', 'test']);
});

Tell me more!

Go on a Stream Adventure!

nodeschool.io/#stream-adventure

Read the Stream Handbook

github.com/substack/stream-handbook

by @substack

Read the API docs

nodejs.org/api/stream.html

gulpjs.com

Build Wars

by @markdalgleish

Image Credits

Fine Art - Garden Within, Forest Stream

Donut Production Line

Breakdancing

Synchronised Dance

Super Mario

Big Bang Theory Ball Pit

Mind Blown

We're Hiring!

Orion Health

www.orionhealth.com/nz/careers/