http: refactor websocket handling.

lib/bcoin/http/server.js

@@ -946,17 +946,17 @@ HTTPServer.prototype._initIO = function _initIO() {
       socket.destroy();
     });
 
-    socket.once('auth', function(apiKey, callback) {
-      if (typeof callback !== 'function')
-        return socket.destroy();
+    socket.on('auth', function(args, callback) {
+      var apiKey = args[0];
+
+      if (socket.auth)
+        return callback({ error: 'Already authed.' });
 
       socket.stop();
 
       if (self.apiHash) {
-        if (!utils.ccmp(hash256(apiKey), self.apiHash)) {
-          socket.destroy();
+        if (!utils.ccmp(hash256(apiKey), self.apiHash))
           return callback({ error: 'Bad key.' });
-        }
       }
 
       socket.auth = true;
@@ -976,9 +976,9 @@ HTTPServer.prototype._initIO = function _initIO() {
   });
 
   this.on('websocket', function(socket) {
-    socket.on('wallet join', function(id, token, callback) {
-      if (typeof callback !== 'function')
-        return socket.destroy();
+    socket.on('wallet join', function(args, callback) {
+      var id = args[0];
+      var token = args[1];
 
       if (typeof id !== 'string')
         return callback({ error: 'Invalid parameter.' });
@@ -988,7 +988,7 @@ HTTPServer.prototype._initIO = function _initIO() {
         return callback();
       }
 
-      if (typeof token !== 'string')
+      if (!utils.isHex256(token))
         return callback({ error: 'Invalid parameter.' });
 
       self.walletdb.auth(id, token, function(err, wallet) {
@@ -1008,9 +1008,8 @@ HTTPServer.prototype._initIO = function _initIO() {
       });
     });
 
-    socket.on('wallet leave', function(id, callback) {
-      if (typeof callback !== 'function')
-        return socket.destroy();
+    socket.on('wallet leave', function(args, callback) {
+      var id = args[0];
 
       if (typeof id !== 'string')
         return callback({ error: 'Invalid parameter.' });
@@ -1020,48 +1019,57 @@ HTTPServer.prototype._initIO = function _initIO() {
       callback();
     });
 
-    socket.on('watch chain', function() {
+    socket.on('watch chain', function(args, callback) {
       socket.watchChain();
+      callback();
     });
 
-    socket.on('unwatch chain', function() {
+    socket.on('unwatch chain', function(args, callback) {
       socket.unwatchChain();
+      callback();
     });
 
-    socket.on('watch address', function(addresses) {
+    socket.on('watch address', function(args, callback) {
+      var addresses = args[0];
+
       if (!Array.isArray(addresses))
-        return socket.destroy();
+        return callback({ error: 'Invalid parameter.' });
 
       try {
         socket.addFilter(addresses);
       } catch (e) {
-        return socket.destroy();
+        return callback({ error: e.message });
       }
+
+      callback();
     });
 
-    socket.on('unwatch address', function(addresses) {
+    socket.on('unwatch address', function(args, callback) {
+      var addresses = args[0];
+
       if (!Array.isArray(addresses))
-        return socket.destroy();
+        return callback({ error: 'Invalid parameter.' });
 
       try {
         socket.removeFilter(addresses);
-      } catch (e) {
-        return socket.destroy();
-      }
-    });
-
-    socket.on('scan chain', function(start, callback) {
-      if (typeof callback !== 'function')
-        return socket.destroy();
-
-      if (typeof start !== 'string' && !utils.isNumber(start))
-        return callback({ error: 'Invalid parameter.' });
-
-      try {
-        socket.scan(start, callback);
       } catch (e) {
         return callback({ error: e.message });
       }
+
+      callback();
+    });
+
+    socket.on('scan chain', function(args, callback) {
+      var start = args[0];
+
+      if (!utils.isHex256(start) && !utils.isNumber(start))
+        return callback({ error: 'Invalid parameter.' });
+
+      socket.scan(start, function(err) {
+        if (err)
+          return callback({ error: err.message });
+        callback();
+      });
     });
   });
 
@@ -1232,7 +1240,16 @@ ClientSocket.prototype._init = function _init() {
   socket.onevent = function(packet) {
     var result = onevent(packet);
     var args = packet.data || [];
-    emit.apply(self, args);
+    var event = args.shift();
+    var ack;
+
+    if (typeof args[args.length - 1] === 'function')
+      ack = args.pop();
+    else
+      ack = self.socket.ack(packet.id);
+
+    emit.apply(self, [event, args, ack]);
+
     return result;
   };
 
@@ -1381,12 +1398,9 @@ ClientSocket.prototype.scan = function scan(start, callback) {
       txs[i] = txs[i].toJSON();
 
     self.emit('block tx', entry.toJSON(), txs);
+
     next();
-  }, function(err) {
-    if (err)
-      return callback({ error: err.message });
-    callback();
-  });
+  }, callback);
 };
 
 ClientSocket.prototype.join = function join(id) {