redesign event

coordinator.js

@@ -12,6 +12,8 @@ const Coordinator = function(param={}) {
 	let that = this;
 	let resettimer = 0;
 
+	this.topicstore = require('./topicstore').create();
+
 	this.x = 0;
 
 	this.started = false;
@@ -20,19 +22,34 @@ const Coordinator = function(param={}) {
 	this.mqttclient = param.mqttclient;
 	this.flowemitter = param.flowemitter   || null;
 
-	this.flowemitter.on('newclient', function(client){
+	this.flowemitter.on('newshadowclient', function(client){
 		let deviceid = client.deviceid;
 		let devicetoken = client.devicetoken;
 
+console.log
+
 		that.mqttclient.subscribe(`@tap/shadow/updated/${deviceid}:${devicetoken}`);
 		that.mqttclient.subscribe(`@tap/device/changed/${deviceid}:${devicetoken}`);
 
 		if (DEBUG_SUBLIST) {
-			console.log(`\nshadow client #${deviceid} connected -->`)
-			console.log(that.flowemitter.sublist);
+			// console.log(`\nshadow client #${deviceid} connected -->`)
+			// console.log(that.flowemitter.sublist);
+		}
+	});
+
+	this.flowemitter.on('newmsgclient', function(client){
+		let deviceid = client.deviceid;
+		let devicetoken = client.devicetoken;
+
+		that.mqttclient.subscribe(`@tap/device/changed/${deviceid}:${devicetoken}`);
+
+		if (DEBUG_SUBLIST) {
+			// console.log(`\nshadow client #${deviceid} connected -->`)
+			// console.log(that.flowemitter.sublist);
 		}
 	});
 
+
 	this.flowemitter.on('flowpub', function(topic, payload, client){
 		let outtopic, outmsg;
 		let deviceid = client.deviceid;
@@ -65,14 +82,34 @@ const Coordinator = function(param={}) {
 
 
 	this.flowemitter.on('flowsub', function(topic, client) {
+		console.log(client.nodeid + '  flowsub --> '+topic)
+
 		let deviceid = client.deviceid;
 		let devicetoken = client.devicetoken;
 		if (topic.startsWith('@local/msgin/')) {
 			let msgpart = topic.split('/').splice(4).join('/');
 
 			outtopic = `@tap/msg/topic/${deviceid}:${devicetoken}/${msgpart}`;
-			that.mqttclient.subscribe(outtopic);
+			if (that.topicstore.addTopic(outtopic, client.nodeid)) {
+				that.mqttclient.subscribe(outtopic);
+			}
+		}
+		else {
+		}
+	});
+
+	this.flowemitter.on('flowunsub', function(topic, client) {
+		console.log(client.nodeid + '  flowunsub --> '+topic)
 
+		let deviceid = client.deviceid;
+		let devicetoken = client.devicetoken;
+		if (topic.startsWith('@local/msgin/')) {
+			let msgpart = topic.split('/').splice(4).join('/');
+			outtopic = `@tap/msg/topic/${deviceid}:${devicetoken}/${msgpart}`;
+			if (that.topicstore.delTopic(outtopic, client.nodeid)) {
+				//console.log('----- unsubscribe : '+outtopic)
+				that.mqttclient.unsubscribe(outtopic);
+			}
 		}
 		else {
 		}

flowemitter.js

@@ -12,11 +12,11 @@ const FlowEmitter = function(param = {}) {
 FlowEmitter.prototype.start = function() {
     let that = this;
 
-    this.red.events.on(`${that.namespace}-register`, function(client) {
-    	that.emit('newclient', client);
+    this.red.events.on(`${that.namespace}-registershadowlistener`, function(client) {
+    	that.emit('newshadowclient', client);
     });
 
-    this.red.events.on(`${that.namespace}-deregister`, function(client) {
+    this.red.events.on(`${that.namespace}-deregistershadowlistener`, function(client) {
     	that.emit('startcountdownreset');
 	});
 
@@ -25,6 +25,10 @@ FlowEmitter.prototype.start = function() {
     	that.emit('flowsub', topic, client);
     });
 
+    this.red.events.on(`${that.namespace}-flowunsub`, function(topic, client) {
+    	that.emit('flowunsub', topic, client);
+    });
+
     this.red.events.on(`${that.namespace}-flowpub`, function(topic, payload, client) {
     	that.emit('flowpub', topic, payload, client);
     });