redesign subscription flow

coordinator.js

@@ -5,32 +5,65 @@ const DEBUG_SUBLIST = false;
 const events = require('events');
 const cache = require('./cache');
 
+const RESETDELAY = 1000
+
 const Coordinator = function(param={}) {
 	let that = this;
+	let resettimer = 0;
 
 	this.events = new events.EventEmitter;
 	this.remoteclient = param.remoteclient;
 	this.flowemitter = param.flowemitter   || null;
 
+	// function resetSubscription() {
+
+	// }
+
+	// function checkupSubscription() {
+	// 	let nodeList = that.flowemitter.getDeviceList();
+	// 	let sublist = {};
+	// 	let topic = '';
+
+	// 	function safeSub(topic) {
+	// 		if (topic && !sublist[topic]) {
+	// 			that.remoteclient.subscribe(topic);
+	// 			sublist[topic] = true;
+	// 		}
+	// 	}
+
+	// 	for (let nodeid in nodeList) {
+	// 		if (!nodeList[nodeid].subbbed) {
+
+	// 			safeSub(`@tap/shadow/updated/${deviceid}:${devicetoken}`);
+	// 			safeSub(`@tap/device/changed/${deviceid}:${devicetoken}`);
+
+	// 			let tp;
+	// 			for (let i=0; i<nodeList[nodeid].msgtopic.length; i++) {
+	// 				safeSub(nodeList[nodeid].msgtopic[i]);
+	// 			}
+	// 			nodeList[nodeid].subbbed = true;
+	// 		}
+	// 	}
+	// }
+
 
 	this.flowemitter.on('newclient', function(client){
 		let deviceid = client.deviceid;
 		let devicetoken = client.devicetoken;
 
-		let nodeList = that.flowemitter.getDeviceList();
-		
-		if (nodeList[client.nodeid] && !nodeList[client.nodeid].subbed ) {
-			nodeList[client.nodeid].subbed = true;
+		//let nodeList = that.flowemitter.getDeviceList();
+		// if (nodeList[client.nodeid] && !nodeList[client.nodeid].subbed ) {
+		// 	nodeList[client.nodeid].subbed = true;
+		// let deviceid = nodeList[client.nodeid].deviceid;
+		// let devicetoken = nodeList[client.nodeid].devicetoken;
 
-			let deviceid = nodeList[client.nodeid].deviceid;
-			let devicetoken = nodeList[client.nodeid].devicetoken;
+		that.remoteclient.subscribe(`@tap/shadow/updated/${deviceid}:${devicetoken}`);
+		that.remoteclient.subscribe(`@tap/device/changed/${deviceid}:${devicetoken}`);
 
-			that.remoteclient.subscribe(`@tap/shadow/updated/${deviceid}:${devicetoken}`);
-			that.remoteclient.subscribe(`@tap/device/changed/${deviceid}:${devicetoken}`);
+		// that.remoteclient.publish(`@tap/device/get/${deviceid}:${devicetoken}`);
+		// that.remoteclient.publish(`@tap/shadow/get/${deviceid}:${devicetoken}`);			
 
-			that.remoteclient.publish(`@tap/device/get/${deviceid}:${devicetoken}`);
-			that.remoteclient.publish(`@tap/shadow/get/${deviceid}:${devicetoken}`);			
-		}
+		//}
 
 		if (DEBUG_SUBLIST) {
 			console.log(`\nshadow client #${deviceid} connected -->`)
@@ -65,7 +98,6 @@ const Coordinator = function(param={}) {
 							outmsg = payload.toString();
 							that.remoteclient.publish(outtopic, outmsg);
 						}
-
 		}
     });
 
@@ -73,20 +105,29 @@ const Coordinator = function(param={}) {
 	this.flowemitter.on('flowsub', function(topic, client) {
 		let deviceid = client.deviceid;
 		let devicetoken = client.devicetoken;
-		// if client try to subscribe @msg/xxx/yyy --> flowagent subscribe @tap on a remote broker
 		if (topic.startsWith('@local/msgin/')) {
 			let msgpart = topic.split('/').splice(4).join('/');
+
 			outtopic = `@tap/msg/topic/${deviceid}:${devicetoken}/${msgpart}`;
 			that.remoteclient.subscribe(outtopic);
+
 		}
 		else {
 		}
 	});
 
+	this.flowemitter.on('startcountdownreset', function() {
+		if (!resettimer) {
+			resettimer = setTimeout(function() {
+				console.log('                  [info] MQTT resetting...');
+				that.remoteclient.resetbroker();
+			}, RESETDELAY);
+		}
+	});
+
 	this.remoteclient.on('connect', function() {
 		console.log('                  [info] Connected to MQTT broker.')
 		that.remoteclient.subscribe('@private/#');
-		that.emit('mqttconnected');
 		
 		// setTimeout(function() {
 		// 	for (let deviceid in that.flowemitter.sublist) {
@@ -101,7 +142,6 @@ const Coordinator = function(param={}) {
 
 	this.remoteclient.on('close', function() {
 		console.log('                  [info] Disconnect from MQTT broker.')
-		that.emit('mqttdisconnected');
 	});
 
 	this.remoteclient.on('message', function(topic, payload){
@@ -136,7 +176,7 @@ const Coordinator = function(param={}) {
 			cache.setStatus(jsonpayload.deviceid, jsonpayload);
 		    that.flowemitter.pub(`flow:device:${jsonpayload.deviceid}`, {
 		    	deviceget : jsonpayload
-		    });
+		    })
         }
         else if (topic.startsWith('@msg/')) {
     		that.flowemitter.pub(`flow:msg`, {

flowagent.js

@@ -20,6 +20,10 @@ let FlowAgent = function(option = {}) {
     	red : option.red || null
     }
 
+    if (this.option.red) {
+    	this.option.red.events.setMaxListeners(100);
+    }
+
 	this.remoteclient = MQTTClient.create({
 	    host: that.option.broker_uri,
 	    options: {

flowemitter.js

@@ -5,8 +5,8 @@ const events = require('events');
 const FlowEmitter = function(param = {}) {
 	this.red = param.red;		// node-red object
 
-	this.devicelist = {};
-	this.sublist = {};
+	//this.devicelist = {};
+	//this.sublist = {};
 	this.globalcontext = param.globalcontext;
 
 	this.events = new events.EventEmitter;	
@@ -17,19 +17,22 @@ FlowEmitter.prototype.start = function() {
 
     this.red.events.on('register', function(client) {
 
-		that.devicelist[client.nodeid] = {
-			deviceid : client.deviceid,
-			devicetoken: client.devicetoken,
-			msgtopic : []
-		}
+		// that.devicelist[client.nodeid] = {
+		// 	deviceid : client.deviceid,
+		// 	devicetoken: client.devicetoken,
+		// 	msgtopic : [],
+		// 	subbed : false
+		// }
     	that.emit('newclient', client);
     });
 
     this.red.events.on('deregister', function(client) {
-    	delete that.devicelist[client.nodeid];
+    	//delete that.devicelist[client.nodeid];
+    	that.emit('startcountdownreset');
 	});
 
     this.red.events.on('flowsub', function(topic, client) {
+		//that.devicelist[client.nodeid].msgtopic.push(topic);
     	that.emit('flowsub', topic, client);
     });
 
@@ -39,9 +42,16 @@ FlowEmitter.prototype.start = function() {
 
 }
 
-FlowEmitter.prototype.getDeviceList = function() {
-	return this.devicelist;
-}
+// FlowEmitter.prototype.getDeviceList = function() {
+// 	return this.devicelist;
+// }
+
+// FlowEmitter.prototype.clearDeviceSubscriptionState = function() {
+// 	for (let nodeid in this.devicelist) {
+// 		this.devicelist[nodeid].subbed = false;
+// 	}
+// }
+
 
 FlowEmitter.prototype.pub = function(topic, payload) {
 	this.red.events.emit(topic, payload);

mqttclient.js

@@ -9,13 +9,22 @@ const MQTTClient = function(param = {}) {
     this.options = param.options;
 }
 
+function sleep(ms) {
+    return new Promise(resolve => setTimeout(resolve, ms));
+}
 MQTTClient.prototype = new events.EventEmitter;
 
 MQTTClient.prototype.connect = function() {
     var that = this;
+
+    this.subscription = [];
+
     this.client = MQTT.connect(this.host, this.options || {});
 
-    this.client.on('connect', function() {
+    this.client.on('connect', async function() {
+        for (let topic in that.subscription) {
+            await sleep(100);
+        }
         that.emit('connect');
     });
 
@@ -38,10 +47,20 @@ MQTTClient.prototype.connect = function() {
 
 }
 
+MQTTClient.prototype.resetbroker = function(topic, payload) {
+    this.client.end();
+    //this.subscription = {};
+    this.client.reconnect();
+}
+
 MQTTClient.prototype.publish = function(topic, payload) {
 	if (this.client) {
-        console.log('              mqtt pub -->   '+topic);
-	    this.client.publish(topic, payload);
+        //console.log('              mqtt pub -->   '+topic);
+        
+        if (this.subscription[topic]) this.subscription[topic]++;
+        else this.subscription[topic] = 1;
+	    
+        this.client.publish(topic, payload);
 	}
 	else {