goosestream.js

var mongoose = require('mongoose')
  , Schema = mongoose.Schema
  , db = mongoose.connect('localhost', 'testing_streaming').connection
  , Stream = require('stream').Stream
  , express = require('express')

/**
 * Dummy schema.
 */

var Person = new Schema({
    name: String
});

var P = db.model('Person', Person);

/**
 * Create some dummy people.
 */

db.on('open', function () {
  db.db.dropDatabase(function () {
    var people = [];
    for (var i = 0; i < 5000; ++i) {
      people.push({ name: i });
    }

    P.create(people, startup);
  });
});

/**
 * Start up an express app
 */

function startup () {
  var app = express.createServer()

  app.get('/', function stream (req, res, next) {
    if ('HEAD' == req.method) return res.end();

    // output json
    res.contentType('json');

    // use our lame formatter
    var format = new ArrayFormatter;

    // first pipe the querystream to the formatter
    P.find().stream().pipe(format);

    // then pipe the formatter to the response
    // (node 0.4x style pipe non-chaining)
    format.pipe(res);

    // In node 0.6 we can P.find().stream().pipe(format).pipe(res);
  })

  app.listen(8000);
  console.error('listening on http://localhost:8000');
}

/**
 * A hacked querystream formatter which formats the output
 * as a json literal. Not production quality.
 */

function ArrayFormatter () {
  Stream.call(this);
  this.writable = true;
  this._done = false;
}

ArrayFormatter.prototype.__proto__ = Stream.prototype;

ArrayFormatter.prototype.write = function (doc) {
  if (! this._hasWritten) {
    this._hasWritten = true;

    // open an object literal / array string along with the doc
    this.emit('data', '{ "results": [' + JSON.stringify(doc) );

  } else {
    this.emit('data', ',' + JSON.stringify(doc));
  }

  return true;
}

ArrayFormatter.prototype.end =
ArrayFormatter.prototype.destroy = function () {
  if (this._done) return;
  this._done = true;

  // close the object literal / array
  this.emit('data', ']}');
  // done
  this.emit('end');
}