gistfile1.js

var streamWorker = new Transform({ objectMode: true, decodeStrings: false });

var queue = async.queue(function(task, callback) {
    var ObjectId = mongoose.Types.ObjectId
    var query    = task.query
    var update   = task.update
    var defer    = task.defer

    query._id['$in'] = _.map(query._id['$in'], ObjectId)
    var queryStream  = scenarios.find(query).stream()

    queryStream.pipe(streamWorker).pipe(process.stdout)
}

    streamWorker._transform = function(doc, enc, done) {
        console.log('doc:', doc);
        console.log('enc:', enc)
        done(null, 'hi');
    }