cursor.js

Cursor.prototype.streamRecords = function(callback) {
  var
    self = this,
    stream = new process.EventEmitter(),
    recordLimitValue = this.limitValue || 0,
    emittedRecordCount = 0,
    queryCommand = this.generateQueryCommand();
    
  // see http://www.mongodb.org/display/DOCS/Mongo+Wire+Protocol
  queryCommand.numberToReturn = 500; 
  
  execute(queryCommand);
  
  function execute(command) {
    self.db.executeCommand(command, function(err,result) {
      if (!self.queryRun && result) {
        self.queryRun = true;
        self.cursorId = result.cursorId;
        self.state = Cursor.OPEN;
        self.getMoreCommand = new GetMoreCommand(self.db.databaseName + "." + self.collection.collectionName, queryCommand.numberToReturn, result.cursorId);
      }
      if (result.documents && result.documents.length) {
        result.documents.forEach(function(doc){ 
          if (recordLimitValue && emittedRecordCount>recordLimitValue) {
            stream.emit('end', recordLimitValue);
            self.close(function(){});
            return(null);
          }
          emittedRecordCount++;
          stream.emit('data', doc); 
        });
        // rinse & repeat
        execute(self.getMoreCommand);
      } else {
        stream.emit('end', recordLimitValue);
        self.close(function(){});
      }
    });
  }
  return stream;
};