Living the Stream Dream

By Ryan Ramage / @ryan_ramage_

I am ryan ramage, some may know me as ryan_ramage_ on twitter I work at Redman Technologies (formerly with Sean, and with fellow presenter tonight Nick) I am a CouchDB committer Full stack javascript'r I have a firefox phone

I will talk tonight on 'Living the steam dream'. A high level preso on streams, why they are good, and some helpful tools

INTERACTIVE LIVE ACTION DEMO

Just remember:

PEOPLE == DATA

We are going to slow down and magnify a typical request/response session Need three volunteers Browser, Node, DB Small huddle to prepare for non-stream request/response

Typical Search Pattern

Browser is going to request something - "Who is hungry?"

PLEASE WAIT

While everything is happening, the famous loading icon Explain the client who LOVED the loading icon and wanted it BIGGER Execute the query, make sure people stay in spots in each area

Result: ONE BIG TABLE

thing1 thing2 thing3 1 2 3 1 2 3 1 2 3 1 2 3 1 2 3 1 2 3 1 2 3 1 2 3

After all that. We have a nice table for the user. They begin to scan it top to bottom. Lets try and improve the experience.

Better Search Pattern

1 2 3 1 2 3 1 2 3 1 2 3 1 2 3 1 2 3 1 2 3 1 2 3

Sit hungry and non hungry people ordered in chairs People no longer wait. Have processors fire to next asap

What we learned

Streaming decrease user idle time. [time]
Streaming can lower memory requirements [space]

The user is able to process the information in their brain as soon as it comes. We did not have to hold people together. They could leave as soon as ready.

GTSTTUASAP principle

Get the shituff to the user as fast as possible

Because Speed is King in applications

Streams Are Good Abstractions

Garden hose

Dealing with disk/network/io/processor variability

Keep memory spikes down

What is the stream dream?

Streaming support across your stack

Streaming Transport Level
Streaming Browser Libs
Streaming Middleware
Streaming DB
Streaming Build System (optional)

Streaming Transport Level

Notice this scenario. Google, the huge, resource heavy bohemoth is blasting data your pidly little phone. Through lots of internet magic, your phone does not just erupt into flames in your hand.

TCP

The internet was built for streams.

TCP uses an end-to-end flow control protocol to avoid having the sender send data too fast for the TCP receiver to receive and process it reliably.

-Wikipedia on TCP Flow control

The Stream Dream?

Streaming support across your stack

Streaming Transport Level
Streaming Browser Libs
Streaming Middleware
Streaming DB
Streaming Build System (optional)

Oboe.js

Wrapping xhr with a progressive streaming interface A JSON parser that sits somewhere between SAX and DOM.

Oboe.js

Dealing with failure mid request/response

Sarah is sitting on a train using her mobile phone to check her email. The phone has almost finished downloading her inbox when her train goes under a tunnel. Luckily, her webmail developers used Oboe.js so instead of the request failing she can still read most of her emails. When the connection comes back again later the webapp is smart enough to just re-request the part that failed.

Oboe.js

Incremental loading, and browser caching

Arnold has a query that is many-dimensional which sometimes takes a long time. To speed things up, each result can be streamed and displayed as soon as it is found. Since Oboe isn't true streaming it plays nice with the browser cache. Later, Arnold revisits the same query page. he see the same results instantly from cache.

Oboe.js

Easy JSON slice and dice

Oboe.js provides a neat way to route different parts of a json response to different parts of his application. One less bit to write.

GET /myapp/things

{
   "foods": [
      {"name":"aubergine",    "colour":"purple"},
      {"name":"apple",        "colour":"red"},
      {"name":"nuts",         "colour":"brown"}
   ],
   "badThings": [
      {"name":"poison",       "colour":"pink"},
      {"name":"broken_glass", "colour":"green"}
   ]
}

MyApp.showSpinner('#foods');

oboe('/myapp/things')
   .node('foods.*', function( foodThing ){
      $.templateThing.append( foodThing.name + ' is ' + foodThing.colour );
   })
   .node('badThings.*', function( badThing ){
      console.log( 'Danger! stay away from ' + badThings.name );
   })
   .done( function(things){
	   	MyApp.hideSpinner('#foods');
      console.log( 'there are ' + things.foods.length + ' things you can eat ' +
                   'and ' + things.nonFoods.length + ' that you shouldn\'t.' );
   });

Duck Typing

oboe('/myapp/things')
   .node('{name colour}', function( foodObject ) {
      // I'll get called for every object found that
      // has both a name and a colour
   };

Shoe

Streaming between node and the browser

The Stream Dream?

Streaming support across your stack

Streaming Transport Level
Streaming Browser Libs
Streaming Middleware
Streaming DB
Streaming Build System (optional)

nodejs

"Streams in node are one of the rare occasions when doing something the fast way is actually easier. SO USE THEM. not since bash has streaming been introduced into a high level language as nicely as it is in node."

-@dominictarr in his high level node style guide

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

var server = http.createServer(function (req, res) {
    fs.readFile(__dirname + '/movie.mp4', function (err, data) {
        res.end(data);
    });
});
server.listen(8000);

BAD

This code works but buffers up the entire data.txt file into memory for every request before writing the result back to clients. If movie.mp4 is very large, your program could start eating a lot of memory as it serves lots of users concurrently, particularly for users on slow connections

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

var server = http.createServer(function (req, res) {
    var stream = fs.createReadStream(__dirname + '/movie.mp4');
    stream.pipe(res);
});
server.listen(8000);

Good

Here .pipe() takes care of listening for 'data' and 'end' events from the fs.createReadStream(). This code is not only cleaner, but now the movie.mp4 file will be written to clients one chunk at a time immediately as they are received from the disk. Using .pipe() has other benefits too, like handling backpressure automatically so that node won't buffer chunks into memory needlessly when the remote client is on a really slow or high-latency connection.

Copy A file

var fs = require("fs");

// Read File
fs.createReadStream("input/people.json")
    // Write File
    .pipe(fs.createWriteStream("output/people.json"));

pipe

All the different types of streams use .pipe() to pair inputs with outputs.

.pipe() is just a function that takes a readable source stream src and hooks the output to a destination writable stream dst:

src.pipe(dst)

pipe

a.pipe(b);
b.pipe(c);
c.pipe(d);

.pipe(dst) returns dst so that you can chain together multiple .pipe() calls together, which is the same as:

a.pipe(b).pipe(c).pipe(d)

This is very much like what you might do on the command-line to pipe programs together:

a | b | c | d

Un-Gzipping a File

var fs = require("fs");
var zlib = require("zlib");

// Read File
fs.createReadStream("input/people.csv.gz")
    // Un-Gzip
    .pipe(zlib.createGunzip())
    // Write File
    .pipe(fs.createWriteStream("output/people.csv"));

Node Modules For Streaming

Advice: Watch for > 1 year inactivity. Streams have changed a lot

Look for trusted names in the stream eco-system

dominictarr
substack
maxogden
juliangruber
raynos

Node Modules For Streaming

mikeal/request

You can stream any response to a file stream.

request('http://google.com/doodle.png').pipe(fs.createWriteStream('doodle.png'))

Node Modules For Streaming

dominictarr/event-stream

The EventStream functions resemble the array functions, because Streams are like Arrays, but laid out in time, rather than in memory.

Example of event stream

var inspect = require('util').inspect
var es = require('event-stream')    //load event-stream

es.pipe(                            //pipe joins streams together
  process.openStdin(),              //open stdin
  es.split(),                       //split stream to break on newlines
  es.map(function (data, callback) {//turn this async function into a stream
    var j
    try {
      j = JSON.parse(data)          //try to parse input into json
    } catch (err) {
      return callback(null, data)   //if it fails just pass it anyway
    }
    callback(null, inspect(j))      //render it nicely
  }),
  process.stdout                    // pipe it to stdout !
  )


// curl -sS registry.npmjs.org/event-stream | node pretty.js
//

Stream Adventure

> npm install -g stream-adventure
> stream-adventure

The Stream Dream?

Streaming support across your stack

Streaming Transport Level
Streaming Browser Libs
Streaming Middleware
Streaming DB
Streaming Build System (optional)

CouchDB

of course

Streamable HTTP

request('http://localhost:5984/db/_design/app/_view/things_by_date').pipe(resp);

Filter with node!

var filter = function(data, emit) {
      data.forEach(function(db){
          if (db.indexOf('dumb_user-') === 0) {
              emit('"' +  strip_prefix(db) + '"');
          }
      });
  }
  var filter_through = new FilterThrough(filter);
  request('http://localhost:5984/_all_dbs')
	  .pipe(filter_through)
	  .pipe(resp);

Continuous Changes

00:00: > curl -X GET "http://localhost:5984/db/_changes?feed=continuous&since=3&filter=app/important"
{"seq":4,"id":"test4","changes":[{"rev":"1-02c6b758b08360abefc383d74ed5973d"}]}
{"seq":5,"id":"test5","changes":[{"rev":"1-02c6b758b08360abefc383d74ed5973d"}]}

Note that the continuous changes API result doesn’t include a wrapping JSON object it includes only a raw line per notification. Also note that the lines are no longer separated by a comma. Note the since, and filter. Ways to slice what you want to watch for

rvagg/node-levelup

Everything streams

var levelup = require('levelup');
var srcdb = levelup('./srcdb');
var dstdb = levelup('./destdb');

srcdb.put('name', 'LevelUP');

srcdb.createReadStream().pipe(dstdb.createWriteStream()).on('close', onDone)

rvagg/node-levelup

Start/End keys

srcdb.createReadStream({
	start: 'n',
	end: 'k'
}).pipe(resp)

The Stream Dream?

Streaming support across your stack

Streaming Transport Level
Streaming Browser Libs
Streaming Middleware
Streaming DB
Streaming Build System (optional)

Gulp - Streaming builds

Sample Gulpfile

								var scriptFiles = './src/**/*.js';

gulp.task('compile', function(){
  // concat all scripts, minify, and output
  gulp.src(scriptFiles)
    .pipe(concat({fileName: pkg.name+".js"})
    .pipe(minify())
    .pipe(gulp.dest('./dist/'));
});

gulp.task('test', function(){
  // lint our scripts
  gulp.src(scriptFiles).pipe(jshint());

  // run our tests
  spawn('npm', ['test'], {stdio: 'inherit'});
});

gulp.task('default', function(){
  gulp.run('test', 'compile');
  gulp.watch(scriptFiles, function(){
    gulp.run('test', 'compile');
  });
});

Cons

Because nothing is free

Might not play well with your 'Framework'

MVC-rap routes and functions

								// please provide
class People extends MVCrap {
	getAll()
	getByX()
}

much confuse

stream-to-pull-stream: Convert a classic-stream, or a new-stream into a pull-stream

invert-stream: Create a pair of streams (A, B) such that A.write(X) -> B.emit('data', X) and B.write(X) -> A.emit('data', X)

Backpressure and Unbounded Concurrency

The Node concurrency model is kind of like a credit card for computing work. Credit cards free you from the hassle and risks of carrying cash, (...) unless you spend more than you make. Try making 200,000 HTTP client requests in a tight loop, starting each request without waiting for the previous response.

Backpressure and Unbounded Concurrency

Once a process can't keep up with its work and starts to get slow, this slowness often cascades into other processes in the system. Streams are not the solution here. In fact, streams are kind of what gets us into this mess in the first place by providing a convenient fire and forget API for an individual socket or HTTP request without regard to the health of the rest of the system

- Matt Ranney

But the dream lives on...

Mix and match:

Oboe.js
Node Streams
CouchDB
node-LevelDB
gulp

Try them out today!

THANKS!