Upgrade the WebSocket libraries

lib/api/core/entrypoints/embedded/protocols/http.js

@@ -73,16 +73,14 @@ class HttpProtocol extends Protocol {
   init(entryPoint) {
     return super.init('http', entryPoint)
       .then(() => {
-        const config = entryPoint.config.protocols.http;
-
-        if (config.enabled === false) {
+        if (this.config.enabled === false) {
           return false;
         }
 
-        debug('initializing http Server with config: %a', entryPoint.config);
+        debug('initializing http Server with config: %a', this.config);
 
-        this.maxFormFileSize = bytes.parse(config.maxFormFileSize);
-        this.maxEncodingLayers = config.maxEncodingLayers;
+        this.maxFormFileSize = bytes.parse(this.config.maxFormFileSize);
+        this.maxEncodingLayers = this.config.maxEncodingLayers;
         this.server = entryPoint.httpServer;
 
         for (const value of ['maxFormFileSize', 'maxEncodingLayers']) {
@@ -400,7 +398,7 @@ class HttpProtocol extends Protocol {
    */
   _setDecoders() {
     const
-      allowCompression = this.entryPoint.config.protocols.http.allowCompression,
+      allowCompression = this.config.allowCompression,
       disabledfn = () => {
         throw new BadRequestError('Compression support is disabled');
       };

lib/api/core/entrypoints/embedded/protocols/mqtt.js

@@ -51,15 +51,15 @@ class MqttProtocol extends Protocol {
   init (entryPoint) {
     return super.init('mqtt', entryPoint)
       .then(() => {
-        if (entryPoint.config.protocols.mqtt.enabled === false) {
+        if (this.config.enabled === false) {
           return false;
         }
 
-        debug('initializing MQTT Server with config: %a', entryPoint.config.protocols.mqtt);
+        debug('initializing MQTT Server with config: %a', this.config);
 
         this.entryPoint = entryPoint;
         this.kuzzle = this.entryPoint.kuzzle;
-    
+
         this.config = Object.assign({
           allowPubSub: false,
           developmentMode: false,
@@ -69,10 +69,10 @@ class MqttProtocol extends Protocol {
           server: {
             port: 1883
           }
-        }, entryPoint.config.protocols.mqtt || {});
-    
+        }, this.config);
+
         this.server = new mosca.Server(this.config.server);
-    
+
         /*
          To avoid ill-use of our topics, we need to configure authorizations:
          * "requestTopic": should be publish-only, so no one but this plugin can listen to this topic
@@ -89,25 +89,25 @@ class MqttProtocol extends Protocol {
             callback(null, topic === this.config.requestTopic);
           }
         };
-    
+
         this.server.authorizeSubscribe = (client, topic, callback) => {
           const isAllowed = topic !== this.config.requestTopic
             && topic.indexOf('#') === -1
             && topic.indexOf('+') === -1;
-    
+
           callback(null, isAllowed);
         };
-    
+
         return new Bluebird(resolve => {
           this.server.on('ready', () => {
             this.server.on('clientConnected', client => this.onConnection(client));
             this.server.on('clientDisconnecting', client => this.onDisconnection(client));
             this.server.on('clientDisconnected', client => this.onDisconnection(client));
             this.server.on('published', (packet, client) => this.onMessage(packet, client));
-    
+
             resolve(true);
           });
-        });    
+        });
       });
   }
 

lib/api/core/entrypoints/embedded/protocols/protocol.js

@@ -28,6 +28,7 @@ class Protocol {
     this.maxRequestSize = null;
     this.entryPoint = null;
     this.name = null;
+    this.config = {};
   }
 
   /**
@@ -48,6 +49,10 @@ class Protocol {
           typeof this.name === 'string' && this.name.length > 0,
           'Invalid "name" parameter value: expected a non empty string value');
 
+        if (entryPoint.config.protocols && entryPoint.config.protocols[name]) {
+          this.config = entryPoint.config.protocols[name];
+        }
+
         assert(
           Number.isInteger(this.maxRequestSize),
           'Invalid "maxRequestSize" parameter value: expected a numeric value');

lib/api/core/entrypoints/embedded/protocols/socketio.js

@@ -47,28 +47,28 @@ class SocketIoProtocol extends Protocol {
   init(entryPoint) {
     return super.init('socketio', entryPoint)
       .then(() => {
-        if (entryPoint.config.protocols.socketio.enabled === false) {
+        if (this.config.enabled === false) {
           return false;
         }
-        
-        debug('initializing socketIo Server with config: %a', entryPoint.config.socketio);
-    
+
+        debug('initializing socketIo Server with config: %a', this.config);
+
         this.kuzzle = this.entryPoint.kuzzle;
-    
+
         // SocketIo server listens by default to "/socket.io" path
         // (see (http://socket.io/docs/server-api/#server#path(v:string):server))
         this.io = require('socket.io')(entryPoint.httpServer, {
           // maxHttpBufferSize is passed to the ws library as the maxPayload option
           // see https://github.com/socketio/socket.io/issues/2326#issuecomment-292146468
           maxHttpBufferSize: this.maxRequestSize
         });
-    
-        this.io.set('origins', entryPoint.config.protocols.socketio.origins);
-    
+
+        this.io.set('origins', this.config.origins);
+
         this.io.on('connection', socket => this.onConnection(socket));
         this.io.on('error', error => this.onServerError(error));
 
-        return true;    
+        return true;
       });
   }
 

lib/api/core/entrypoints/embedded/protocols/websocket.js

@@ -72,8 +72,6 @@ class WebSocketProtocol extends Protocol {
     this.kuzzle = null;
     this.heartbeatInterval = null;
     this.idleSweepInterval = null;
-    this.heartbeatTimeout = 0;
-    this.idleTimeout = 0;
 
     // Prevents thousands of "Date.now()" per second: it's far more efficient
     // to have a timestamp property refreshed every second or so, since we
@@ -97,16 +95,14 @@ class WebSocketProtocol extends Protocol {
   init (entryPoint) {
     return super.init('websocket', entryPoint)
       .then(() => {
-        const config = entryPoint.config.protocols.websocket;
-
-        if (config.enabled === false) {
+        if (this.config.enabled === false) {
           return false;
         }
 
-        debug('initializing WebSocket Server with config: %a', config);
+        debug('initializing WebSocket Server with config: %a', this.config);
 
-        this._startHeartbeat(config.heartbeat);
-        this._startIdleSweeps(config.idleTimeout);
+        this._startHeartbeat();
+        this._startIdleSweeps();
 
         this.kuzzle = this.entryPoint.kuzzle;
 
@@ -387,36 +383,34 @@ class WebSocketProtocol extends Protocol {
     }
   }
 
-  _startHeartbeat(heartbeat) {
+  _startHeartbeat() {
     assert(
-      Number.isInteger(heartbeat) && heartbeat >= 0,
-      `WebSocket: invalid heartbeat value ${heartbeat}`);
+      Number.isInteger(this.config.heartbeat) && this.config.heartbeat >= 0,
+      `WebSocket: invalid heartbeat value ${this.config.heartbeat}`);
 
-    if (heartbeat > 0) {
-      this.heartbeatTimeout = heartbeat;
+    if (this.config.heartbeat > 0) {
       this.heartbeatInterval = setInterval(
         this._doHeartbeat.bind(this),
-        heartbeat);
+        this.config.heartbeat);
     }
   }
 
-  _startIdleSweeps(idleTimeout) {
+  _startIdleSweeps() {
     assert(
-      Number.isInteger(idleTimeout) && idleTimeout >= 0,
-      `WebSocket: invalid idleTimeout value ${idleTimeout}`);
+      Number.isInteger(this.config.idleTimeout) && this.config.idleTimeout >= 0,
+      `WebSocket: invalid idleTimeout value ${this.config.idleTimeout}`);
 
-    if (idleTimeout > 0) {
-      this.idleTimeout = idleTimeout;
+    if (this.config.idleTimeout > 0) {
       this.idleSweepInterval = setInterval(
         this._sweepIdleSockets.bind(this),
-        this.idleTimeout);
+        this.config.idleTimeout);
     }
   }
 
   _doHeartbeat() {
     debug('Heartbeat');
     const
-      lastActivityThreshold = this.activityTimestamp - this.heartbeatTimeout;
+      lastActivityThreshold = this.activityTimestamp - this.config.heartbeat;
 
     for (const connection of this.connectionPool.values()) {
       if (connection.alive === false) {
@@ -439,7 +433,7 @@ class WebSocketProtocol extends Protocol {
     const now = Date.now();
 
     for (const connection of this.connectionPool.values()) {
-      if ((now - connection.lastActivity) > this.idleTimeout) {
+      if ((now - connection.lastActivity) > this.config.idleTimeout) {
         // correctly triggers the 'close' event handler on that socket
         connection.socket.terminate();
       }

test/api/core/entrypoints/embedded/index.test.js

@@ -1,5 +1,7 @@
 'use strict';
 
+const root = '../../../../../';
+
 const
   {
     Request,
@@ -11,18 +13,18 @@ const
     }
   } = require('kuzzle-common-objects'),
   path = require('path'),
-  KuzzleMock = require('../../../../mocks/kuzzle.mock'),
+  KuzzleMock = require(`${root}/test/mocks/kuzzle.mock`),
   mockrequire = require('mock-require'),
   rewire = require('rewire'),
   should = require('should'),
   sinon = require('sinon'),
   Bluebird = require('bluebird');
 
-class FakeProtocol { 
-  constructor (name) { 
-    this.name = name; 
+class FakeProtocol {
+  constructor (name) {
+    this.name = name;
   }
-}  
+}
 
 class FakeHttpProtocol extends FakeProtocol {
   constructor () { super('http'); }
@@ -83,10 +85,11 @@ describe('lib/core/api/core/entrypoints/embedded/index', () => {
     winstonTransportFile = sinon.spy();
     winstonTransportSyslog = sinon.spy();
 
-    mockrequire('../../../../../lib/api/core/entrypoints/embedded/protocols/http', HttpMock);
-    mockrequire('../../../../../lib/api/core/entrypoints/embedded/protocols/websocket', WebSocketMock);
-    mockrequire('../../../../../lib/api/core/entrypoints/embedded/protocols/socketio', SocketIOMock);
-    mockrequire('../../../../../lib/api/core/entrypoints/embedded/protocols/mqtt', MqttMock);
+    const embeddedPath = `${root}/lib/api/core/entrypoints/embedded`;
+    mockrequire(`${embeddedPath}/protocols/http`, HttpMock);
+    mockrequire(`${embeddedPath}/protocols/websocket`, WebSocketMock);
+    mockrequire(`${embeddedPath}/protocols/socketio`, SocketIOMock);
+    mockrequire(`${embeddedPath}/protocols/mqtt`, MqttMock);
 
     mockrequire('http', httpMock);
     mockrequire('winston', {
@@ -101,15 +104,16 @@ describe('lib/core/api/core/entrypoints/embedded/index', () => {
     mockrequire('winston-syslog', winstonTransportSyslog);
 
     // Disables unnecessary console warnings
-    AbstractManifest = rewire('../../../../../lib/api/core/abstractManifest');
+    AbstractManifest = rewire(`${root}/lib/api/core/abstractManifest`);
     AbstractManifest.__set__({ console: { warn: sinon.stub() }});
-    mockrequire('../../../../../lib/api/core/abstractManifest', AbstractManifest);
+    mockrequire(`${root}/lib/api/core/abstractManifest`, AbstractManifest);
 
-    Manifest = rewire('../../../../../lib/api/core/entrypoints/embedded/manifest');
+    Manifest = rewire(`${embeddedPath}/manifest`);
     Manifest.__set__({ console: { warn: sinon.stub() }});
-    mockrequire('../../../../../lib/api/core/entrypoints/embedded/manifest', Manifest);
+    mockrequire(`${embeddedPath}/manifest`, Manifest);
 
-    // Bluebird.map forces a different context, preventing rewire to mock "require"
+    // Bluebird.map forces a different context, preventing rewire to mock
+    // "require"
     mockrequire('bluebird', {
       map: (arr, fn) => Promise.all(arr.map(e => {
         let result;
@@ -127,7 +131,7 @@ describe('lib/core/api/core/entrypoints/embedded/index', () => {
       all: Bluebird.all
     });
 
-    EntryPoint = mockrequire.reRequire('../../../../../lib/api/core/entrypoints/embedded');
+    EntryPoint = mockrequire.reRequire(embeddedPath);
 
     entrypoint = new EntryPoint(kuzzle);
 
@@ -140,10 +144,9 @@ describe('lib/core/api/core/entrypoints/embedded/index', () => {
       }
     });
 
-    HttpMock.prototype.init = sinon.stub().resolves(true);
-    WebSocketMock.prototype.init = sinon.stub().resolves(true);
-    SocketIOMock.prototype.init = sinon.stub().resolves(true);
-    MqttMock.prototype.init = sinon.stub().resolves(true);
+    for (const Class of [HttpMock, WebSocketMock, SocketIOMock, MqttMock]) {
+      Class.prototype.init = sinon.stub().resolves(true);
+    }
   });
 
   afterEach(() => {

test/api/core/entrypoints/embedded/protocols/websocket.test.js

@@ -599,6 +599,7 @@ describe('/lib/api/core/entrypoints/embedded/protocols/websocket', () => {
         ['active', new Connection(true, Date.now())]
       ]);
 
+      protocol.config.heartbeat = 1000;
       protocol._doHeartbeat();
 
       // inactive sockets are pinged
@@ -667,11 +668,11 @@ describe('/lib/api/core/entrypoints/embedded/protocols/websocket', () => {
           should(protocol.idleTimeoutInterval).not.be.null();
           should(idleTimeoutSpy).not.be.called();
 
-          clock.tick(protocol.idleTimeout);
+          clock.tick(protocol.config.idleTimeout);
 
           should(idleTimeoutSpy).be.calledOnce();
 
-          clock.tick(protocol.idleTimeout);
+          clock.tick(protocol.config.idleTimeout);
 
           should(idleTimeoutSpy).be.calledTwice();
 
@@ -695,12 +696,15 @@ describe('/lib/api/core/entrypoints/embedded/protocols/websocket', () => {
           };
         };
 
-      protocol.idleTimeout = 1000;
+      protocol.config.idleTimeout = 1000;
 
       protocol.connectionPool = new Map([
         ['ahAhAhAhStayinAliveStayinAlive', new Connection(now)],
-        ['dead', new Connection(now - protocol.idleTimeout - 1)],
-        ['ahAhAhAhStayinAliiiiiiiiive', new Connection(now - protocol.idleTimeout + 100)]
+        ['dead', new Connection(now - protocol.config.idleTimeout - 1)],
+        [
+          'ahAhAhAhStayinAliiiiiiiiive',
+          new Connection(now - protocol.config.idleTimeout + 100)
+        ]
       ]);
 
       protocol._sweepIdleSockets();