make sync work

app/controllers/socket.js

@@ -4,6 +4,7 @@
 var ios = null; // io is already taken in express
 var util = require('bitcore').util;
 var mdb = require('../../lib/MessageDb').default();
+var microtime = require('microtime');
 
 module.exports.init = function(io_ext) {
   ios = io_ext;
@@ -12,16 +13,26 @@ module.exports.init = function(io_ext) {
     ios.sockets.on('connection', function(socket) {
       // when it subscribes, make it join the according room
       socket.on('subscribe', function(topic) {
-        socket.join(topic);
+        if (socket.rooms.length === 1) {
+          console.log('subscribe to '+topic);
+          socket.join(topic);
+        }
       });
 
       // when it requests sync, send him all pending messages
       socket.on('sync', function(ts) {
-        mdb.getMessages(to, lower_ts, upper_ts, function(err, messages) {
+        var rooms = socket.rooms;
+        if (rooms.length !== 2) {
+          socket.emit('insight-error', 'Must subscribe with public key before syncing');
+          return;
+        }
+        var to = rooms[1];
+        var upper_ts = Math.round(microtime.now());
+        mdb.getMessages(to, ts, upper_ts, function(err, messages) {
           if (err) {
             throw new Error('Couldn\'t get messages on sync request: ' + err);
           }
-          for (var i = 0; i < message.length; i++) {
+          for (var i = 0; i < messages.length; i++) {
             broadcastMessage(messages[i]);
           }
         });

examples/messages.html

@@ -104,7 +104,8 @@
 
       // receive chat handler
       socket.emit('subscribe', pubkey);
-      socket.emit('sync', pubkey);
+      var ts = undefined; // timestamp to sync, undefined syncs all
+      socket.emit('sync', ts);
       socket.on('message', function(msg)
       {
         try {

lib/MessageDb.js

@@ -98,6 +98,7 @@ MessageDb.fromStorage = function(data) {
 
 MessageDb.prototype.getMessages = function(to, lower_ts, upper_ts, cb) {
   var list = [];
+  lower_ts = lower_ts || 1;
   var opts = {
     end: messageKey(to, lower_ts),
     start: messageKey(to, upper_ts),

test/test.MessageDb.js

@@ -61,8 +61,10 @@ describe('MessageDb', function() {
       });
     });
   });
-  it('should be able to add many messages and read them', function(done) {
+  var sharedMDB;
+  it('should be able to add many messages and read some', function(done) {
     var mdb = new MessageDb(opts);
+    sharedMDB = mdb;
     var lower_ts = microtime.now();
     mdb.addMessage(message, function(err) {
       expect(err).to.not.exist;
@@ -90,6 +92,22 @@ describe('MessageDb', function() {
       });
     });
   });
+  it('should be able to add many messages and read all', function(done) {
+    var mdb = sharedMDB;
+    mdb.getMessages(to, null, null, function(err, messages) {
+      expect(err).to.not.exist;
+      messages.length.should.equal(4);
+      var m0 = AuthMessage.decode(tpk, messages[0]).payload;
+      JSON.stringify(m0).should.equal('{"a":1,"b":2}');
+      var m1 = AuthMessage.decode(tpk, messages[1]).payload;
+      JSON.stringify(m1).should.equal('{"a":1,"b":2}');
+      var m2 = AuthMessage.decode(tpk, messages[2]).payload;
+      JSON.stringify(m2).should.equal('{}');
+      var m3 = AuthMessage.decode(tpk, messages[3]).payload;
+      JSON.stringify(m3).should.equal('["a","b"]');
+      done();
+    });
+  });
   it('should be able to close instance', function() {
     var mdb = new MessageDb(opts);
     mdb.close();