add mutex and fix bug

arduino-cloud.js

@@ -30,6 +30,8 @@ module.exports = function (RED) {
               else
                 this.status({});
             });
+          }else{
+            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(); });

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

@@ -116,7 +116,7 @@ class ArduinoClientMqtt {
           new ArduinoCloudError(errorCode, errorMessage),
         );
       });
-      
+
       client.on("message", (topic, msg) => {
         if (topic.indexOf('/s/o') > -1) {
           client.topics[topic].forEach((cb) => {
@@ -562,7 +562,7 @@ class ArduinoClientMqtt {
       topic: propOutputTopic,
       cb,
     };
-
+    this.numSubscriptions++;
     if (!this.propertyCallback[propOutputTopic]) {
       this.propertyCallback[propOutputTopic] = {};
       this.propertyCallback[propOutputTopic][name] = cb;
@@ -571,7 +571,6 @@ class ArduinoClientMqtt {
 
     if (this.propertyCallback[propOutputTopic] && !this.propertyCallback[propOutputTopic][name]) {
       this.propertyCallback[propOutputTopic][name] = cb;
-      this.numSubscriptions++;
     }
     return Promise.resolve(propOutputTopic);
   };

package.json

@@ -20,6 +20,7 @@
   "dependencies": {
     "@arduino/arduino-iot-client": "^1.0.1",
     "@arduino/cbor-js": "github:arduino/cbor-js",
+    "async-mutex": "^0.1.4",
     "async-request": "^1.2.0",
     "jws": "^3.2.2",
     "lodash": "^4.17.15",

utils/arduino-connection-manager.js

@@ -4,7 +4,7 @@ const ArduinoClientMqtt = require ('../arduino-iot-client-mqtt/arduino-iot-clien
 const accessTokenUri = process.env.NODE_RED_ACCESS_TOKEN_URI || 'https://login.arduino.cc/oauth/token';
 const accessTokenAudience = process.env.NODE_RED_ACCESS_TOKEN_AUDIENCE || 'https://api2.arduino.cc/iot';
 const arduinoCloudHost = process.env.NODE_RED_MQTT_HOST || 'wss.iot.arduino.cc';
-
+const Mutex = require('async-mutex').Mutex;
 /**
  * {
  *  clientId: clientId,
@@ -17,6 +17,9 @@ const arduinoCloudHost = process.env.NODE_RED_MQTT_HOST || 'wss.iot.arduino.cc';
  * }
  */
 var connections=[];
+const getClientMutex = new Mutex();
+
+
 
 async function getToken(connectionConfig){
   var options = {
@@ -48,9 +51,11 @@ async function getToken(connectionConfig){
 }
 
 async function getClientMqtt(connectionConfig){
+
   if (!connectionConfig || !connectionConfig.credentials) {
     throw new Error("Cannot find cooonection config or credentials.");
   }
+  const releaseMutex = await getClientMutex.acquire();
   try{
   let user = findUser(connectionConfig.credentials.clientid);
   let clientMqtt;
@@ -102,17 +107,22 @@ async function getClientMqtt(connectionConfig){
       connections[user].clientMqtt=clientMqtt;
     }
   }
+  releaseMutex();
+
   return clientMqtt;
   }catch(err){
     console.log(err);
+    releaseMutex();
   }
 
 }
 
 async function getClientHttp(connectionConfig){
+
   if (!connectionConfig || !connectionConfig.credentials) {
     throw new Error("Cannot find cooonection config or credentials.");
   }
+  const releaseMutex = await getClientMutex.acquire();
   try{
   var user = findUser(connectionConfig.credentials.clientid);
   var clientHttp;
@@ -143,10 +153,14 @@ async function getClientHttp(connectionConfig){
       connections[user].clientHttp=clientHttp;
     }
   }
+  releaseMutex();
   return clientHttp;
   }catch(err){
     console.log(err);
+    releaseMutex();
+
   }
+
 }
 
 function findUser(clientId) {