Merge remote-tracking branch 'origin/6.x' into #102-auth-ctrl

# Conflicts:
#	src/Kuzzle.js
#	src/networkWrapper/protocols/abstract/realtime.js
#	test/network/offlineQueue.test.js
#	test/network/query.test.js

Author: jenow

package-lock.json

@@ -28,6 +28,7 @@
   "license": "Apache-2.0",
   "dependencies": {
     "bluebird": "3.5.1",
+    "mock-require": "^3.0.1",
     "uuid": "3.2.1",
     "uws": "^9.14.0"
   },

package.json

@@ -227,11 +227,9 @@ Collection.prototype.deleteDocument = function (arg, options, cb) {
 
   this.kuzzle.query(this.buildQueryArgs('document', action), data, options, cb && function (err, res) {
     if (err) {
-      cb(err);
-    }
-    else {
-      cb(null, (action === 'delete' ? [res.result._id] : res.result.ids));
+      return cb(err);
     }
+    cb(null, (action === 'delete' ? [res.result._id] : res.result.ids));
   });
 
   return this;

src/Collection.js

@@ -345,7 +345,7 @@ class Kuzzle extends KuzzleEventEmitter {
    * @param {function} [cb] Connection callback
    */
   connect (cb) {
-    if (this.network.state !== 'offline') {
+    if (this.network.isReady()) {
       if (cb) {
         cb(null, this);
       }
@@ -374,7 +374,10 @@ class Kuzzle extends KuzzleEventEmitter {
     });
 
     this.network.addListener('disconnect', () => {
-      this.disconnect();
+      for (const collection of Object.keys(this.collections)) {
+        delete this.collections[collection];
+      }
+
       this.emit('disconnected');
     });
 

src/Kuzzle.js

@@ -8,6 +8,8 @@
 
 function network(protocol, host, options) {
   switch (protocol) {
+    case 'http':
+      return new (require('./protocols/http'))(host, options);
     case 'websocket':
       if (typeof window !== 'undefined' && typeof WebSocket === 'undefined') {
         throw new Error('Aborting: no websocket support detected.');

src/networkWrapper/index.js

@@ -0,0 +1,303 @@
+'use strict';
+
+const
+  KuzzleEventEmitter = require('../../../eventEmitter');
+
+class AbtractWrapper extends KuzzleEventEmitter {
+
+  constructor (host, options) {
+    super();
+
+    Object.defineProperties(this, {
+      host: {
+        value: host,
+        enumerable: true
+      },
+      port: {
+        value: (options && typeof options.port === 'number') ? options.port : 7512,
+        enumerable: true
+      },
+      ssl: {
+        value: (options && typeof options.sslConnection === 'boolean') ? options.sslConnection : false,
+        enumerable: true
+      },
+      queuing: {
+        value: false,
+        writable: true
+      },
+      // configuration properties
+      autoQueue: {
+        value: false,
+        enumerable: true,
+        writable: true
+      },
+      autoReplay: {
+        value: false,
+        enumerable: true,
+        writable: true
+      },
+      state: {
+        value: 'offline',
+        enumerable: true,
+        writable: true
+      },
+      /*
+        Offline queue use the following format:
+              [
+                {
+                  ts: <query timestamp>,
+                  query: 'query',
+                  cb: callbackFunction
+                }
+              ]
+       */
+      offlineQueue: {
+        value: [],
+        enumerable: true,
+        writable: true
+      },
+      queueFilter: {
+        value: null,
+        enumerable: true,
+        writable: true
+      },
+      queueMaxSize: {
+        value: 500,
+        enumerable: true,
+        writable: true
+      },
+      queueTTL: {
+        value: 120000,
+        enumerable: true,
+        writable: true
+      },
+      replayInterval: {
+        value: 10,
+        enumerable: true,
+        writable: true
+      },
+      offlineQueueLoader: {
+        value: null,
+        enumerable: true,
+        writable: true
+      }
+    });
+
+    if (options) {
+      Object.keys(options).forEach(opt => {
+        if (this.hasOwnProperty(opt) && Object.getOwnPropertyDescriptor(this, opt).writable) {
+          this[opt] = options[opt];
+        }
+      });
+    }
+  }
+
+  /* @Abstract */
+  connect() {
+    throw new Error('Method "connect" is not implemented');
+  }
+
+  /**
+   * Called when the client's connection is established
+   */
+  clientConnected(state, wasConnected) {
+    this.state = state || 'ready';
+    this.emit(wasConnected && 'reconnect' || 'connect');
+
+    if (this.autoQueue) {
+      this.stopQueuing();
+    }
+
+    if (this.autoReplay) {
+      this.playQueue();
+    }
+  }
+
+  /**
+   * Called when the client's connection is closed
+   */
+  disconnect() {
+    this.state = 'offline';
+    if (this.autoQueue) {
+      this.startQueuing();
+    }
+  }
+
+  /**
+   * Empties the offline queue without replaying it.
+   */
+  flushQueue() {
+    this.offlineQueue = [];
+  }
+
+  /**
+   * Replays the requests queued during offline mode.
+   */
+  playQueue() {
+    if (this.isReady()) {
+      cleanQueue(this);
+      dequeue(this);
+    }
+  }
+
+  /**
+   * Starts the requests queuing. Works only during offline mode, and if the autoQueue option is set to false.
+   */
+  startQueuing() {
+    this.queuing = true;
+  }
+
+  /**
+   * Stops the requests queuing. Works only during offline mode, and if the autoQueue option is set to false.
+   */
+  stopQueuing() {
+    this.queuing = false;
+  }
+
+  subscribe(object, options, notificationCB, cb) { // eslint-disable-line
+    throw new Error('Not Implemented');
+  }
+
+  unsubscribe(object, options, channel, cb) { // eslint-disable-line
+    throw new Error('Not Implemented');
+  }
+
+  query(object, options, cb) {
+    let queuable = options && (options.queuable !== false) || true;
+
+    if (this.queueFilter) {
+      queuable = queuable && this.queueFilter(object);
+    }
+
+    if (this.queuing && queuable) {
+      cleanQueue(this, object, cb);
+      this.emit('offlineQueuePush', {query: object, cb: cb});
+      return this.offlineQueue.push({ts: Date.now(), query: object, cb: cb});
+    }
+
+    if (this.isReady()) {
+      return emitRequest(this, object, cb);
+    }
+
+    return discardRequest(object, cb);
+  }
+
+  isReady() {
+    return this.state === 'ready';
+  }
+}
+
+/**
+ * Emit a request to Kuzzle
+ *
+ * @param {AbstractWrapper} network
+ * @param {object} request
+ * @param {responseCallback} [cb]
+ */
+function emitRequest (network, request, cb) {
+  if (request.jwt !== undefined || cb) {
+    network.once(request.requestId, response => {
+      let error = null;
+
+      if (request.action !== 'logout' && response.error && response.error.message === 'Token expired') {
+        network.emit('tokenExpired', request, cb);
+      }
+
+      if (response.error) {
+        error = new Error(response.error.message);
+        Object.assign(error, response.error);
+        error.status = response.status;
+        network.emit('queryError', error, request, cb);
+      }
+
+      if (cb) {
+        cb(error, response);
+      }
+    });
+  }
+  // Track requests made to allow Room.subscribeToSelf to work
+  network.send(request);
+}
+
+function discardRequest(object, cb) {
+  if (cb) {
+    cb(new Error('Unable to execute request: not connected to a Kuzzle server.\nDiscarded request: ' + JSON.stringify(object)));
+  }
+}
+
+/**
+ * Clean up the queue, ensuring the queryTTL and queryMaxSize properties are respected
+ * @param {AbstractWrapper} network
+ */
+function cleanQueue (network) {
+  const now = Date.now();
+  let lastDocumentIndex = -1;
+
+  if (network.queueTTL > 0) {
+    network.offlineQueue.forEach((query, index) => {
+      if (query.ts < now - network.queueTTL) {
+        lastDocumentIndex = index;
+      }
+    });
+
+    if (lastDocumentIndex !== -1) {
+      network.offlineQueue
+        .splice(0, lastDocumentIndex + 1)
+        .forEach(droppedRequest => {
+          network.emit('offlineQueuePop', droppedRequest.query);
+        });
+    }
+  }
+
+  if (network.queueMaxSize > 0 && network.offlineQueue.length > network.queueMaxSize) {
+    network.offlineQueue
+      .splice(0, network.offlineQueue.length - network.queueMaxSize)
+      .forEach(droppedRequest => {
+        network.emit('offlineQueuePop', droppedRequest.query);
+      });
+  }
+}
+
+/**
+ * Play all queued requests, in order.
+ */
+function dequeue (network) {
+  const
+    uniqueQueue = {},
+    dequeuingProcess = () => {
+      if (network.offlineQueue.length > 0) {
+        emitRequest(network, network.offlineQueue[0].query, network.offlineQueue[0].cb);
+        network.emit('offlineQueuePop', network.offlineQueue.shift());
+
+        setTimeout(() => {
+          dequeuingProcess();
+        }, Math.max(0, network.replayInterval));
+      }
+    };
+
+  if (network.offlineQueueLoader) {
+    if (typeof network.offlineQueueLoader !== 'function') {
+      throw new Error('Invalid value for offlineQueueLoader property. Expected: function. Got: ' + typeof network.offlineQueueLoader);
+    }
+
+    const additionalQueue = network.offlineQueueLoader();
+    if (Array.isArray(additionalQueue)) {
+      network.offlineQueue = additionalQueue
+        .concat(network.offlineQueue)
+        .filter(request => {
+          // throws if the query object does not contain required attributes
+          if (!request.query || request.query.requestId === undefined || !request.query.action || !request.query.controller) {
+            throw new Error('Invalid offline queue request. One or more missing properties: requestId, action, controller.');
+          }
+
+          return uniqueQueue.hasOwnProperty(request.query.requestId) ? false : (uniqueQueue[request.query.requestId] = true);
+        });
+    } else {
+      throw new Error('Invalid value returned by the offlineQueueLoader function. Expected: array. Got: ' + typeof additionalQueue);
+    }
+  }
+
+  dequeuingProcess();
+}
+
+module.exports = AbtractWrapper;