mongo / socket.io / node.js / express

server.js

//var io = require('socket.io');
var express = require('express')
, stylus = require('stylus')
, http = require('http')
, nib = require('nib');


var app = express();
var server = http.createServer(app);
var io = require('socket.io').listen(server);
server.listen(8000);

app.configure(function () {
  app.use(stylus.middleware({ src: __dirname + '/public', compile: compile }));
  app.use(express.static(__dirname + '/public'));
  app.set('views', __dirname);
  app.set('view engine', 'jade');

  function compile (str, path) {
    return stylus(str)
    .set('filename', path)
    .use(nib());
  };
});

app.get('/', function (req, res) {
  res.render('index', { layout: false });
});


io.sockets.on('connection', function (socket) {
  console.log('A socket connected!');
  //  socket.emit('news', { hello: 'world' });
  socket.on('my other event', function (data) {
    console.log(data);
  });
  //  var timestamp = Math.round((new Date()).getTime() / 1000);
  socket.on('next_payloads', function (timestamp) {
    console.log('timestamp is ' + timestamp);
    console.log('timestamp class is ' + typeof(timestamp))
    var mongodb = require('mongodb');
    var server = new mongodb.Server("ec2-xxx.compute-1.amazonaws.com", 27017, {auto_reconnect:true});
    new mongodb.Db('universal', server, {}).open(function (error, client) {
            //      client.on("close", function(error){
            //        console.log("Connection to the database was closed!");
            //      });


      if (error) throw error;
      var collection = new mongodb.Collection(client, 'y_201207');
      var stream = collection.find({'timestamp':{'$gt':timestamp}}, {limit:20, sort:[['timestamp', 1]]}).streamRecords();
      //var stream = collection.find({}, {limit:10}).streamRecords();
      stream.on("data", function(doc) {
        console.log('old timestamp is ' + timestamp);
        timestamp = doc.timestamp
        console.log('new timestamp is ' + timestamp);
        var date = new Date(timestamp * 1000);
        var retval = "(" + date + ") " + doc.username + ": " + doc.text;
        console.dir("new timestamp is " + timestamp.toString());
        socket.emit('payload', {'payload_id':doc.id, 'data': retval});
      });
      stream.on("end", function() {
        console.log("finished with cursor")
        console.log('last timestamp is ' + timestamp);
        socket.emit('done', timestamp)
      });
    });
  });
  socket.on('disconnect', function () {
    console.log('user disconnected');
  });
});