[NRRPLT-8384-mqtt Updated onMessage function]

src/components/nrp-core-dashboard/nrp-core-dashboard.js

 import React from 'react';
 
 import MqttClientService from '../../services/mqtt-client-service';
+//import subscribeToTopic from MqttClientService;
 
 export default class NrpCoreDashboard extends React.Component {
   constructor(props) {
@@ -14,13 +15,15 @@ export default class NrpCoreDashboard extends React.Component {
     MqttClientService.instance.connect(this.mqttBrokerUrl);
   }
 
-  onMqttClientConnected(mqttClient) {
-    mqttClient.subscribeToTopic('test_topic');
-    //mqttClient.subscribe('#', (err) => {
-    //  if (err) {
-    //    console.error(err);
-    //  }
-    //};
+  onMqttClientConnected(MqttClient) {
+    MqttClient.subscribe('#', (err) => {
+      if (err) {
+        console.error(err);
+      }
+    });
+    // As a test to make sure MqttClientService can subscribe to multiple topics at once we use these two for testing
+    let token = MqttClientService.instance.subscribeToTopic('test_topic', (param1) => (console.info(param1)));
+    token = MqttClientService.instance.subscribeToTopic('test_topic_proto', (param1) => (console.info(param1)));
   }
 
   render() {

src/services/mqtt-client-service.js

@@ -2,9 +2,9 @@ import mqtt from 'mqtt';
 import { EventEmitter } from 'events';
 
 //import * as proto from 'nrp-jsproto/nrp-engine_msgs-protobuf.js';
-import { DataPackMessage } from '../../node_modules/nrp-jsproto/engine_grpc_pb.js';
-import jspb from 'google-protobuf';
-//import { hasSubscribers } from 'diagnostics_channel';
+//import { DataPackMessage } from 'nrp-jsproto/nrp-engine_msgs-protobuf.js';
+import jspb from '../../node_modules/google-protobuf/google-protobuf';
+import { hasSubscribers } from 'diagnostics_channel';
 
 let _instance = null;
 const SINGLETON_ENFORCER = Symbol();
@@ -21,7 +21,7 @@ export default class MqttClientService extends EventEmitter {
 
     this.subTokensMap = new Map();
 
-    console.info(DataPackMessage);
+    //console.info(DataPackMessage);
   }
 
   static get instance() {
@@ -41,7 +41,9 @@ export default class MqttClientService extends EventEmitter {
       this.emit(MqttClientService.EVENTS.CONNECTED, this.client);
     });
     this.client.on('error', this.onError);
-    this.client.on('message', this.onMessage);
+    this.client.on('message', (params) => {
+      this.onMessage(params);
+    });
   }
 
   onError(error) {
@@ -49,10 +51,19 @@ export default class MqttClientService extends EventEmitter {
   }
 
   onMessage(topic, payload, packet) {
-    console.info('MQTT message: [topic, payload, packet]');
-    console.info([topic, payload, packet]);
+    //console.info('MQTT message: [topic, payload, packet]');
+    //console.info([topic, payload, packet]);
+    //Deserializatin of Data must happen here
+    //Now we see which callbacks have been assigned for a topic
+    if (typeof this.subTokensMap.get(topic) !== 'undefined') {
+      for (var token in this.subTokensMap.get(topic)){
+        if (typeof token.callback === 'function' && payload !== 'undefined'){
+          token.callback(payload);
+        }
+      };
+    };
 
-    try {
+    /*try {
       if (topic.endsWith('/type')) {
         let msg = String(payload);
         console.info('"' + topic + '" message format = ' + msg);
@@ -65,9 +76,10 @@ export default class MqttClientService extends EventEmitter {
     }
     catch (error) {
       console.error(error);
-    }
+    }*/
   }
 
+  //callback should have args topic, payload
   subscribeToTopic(topic, callback=Function()){
     const token = {
       topic: topic,
@@ -86,6 +98,7 @@ export default class MqttClientService extends EventEmitter {
       );
     }
     console.info('You have been subscribed to topic ' + topic);
+    console.info(this.subTokensMap);
     return token;
   }
 
@@ -93,13 +106,13 @@ export default class MqttClientService extends EventEmitter {
     return jspb.Message.getField(protoMsg, oneofCaseNumber);
   }
 
-  static getDataPackMessageOneofCaseString(protoMsg) {
+  /*static getDataPackMessageOneofCaseString(protoMsg) {
     for (let dataCase in DataPackMessage.DataCase) {
       if (DataPackMessage.DataCase[dataCase] === protoMsg.getDataCase()) {
         return dataCase;
       }
     }
-  }
+  }*/
 }
 
 MqttClientService.EVENTS = Object.freeze({