Node js streams with mongo and filter transforms! note the ???ObjectMode settings!

mongodb node.js stream.js

var mongodb = require("mongodb");
var MongoClient = mongodb.MongoClient;
var fs = require("fs");

var Transform = require("stream").Transform;

var name_surname_email_filter = new Transform({readableObjectMode:true, writableObjectMode:true} );
name_surname_email_filter._transform = function(data, enc, cb) {
    //console.log(data);
    var newdata = { first_name: data.first_name,
		    last_name : data.last_name,
		    email : data.email };
    name_surname_email_filter.push(newdata);
    cb(); //, '\n---'+data);    
}

var name_surname_filter = new Transform({readableObjectMode: true, writableObjectMode: true});
name_surname_filter._transform = function(data,enc, cb){
    var newdata = { first_name : data.first_name }
    name_surname_filter.push(newdata);
    cb();
};


MongoClient.connect("mongodb://localhost:27017/test", function(err, db) {
    var collection = db.collection("persons");
    var stream = collection.find({}).limit(20).stream();
    var fileout = fs.createWriteStream("bla.json");
    stream.on("end", function(err) {
    	db.close();
    });
    stream
	.pipe(name_surname_email_filter)
	.pipe(name_surname_filter)
	.on("data", function(data) { console.log(">>>", JSON.stringify(data))}); //.pipe(fileout);
});


    // stream.on("data", function(data) {
    // 	console.log(">>>" , data);
    // })
    // stream.on('readable', function() {
    // 	while (null !== (chunk = stream.read())) {
    // 	    console.log("....", chunk);
    // 	}
    // });
    // stream.on('end', function() {
    // 	db.close()
    // });