fix: AppWorkerClient subscribe same data failed issue (#110)

lib/core/app_worker_client.js

@@ -6,6 +6,7 @@ const Base = require('sdk-base');
 
 const MAX_VALUE = Math.pow(2, 31) - 10;
 const PROCESS_ID = String(process.pid);
+const hasOwnProperty = Object.prototype.hasOwnProperty;
 
 const defaultOptions = {
   responseTimeout: '5s',
@@ -29,7 +30,7 @@ class AppWorkerClient extends Base {
 
     super();
 
-    // hsf 服务依赖肯定会比较多的，设置为 100 个
+    // 服务依赖肯定会比较多的，设置为 100 个
     this.setMaxListeners(100);
 
     this.options = {};
@@ -230,14 +231,18 @@ class AppWorkerClient extends Base {
   _subscribe(info, listener) {
     const key = this._formatKey(info);
     this.on(key, listener);
-    if (!this._subscriptions.has(key)) {
+    const subInfo = this._subscriptions.get(key);
+    if (!subInfo) {
       const subData = {
         key,
         info,
         pid: this.pid,
       };
-      this._subscriptions.set(key, subData);
+      this._subscriptions.set(key, { subData });
       this._sendToAgent(this.commands.sub, subData);
+    } else if (hasOwnProperty.call(subInfo, 'value')) {
+      // trigger listener immediately
+      listener(subInfo.value);
     }
     return this;
   }
@@ -299,7 +304,7 @@ class AppWorkerClient extends Base {
     // 重新订阅
     for (const key of this._subscriptions.keys()) {
       const info = this._subscriptions.get(key);
-      this._sendToAgent(this.commands.sub, info);
+      this._sendToAgent(this.commands.sub, info.subData);
     }
 
     this.logger.info('[egg:worker] [%s] reSubscribe done for "agent restart"', this.name);
@@ -345,6 +350,8 @@ class AppWorkerClient extends Base {
     const key = data.key;
     if (this._subscriptions.has(key)) {
       this.logger.info('[egg:worker] [%s] key[%s] value changed, new value: %j', this.name, key, data.value);
+      const info = this._subscriptions.get(key);
+      info.value = data.value;
       this.emit(key, data.value);
     }
   }

test/fixtures/apps/agent-client-app/agent.js

@@ -0,0 +1,19 @@
+'use strict';
+
+module.exports = function initAgent(agent) {
+  const client = {
+    'mock-data': 'mock-data',
+    'not-exist-data': null,
+    ready(cb) {
+      setImmediate(cb);
+    },
+  };
+
+  agent.startAgent({
+    name: 'sub-client',
+    client,
+    subscribe(info, listener) {
+      listener(client[info]);
+    },
+  });
+};

test/fixtures/apps/agent-client-app/app.js

@@ -0,0 +1,10 @@
+'use strict';
+
+module.exports = function initApp(app) {
+  app.subClient = app.createAppWorkerClient('sub-client', {
+    subscribe(info, listener) {
+      this._subscribe(info, listener);
+      return this;
+    },
+  });
+};

test/fixtures/apps/agent-client-app/app/router.js

@@ -0,0 +1,37 @@
+'use strict';
+
+module.exports = function(app) {
+  const done = app.readyCallback('app_subscribe_data');
+  app.subClient.subscribe('mock-data', val => {
+    app.mockData = val;
+    done();
+  });
+
+  const done1 = app.readyCallback('app_subscribe_not_exist_data');
+  app.subClient.subscribe('not-exist-data', val => {
+    app.notExistData = val;
+    done1();
+  });
+
+  app.get('/', function*() {
+    const val = yield cb => app.subClient.subscribe('mock-data', val => {
+      cb(null, val);
+    });
+
+    this.body = {
+      'mock-data': val,
+      'app-mock-data': app.mockData,
+    };
+  });
+
+  app.get('/not-exist', function*() {
+    const val = yield cb => app.subClient.subscribe('not-exist-data', val => {
+      cb(null, val);
+    });
+
+    this.body = {
+      'not-exist-data': null,
+      'app-not-exist-data': null,
+    };
+  });
+};

test/fixtures/apps/agent-client-app/package.json

@@ -0,0 +1,3 @@
+{
+  "name": "agent-client-app"
+}