feat: impl subscribeTimeout (#56)

If subscribe listener not be called before subscribeTimeout,
ApiClient will emit timeout error.

.github/workflows/nodejs.yml

@@ -22,7 +22,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        node-version: [8, 10]
+        node-version: [8, 10, 12, 14, 16]
         os: [ubuntu-latest, windows-latest, macos-latest]
 
     steps:

lib/api_client.js

@@ -33,7 +33,7 @@ class APIClientBase extends Base {
 
   close() {
     if (is.function(this._client.close)) {
-      return this._client.close();
+      return this._client.close().then(() => cluster.close(this._client));
     }
     return cluster.close(this._client);
   }

lib/connection.js

@@ -13,7 +13,7 @@ class Connection extends Base {
    *  - {Socket} socket - the socket instance
    *  - {Number} responseTimeout - the response timeout
    *  - {Transcode} transcode - serialze / deserialze methods
-   * @constructor
+   * @class
    */
   constructor(options) {
     super(options);

lib/follower.js

@@ -16,7 +16,8 @@ class Follower extends Base {
    *  - {Map} descriptors - interface descriptors
    *  - {Transcode} transcode - serialze / deserialze methods
    *  - {Number} responseTimeout - the timeout
-   * @constructor
+   *  - {Number} subscribeTimeout - the first subscribe callback timeout
+   * @class
    */
   constructor(options) {
     // local address
@@ -27,6 +28,7 @@ class Follower extends Base {
     this._subData = new Map();
     this._transcode = options.transcode;
     this._closeByUser = false;
+    this._subscribeTimeoutListeners = new Map();
 
     this.on('request', req => this._handleRequest(req));
     // avoid warning message
@@ -107,9 +109,41 @@ class Follower extends Base {
     return '$$inner$$__' + this.options.formatKey(reg);
   }
 
+  _clearSubscribeTimeout(key) {
+    if (!this._subscribeTimeoutListeners.has(key)) {
+      return;
+    }
+    const timeout = this._subscribeTimeoutListeners.get(key);
+    clearTimeout(timeout);
+    this._subscribeTimeoutListeners.delete(key);
+  }
+
+  _setSubscribeTimeout(key) {
+    this._clearSubscribeTimeout(key);
+    const start = Date.now();
+    const subscribeTimeout = this.options.subscribeTimeout;
+    const timeout = setTimeout(() => {
+      const error = new Error(`subscribe timeout for ${key}, cost: ${Date.now() - start}ms, expect is ${subscribeTimeout}ms`);
+      this.emit('error', error);
+    }, subscribeTimeout);
+    this._subscribeTimeoutListeners.set(key, timeout);
+  }
+
+  _listenSubscribeTimeout(key) {
+    this._setSubscribeTimeout(key);
+    const dataListener = () => {
+      this.removeListener(key, dataListener);
+      this._clearSubscribeTimeout(key);
+    };
+    this.on(key, dataListener);
+  }
+
   subscribe(reg, listener) {
     const key = this.formatKey(reg);
     this.on(key, listener);
+    if (this.options.subscribeTimeout) {
+      this._listenSubscribeTimeout(key);
+    }
 
     // no need duplicate subscribe
     if (!this._subInfo.has(key)) {

lib/index.js

@@ -12,7 +12,7 @@ const defaultOptions = {
   port: parseInt(process.env.NODE_CLUSTER_CLIENT_PORT) || 7777,
   singleMode: process.env.NODE_CLUSTER_CLIENT_SINGLE_MODE === '1',
   maxWaitTime: 30000,
-  connectTimeout: parseInt(process.env.NODE_CLUSTER_CLIENT_CONNECT_TIMEOUT)  || 10000,
+  connectTimeout: parseInt(process.env.NODE_CLUSTER_CLIENT_CONNECT_TIMEOUT) || 10000,
   responseTimeout: 3000,
   heartbeatInterval: 20000,
   autoGenerate: true,
@@ -43,7 +43,7 @@ class ClientGenerator {
    *    - {Function} decode - custom deserialize method
    *  - {Boolean} [isLeader|null] - specify whether current instance is leader
    *  - {Number} [maxWaitTime|30000] - leader startup max time (ONLY effective on isLeader is true)
-   * @constructor
+   * @class
    */
   constructor(clientClass, options) {
     this._clientClass = clientClass;

lib/leader.js

@@ -21,7 +21,7 @@ class Leader extends Base {
    *  - {Boolean} isBroadcast - whether broadcast subscrption result to all followers or just one, default is true
    *  - {Number} heartbeatInterval - the heartbeat interval
    *  - {Function} createRealClient - to create the real client
-   * @constructor
+   * @class
    */
   constructor(options) {
     super(options);

lib/protocol/packet.js

@@ -39,7 +39,7 @@ class Packet {
    *   - @param {Number} timeout - The timeout
    *   - @param {Object} connObj - connection object
    *   - @param {Buffer} data - app data
-   * @constructor
+   * @class
    */
   constructor(options) {
     this.id = options.id;

lib/server.js

@@ -69,7 +69,7 @@ class ClusterServer extends Base {
   /**
    * Manage all TCP Connections，assign them to proper channel
    *
-   * @constructor
+   * @class
    * @param {Object} options
    *  - {net.Server} server - the server
    *  - {Number} port - the port

lib/wrapper/base.js

@@ -33,7 +33,7 @@ class WrapperBase extends Base {
    *  - {Boolean} isLeader - wehether is leader or follower
    *  - {Number} maxWaitTime - leader startup max time (ONLY effective on isLeader is true)
    *  - {Function} createRealClient - to create the real client instance
-   * @constructor
+   * @class
    */
   constructor(options) {
     super(options);

package.json

@@ -10,7 +10,7 @@
   "scripts": {
     "autod": "autod",
     "lint": "eslint . --ext .js",
-    "cov": "TEST_TIMEOUT=20000 egg-bin cov",
+    "cov": "egg-bin cov --timeout 20000",
     "test": "npm run lint && npm run test-local",
     "test-local": "egg-bin test",
     "pkgfiles": "egg-bin pkgfiles --check",
@@ -67,6 +67,6 @@
     "node": ">=8.0.0"
   },
   "ci": {
-    "version": "8, 10"
+    "version": "8, 10, 12, 14, 16"
   }
 }

test/subscribe.test.js

@@ -0,0 +1,48 @@
+'use strict';
+
+const assert = require('assert');
+const sleep = require('mz-modules/sleep');
+const ApiClient = require('./supports/sub_timeout/api_client');
+
+describe('test/subscrib.test.js', () => {
+  describe('timeout case', () => {
+    it('should timeout', async () => {
+      const leader = new ApiClient({
+        singleMode: false,
+        isLeader: true,
+      });
+      const follower = new ApiClient({
+        clusterOptions: {
+          subscribeTimeout: 1000,
+        },
+        singleMode: false,
+        isLeader: false,
+      });
+      await follower.ready();
+      const errors = [];
+      const values = [];
+      follower.on('error', err => {
+        errors.push(err);
+      });
+      follower.subscribe({
+        key: 'timeout:500',
+      }, value => {
+        values.push(value);
+      });
+      follower.subscribe({
+        key: 'timeout:1500',
+      }, value => {
+        values.push(value);
+      });
+      await sleep(2000);
+      assert.deepStrictEqual(values, [
+        'hello:500',
+        'hello:1500',
+      ]);
+      assert(errors.length === 1);
+      assert(/subscribe timeout for/.test(errors[0]));
+      await follower.close();
+      await leader.close();
+    });
+  });
+});

test/supports/cluster_server.js

@@ -23,7 +23,7 @@ function startServer(port) {
     get clusterOptions() {
       return {
         port,
-        responseTimeout: 1000,
+        responseTimeout: 10000,
         name: `cluster-server-test-${process.version}`,
       };
     }
@@ -55,7 +55,9 @@ function startServer(port) {
 
     // Fork workers.
     for (let i = 0; i < numCPUs; i++) {
-      cluster.fork();
+      cluster.fork({
+        CLUSTER_PORT: port,
+      });
     }
 
     cluster.on('exit', (worker, code, signal) => {
@@ -105,11 +107,15 @@ function startServer(port) {
 }
 
 const server = net.createServer();
-server.listen(0, () => {
-  const address = server.address();
-  console.log('using port =>', address.port);
-  server.close();
-  setTimeout(() => {
-    startServer(address.port);
-  }, 100);
-});
+if (cluster.isMaster) {
+  server.listen(0, () => {
+    const address = server.address();
+    console.log('using port =>', address.port);
+    server.close(() => {
+      startServer(address.port);
+    });
+  });
+} else {
+  console.log('child process using port =>', process.env.CLUSTER_PORT);
+  startServer(+process.env.CLUSTER_PORT);
+}

test/supports/server.js

@@ -64,7 +64,7 @@ class ClusterServer extends Base {
   /**
    * Manage all TCP Connections，assign them to proper channel
    *
-   * @constructor
+   * @class
    * @param {Object} options
    *  - {net.Server} server - the server
    *  - {Number} port - the port

test/supports/sub_timeout/api_client.js

@@ -0,0 +1,40 @@
+'use strict';
+
+const DataClient = require('./data_client');
+const APIClientBase = require('../../../lib/api_client');
+
+class ApiClient extends APIClientBase {
+  constructor(options = {}) {
+    super(Object.assign({}, options, { initMethod: '_init' }));
+  }
+
+  get DataClient() {
+    return DataClient;
+  }
+
+  get clusterOptions() {
+    return {
+      port: parseInt(process.env.NODE_CLUSTER_CLIENT_PORT || 7777),
+      singleMode: false,
+      subscribeTimeout: 1000,
+    };
+  }
+
+  async _init() {
+    await this._client.ready();
+  }
+
+  subscribe(config, listener) {
+    this._client.subscribe(config, listener);
+  }
+
+  publish(reg) {
+    this._client.publish(reg);
+  }
+
+  close() {
+    return this._client.close();
+  }
+}
+
+module.exports = ApiClient;

test/supports/sub_timeout/data_client.js

@@ -0,0 +1,29 @@
+'use strict';
+
+const Base = require('sdk-base');
+
+class DataClient extends Base {
+  constructor(options = {}) {
+    super(options);
+  }
+
+  async _init() {
+    // ...
+  }
+
+  subscribe(reg, listener) {
+    const { key } = reg;
+    const match = key.match(/timeout:(\d+)/);
+    if (!match) {
+      throw new Error('not a timeout key');
+    }
+    setTimeout(() => {
+      listener('hello:' + match[1]);
+    }, +match[1]);
+  }
+
+  close() {
+  }
+}
+
+module.exports = DataClient;