add sharing topic

Author: fabik111

arduino-cloud.js

@@ -17,7 +17,7 @@ module.exports = function (RED) {
           this.propertyVariableName = config.variableName;
           this.arduinoClient = await connectionManager.getClientMqtt(connectionConfig, RED);
           if (this.arduinoClient && this.arduinoClient.connection.connected) {
-            await this.arduinoClient.onPropertyValue(this.thing, this.propertyVariableName, (msg) => {
+            await this.arduinoClient.onPropertyValue(this.thing, this.propertyVariableName, config.id,(msg) => {
               this.send(
                 {
                   topic: this.propertyName,
@@ -35,7 +35,7 @@ module.exports = function (RED) {
             this.status({ fill: "red", shape: "ring", text: "Connection Error" });
           }
           this.on('close', function (done) {
-            connectionManager.deleteClientMqtt(connectionConfig.credentials.clientid, this.thing, this.propertyName).then(() => { done(); });
+            connectionManager.deleteClientMqtt(connectionConfig.credentials.clientid, this.thing, this.propertyName,config.id).then(() => { done(); });
           });
 
           //this.poll(connectionConfig);

arduino-iot-client-mqtt/arduino-iot-client-mqtt.js

@@ -165,7 +165,9 @@ class ArduinoClientMqtt {
             }
             if (propertyNameKeyPrevious !== propertyNameKey) {
               if (this.propertyCallback[topic][propertyNameKeyPrevious]) {
-                this.propertyCallback[topic][propertyNameKeyPrevious](valueToSend);
+                for(var i=0; i<this.propertyCallback[topic][propertyNameKeyPrevious].length; i++){
+                 this.propertyCallback[topic][propertyNameKeyPrevious][i].callback(valueToSend);
+                }
               }
               propertyNameKeyPrevious = propertyNameKey;
               valueToSend = {};
@@ -178,9 +180,11 @@ class ArduinoClientMqtt {
             }
           });
           if (valueToSend !== {} && this.propertyCallback[topic][propertyNameKey]) {
-            this.propertyCallback[topic][propertyNameKey](valueToSend);
+            for(var i=0; i<this.propertyCallback[topic][propertyNameKey].length; i++){
+             this.propertyCallback[topic][propertyNameKey][i].callback(valueToSend);
+            }
           }
-        }
+       }
       });
 
       if (typeof this.opts.onDisconnect === 'function') {
@@ -560,7 +564,7 @@ class ArduinoClientMqtt {
     return arrayBufferToBase64(cborEncoded);
   };
 
-  onPropertyValue(thingId, name, cb) {
+  onPropertyValue(thingId, name,nodeId, cb) {
     if (!name) {
       throw new Error('Invalid property name');
     }
@@ -576,23 +580,50 @@ class ArduinoClientMqtt {
     this.numSubscriptions++;
     if (!this.propertyCallback[propOutputTopic]) {
       this.propertyCallback[propOutputTopic] = {};
-      this.propertyCallback[propOutputTopic][name] = cb;
+      this.propertyCallback[propOutputTopic][name] = [];
+      this.propertyCallback[propOutputTopic][name].push({
+        nodeId: nodeId,
+        callback:cb
+      });
+
       return this.subscribe(propOutputTopic, cb);
     }
 
     if (this.propertyCallback[propOutputTopic] && !this.propertyCallback[propOutputTopic][name]) {
-      this.propertyCallback[propOutputTopic][name] = cb;
+      this.propertyCallback[propOutputTopic][name] = [];
+      this.propertyCallback[propOutputTopic][name].push({
+        nodeId: nodeId,
+        callback:cb
+      });
+    }else if(this.propertyCallback[propOutputTopic] && this.propertyCallback[propOutputTopic][name]){
+      this.propertyCallback[propOutputTopic][name].push({
+        nodeId: nodeId,
+        callback:cb
+      });
     }
     return Promise.resolve(propOutputTopic);
   };
 
 
-  removePropertyValueCallback(thingId, name) {
+  removePropertyValueCallback(thingId, name, nodeId) {
     if (!name) {
       throw new Error('Invalid property name');
     }
     const propOutputTopic = `/a/t/${thingId}/e/o`;
-    delete this.propertyCallback[propOutputTopic][name];
+    var pos=-1;
+    for(var i=0; i<this.propertyCallback[propOutputTopic][name].length; i++){
+      var cbObject=this.propertyCallback[propOutputTopic][name][i];
+      if(cbObject.nodeId===nodeId){
+        pos=i;
+        break;
+      }
+    }
+    if(pos!=-1){
+      this.propertyCallback[propOutputTopic][name].splice(pos,1);
+    }
+    if(this.propertyCallback[propOutputTopic][name].length===0){
+      delete this.propertyCallback[propOutputTopic][name];
+    }
     this.numSubscriptions--;
 
     return Promise.resolve(this.numSubscriptions);

utils/arduino-connection-manager.js

@@ -229,12 +229,12 @@ async function updateToken(connectionConfig) {
   }
 }
 
-async function deleteClientMqtt(clientId, thing, propertyName) {
+async function deleteClientMqtt(clientId, thing, propertyName, nodeId) {
   const releaseMutex = await getClientMutex.acquire();
   var user = findUser(clientId);
   if (user !== -1) {
     if (connections[user].clientMqtt !== null) {
-      var ret = await connections[user].clientMqtt.removePropertyValueCallback(thing, propertyName);
+      var ret = await connections[user].clientMqtt.removePropertyValueCallback(thing, propertyName,nodeId);
 
       if (ret === 0) {
         await connections[user].clientMqtt.disconnect();