Merge pull request #21 from 123joshuawu/add-async-connect

Add connectAsync

index.js

@@ -139,5 +139,47 @@ module.exports = {
 
     return asyncClient;
   },
+  connectAsync (brokerURL, opts, allowRetries=true) {
+    const client = mqtt.connect(brokerURL, opts);
+    const asyncClient = new AsyncClient(client);
+
+    return new Promise((resolve, reject) => {
+      // Listeners added to client to trigger promise resolution
+      const promiseResolutionListeners = {
+        connect: (connack) => {
+          removePromiseResolutionListeners();
+          resolve(asyncClient);   // Resolve on connect
+        },
+        end: () => {
+          removePromiseResolutionListeners();
+          resolve(asyncClient);   // Resolve on end
+        },
+        error: (err) => {
+          removePromiseResolutionListeners();
+          client.end();
+          reject(err);            // Reject on error
+        }
+      };
+
+      // If retries are not allowed, reject on close
+      if (allowRetries === false) {
+        promiseResolutionListeners.close = () => {
+          promiseResolutionListeners.error("Couldn't connect to server");
+        }
+      }
+
+      // Remove listeners added to client by this promise
+      const removePromiseResolutionListeners = () => {
+        Object.keys(promiseResolutionListeners).forEach((eventName) => {
+          client.off(eventName, promiseResolutionListeners[eventName]);
+        });
+      };
+
+      // Add listeners to client
+      Object.keys(promiseResolutionListeners).forEach((eventName) => {
+        client.on(eventName, promiseResolutionListeners[eventName]);
+      });
+    });
+  },
   AsyncClient
 };

test.js

@@ -27,6 +27,20 @@ function runTests () {
     })
   });
 
+  test('ConnectAsync should return AsyncClient after connection', t => {
+    t.plan(1);
+
+    AsyncMQTT.connectAsync(SERVER_URL, {}, false).then((client) => {
+      t.ok(client.connected, 'AsyncClient is connected');
+
+      client.end();
+    }, (error) => {
+      t.fail(error);
+    });
+
+
+  });
+
   test('Should be able to listen on event on client', t => {
     t.plan(1);