impl addMessageHandler

src/server/service/config-manager.js

@@ -46,6 +46,9 @@ class ConfigManager {
    */
   async setPubsub(configPubsub) {
     this.configPubsub = configPubsub;
+    this.configPubsub.addMessageHandler((message) => {
+      logger.info('message recieved', message);
+    });
   }
 
   /**

src/server/service/config-pubsub/base.js

@@ -16,6 +16,10 @@ class ConfigPubsubDelegator {
     throw new Error('implement this');
   }
 
+  addMessageHandler(handler) {
+    throw new Error('implement this');
+  }
+
 }
 
 module.exports = ConfigPubsubDelegator;

src/server/service/config-pubsub/nchan.js

@@ -15,6 +15,7 @@ class NchanDelegator extends ConfigPubsubDelegator {
     this.subscribePath = subscribePath;
 
     this.channelId = channelId;
+    this.messageHandlers = [];
 
     this.client = null;
     this.connection = null;
@@ -61,6 +62,24 @@ class NchanDelegator extends ConfigPubsubDelegator {
     throw new Error('implement this');
   }
 
+  /**
+   * @inheritdoc
+   */
+  addMessageHandler(handler) {
+    this.messageHandlers.push(handler);
+
+    if (this.connection != null) {
+      this.connection.on('message', (message) => {
+        if (message.type === 'utf8') {
+          handler(message.utf8Data);
+        }
+        else {
+          logger.warn('Only utf8 message is supported.');
+        }
+      });
+    }
+  }
+
   initClient() {
     const client = new WebSocketClient();
 
@@ -77,13 +96,11 @@ class NchanDelegator extends ConfigPubsubDelegator {
       connection.on('close', () => {
         logger.info('WebSocket connection closed');
       });
-      connection.on('message', (message) => {
-        if (message.type === 'utf8') {
-          logger.info(message.utf8Data);
-        }
-      });
 
       this.connection = connection;
+
+      // register all message handlers
+      this.messageHandlers.forEach(handler => this.addMessageHandler(handler));
     });
 
     this.client = client;