update

coordinator.js

@@ -99,17 +99,14 @@ const Coordinator = function(param={}) {
 							outmsg = packet.payload.toString();
 							outmsg = packet.payload.toString();
 							that.remoteclient.publish(outtopic, outmsg);
 							that.remoteclient.publish(outtopic, outmsg);
 						}
 						}
-
 		}
 		}
     });
     });
 	*/
 	*/
 
 
 	this.flowemitter.on('flowpub', function(topic, payload, client){
 	this.flowemitter.on('flowpub', function(topic, payload, client){
-		console.log('client published', client.id);
-
+		//console.log('client published', client.id);
 		let outtopic, outmsg;
 		let outtopic, outmsg;
 		let deviceid = client.id;
 		let deviceid = client.id;
-
 		switch (topic) {
 		switch (topic) {
 			case '@shadow/data/update' :
 			case '@shadow/data/update' :
 						outtopic = `@tap/shadow/update/${deviceid}:${client.token}`;
 						outtopic = `@tap/shadow/update/${deviceid}:${client.token}`;
@@ -139,7 +136,7 @@ const Coordinator = function(param={}) {
 
 
 
 
 	this.flowemitter.on('flowsub', function(topic, client) {
 	this.flowemitter.on('flowsub', function(topic, client) {
-		let deviceid = client.id.split(':')[0];
+		let deviceid = client.id;
 		// if client try to subscribe @msg/xxx/yyy --> flowagent subscribe @tap on a remote broker
 		// if client try to subscribe @msg/xxx/yyy --> flowagent subscribe @tap on a remote broker
 		if (topic.startsWith('@local/msgin/')) {
 		if (topic.startsWith('@local/msgin/')) {
 			let msgpart = topic.split('/').splice(4).join('/');
 			let msgpart = topic.split('/').splice(4).join('/');
@@ -169,12 +166,19 @@ const Coordinator = function(param={}) {
 
 
 	this.remoteclient.on('message', function(topic, payload){
 	this.remoteclient.on('message', function(topic, payload){
 
 
+console.log("Incoming MQTT message-------------------------------------------------");
+console.log({
+	topic,
+	payload: payload.toString()
+})
+
 		let jsonpayload = {};
 		let jsonpayload = {};
 		try {
 		try {
 			jsonpayload = JSON.parse(payload.toString());
 			jsonpayload = JSON.parse(payload.toString());
 		}
 		}
 		catch(e) {}
 		catch(e) {}
 
 
+
 		if (topic.startsWith('@shadow/data/updated')) {
 		if (topic.startsWith('@shadow/data/updated')) {
 			let newpayload = cache.mergeShadow(jsonpayload.deviceid, jsonpayload);
 			let newpayload = cache.mergeShadow(jsonpayload.deviceid, jsonpayload);
 		    // that.flowemitter.publish(`${jsonpayload.deviceid}/shadow/merged`, JSON.stringify(newpayload) );
 		    // that.flowemitter.publish(`${jsonpayload.deviceid}/shadow/merged`, JSON.stringify(newpayload) );
@@ -213,11 +217,11 @@ const Coordinator = function(param={}) {
 		    });
 		    });
         }
         }
         else if (topic.startsWith('@msg/')) {
         else if (topic.startsWith('@msg/')) {
-			let part = topic.split('/').splice(1).join('/');
-			let localtopic = `@local/msgin/${part}`;
+			// let part = topic.split('/').splice(1).join('/');
+			// let localtopic = `@msg/${part}`;
 			
 			
-    		that.flowemitter.pub(`flow:message`, {
-    			topic: localtopic,
+    		that.flowemitter.pub(`flow:msg`, {
+    			topic: topic,
     			payload: payload
     			payload: payload
     		});
     		});