fix(brain,client,server): finally the real fix

Author: imlucas

scout-brain/lib/actions/token.js

@@ -3,7 +3,6 @@ var jwt = require('jsonwebtoken');
 var boom = require('boom');
 var debug = require('debug')('scout-brain:token');
 var _ = require('underscore');
-var assert = require('assert');
 
 var getDeployment = require('./deployment').get;
 var createSession = require('./session').create;
@@ -26,6 +25,7 @@ function verify(token, fn) {
 }
 
 function mount(tokenData, ctx, next) {
+  debug('mounting token data');
   if (!tokenData.session_id) {
     return next(boom.badRequest('Bad token: missing session id'));
   }
@@ -57,23 +57,19 @@ function mount(tokenData, ctx, next) {
         'to `' + ctx.instance_id + '` but it is not in ' +
         'deployment `' + tokenData.deployment_id + '`'));
       }
+    }
 
-      debug('getting connection for session', tokenData.session_id);
-      getSession(tokenData.session_id, function(err, connection) {
-        if (err) return next(err);
-
-        ctx.mongo = connection;
-        return next();
-      });
+    debug('getting connection for session', tokenData.session_id);
+    getSession(tokenData.session_id, function(err, connection) {
+      if (err) return next(err);
 
-    } else {
-      getSession(tokenData.session_id, function(err, connection) {
-        if (err) return next(err);
+      if (!connection) {
+        return next(boom.forbidden('No session for this token.'));
+      }
 
-        ctx.mongo = connection;
-        return next();
-      });
-    }
+      ctx.mongo = connection;
+      return next();
+    });
   });
 }
 

scout-client/lib/client.js

@@ -124,8 +124,8 @@ Client.prototype.connect = function() {
     return this;
   }
   this.token = new Token(this.config)
-    .on('readable', this.onTokenReadable.bind(this))
-    .on('error', this.onTokenError.bind(this));
+    .on('data', this.onTokenReadable.bind(this))
+    .on('error', this.onError.bind(this));
   return this;
 };
 
@@ -558,7 +558,7 @@ Client.prototype.read = function(path, params, fn) {
 
   if (!this.readable) {
     debug('%s not readable.  queueing read', this._id, path, params);
-    return this.on('readable', this.read.bind(this, path, params, fn));
+    return this.once('readable', this.read.bind(this, path, params, fn));
   }
 
   if (typeof params === 'function') {
@@ -618,14 +618,14 @@ Client.prototype.ender = function(fn) {
  * @api private
  */
 Client.prototype.onTokenReadable = function() {
-  debug('token now readable');
-  this.readable = true;
-  this.context.set(this.token.session);
   if (!this.io) {
     this._initSocketio();
   }
-  this.emit('readable', this.token.session);
-  debug('emitted readable on client');
+  debug('authenticating with scout-server socket.io transport...');
+  this.io.emit('authenticate', {
+    token: this.token.toString()
+  });
+
 
   if (!this.original) {
     this.emit('change');
@@ -642,30 +642,27 @@ Client.prototype.onTokenReadable = function() {
   }
 };
 
-/**
- * We couldn't get a token.
- *
- * @api private
- */
-Client.prototype.onTokenError = function(err) {
-  debug('Could not get token.  Server not running?', err);
-  this.emit('error', err);
-};
-
 Client.prototype._initSocketio = function() {
   if (this.io) return;
 
-  this.io = socketio(this.config.scout, {
-    query: 'token=' + this.token.toString()
-  });
+  this.io = socketio(this.config.scout);
   this.io.on('reconnecting', this.emit.bind(this, 'reconnecting'))
     .on('reconnect', this.onReconnect.bind(this))
     .on('reconnect_attempt', this.emit.bind(this, 'reconnect_attempt'))
     .on('reconnect_failed', this.emit.bind(this, 'reconnect_failed'))
     .on('disconnect', this.emit.bind(this, 'disconnect'));
   this.io.on('connect', function() {
     debug('connected to scout-server socket');
-  });
+    this.io.on('authenticated', function() {
+      debug('Success!  Now authenticated with scout-server socket.io transport');
+
+      this.readable = true;
+      this.context.set(this.token.session);
+      debug('now ready for use');
+      this.emit('readable', this.token.session);
+    }.bind(this));
+  }.bind(this));
+  this.io.on('error', this.onError.bind(this));
 };
 
 /**
@@ -689,20 +686,19 @@ Client.prototype.onUnload = function() {
  * @param {Error} err
  * @api private
  */
-Client.prototype.onTokenError = function(err) {
-  this.dead = err;
-  if (err >= 500) {
-    this.dead.message += ' (scout-server dead at ' + this.config.scout + '?)';
+Client.prototype.onError = function(err) {
+  if (err.message === 'Origin is not allowed by Access-Control-Allow-Origin') {
+    err.message = 'scout-server not running at ' + this.config.scout + '?';
   }
+  // @todo: Exponential back-off to retry connecting.
+  this.dead = err;
   this.emit('error', err);
 };
 
 Client.prototype.onReconnect = function() {
   debug('reconnected.  getting new token');
   this.token = new Token(this.config)
     .on('readable', function() {
-      this.readable = true;
-      this.context.set(this.token.session);
       this._initSocketio();
     }.bind(this))
     .on('error', this.emit.bind(this, 'error'));

scout-server/lib/io.js

@@ -18,12 +18,16 @@ var _idToDocument = require('./streams/id-to-document');
 var EJSON = require('mongodb-extended-json');
 var typedParams = require('./middleware/typed-params');
 
-io.use(require('socketio-jwt').authorize({
+
+io.on('connection', require('socketio-jwt').authorize({
   secret: config.get('token:secret').toString('utf-8'),
-  handshake: true
-}));
+  timeout: 15000
+})).on('authenticated', function(socket) {
+  debug('authenticated with token data', socket.decoded_token);
+});
 
 function prepare(socket, req, done) {
+  debug('preparing socket.io request');
   req.params = _.extend({
     ns: req.ns,
     size: req.size,
@@ -35,10 +39,7 @@ function prepare(socket, req, done) {
 
     var tasks = {};
     tasks.token = function(next) {
-      brain.loadToken(socket.decoded_token, req, function() {
-        debug('load token returned', arguments);
-        next();
-      });
+      brain.loadToken(socket.decoded_token, req, next);
     };
 
     if (req.params.ns) {
@@ -55,8 +56,8 @@ function prepare(socket, req, done) {
         done();
       });
     }
-
     async.series(tasks, function(err) {
+      debug('socket.io request now prepared');
       if (err) return done(err);
       done();
     });
@@ -65,7 +66,9 @@ function prepare(socket, req, done) {
 
 io.on('connection', function(socket) {
   ss(socket).on('collection:sample', function(stream, req) {
-    prepare(socket, req, function() {
+    prepare(socket, req, function(err) {
+      if (err) return stream.emit('error', err);
+
       debug('collection:sample got req %j', Object.keys(req));
       var db = req.mongo.db(req.params.database_name);
       createSampleStream(db, req.params.collection_name, {
@@ -79,5 +82,4 @@ io.on('connection', function(socket) {
         .pipe(stream);
     });
   });
-  debug('token data %j', socket.decoded_token);
 });