Stream the output of a LevelDB into a socket, using a Transform stream

stream_leveldb.js

#!/usr/bin/nodejs

var levelup   = require('levelup');
var http      = require('http');
var util      = require('util');
var Transform = require('stream').Transform || require('readable-stream').Transform;

var server = http.createServer(onRequest);

// Save something in db
var db = levelup('db',{"valueEncoding": "json"});
db.put("1",{"leo":1});
db.put("2",{"nardo":2});
db.put("3",{"LEONARDO":3});

server.listen(9090, function() {
  console.log("Server has started --- " + (new Date).toJSON());
});

server.on('error', function (e) {
  if (e.code == 'EADDRINUSE') {
    console.error('Error: address in use');
  }
  else {
    console.error(e + '. Error code: ' + e.code);
  }
});

function onRequest(request, response) {
  response.setHeader("Content-Type", "application/json; charset=utf-8");
  var rs = db.createReadStream();

  request.on('close', function(){
    rs.destroy();
    response.end();
  });

  rs.on('end', function(){
    response.end();
  });
  rs.on('error', function(err){
    console.log('ERROR-RS',err);
    rs.destroy();
  });

  var stringifier = new Transform();
  stringifier._writableState.objectMode = true;
  stringifier._transform = function (data, encoding, done) {
      this.push(JSON.stringify(data));
      this.push('\n');
      done();
  }
  rs.pipe(stringifier).pipe(response);
}

$  node leveldb.js
Server has started --- 2013-12-03T10:04:28.050Z

$  curl http://localhost:9090
{"key":"1","value":{"leo":1}}
{"key":"2","value":{"nardo":2}}
{"key":"3","value":{"LEONARDO":3}}
$