Tests, fixes, docs

Author: RangerMauve

README.md

@@ -2,7 +2,7 @@
 
 Promise wrapper over MQTT.js
 
-**IMPORANT: Make sure you handle rejections from returned promises because otherwise you might not see them**
+**IMPORANT: Make sure you handle rejections from returned promises because they won't crash the process**
 
 ## API
 
@@ -12,3 +12,46 @@ The API is the same as [MQTT.js](https://github.com/mqttjs/MQTT.js#api), except
 - subscribe
 - unsubscribe
 - end
+
+
+## Example
+
+```javascript
+var MQTT = require("async-mqtt");
+
+var client = MQTT.connect("tcp://somehost.com:1883");
+
+// WHen passing async functions as event listeners, make sure to have a try catch block
+client.on("connect", doStuff);
+
+async function doStuff() {
+
+	console.log("Starting");
+	try {
+		await client.publish("wow/so/cool", "It works!");
+		// This line doesn't run until the server responds to the publish
+		await client.end();
+		// This line doesn't run until the client has disconnected without error
+		console.log("Done");
+	} catch (e){
+		// Do something about it!
+		console.log(e.stack);
+		process.exit();
+	}
+}
+```
+
+## Wrapping existing client
+
+```javascript
+var AsyncClient = require("async-mqtt").AsyncClient;
+
+var client = getRegularMQTTClientFromSomewhere();
+
+var asyncClient = new AsyncClient(client);
+
+asyncClient.publish("foo/bar", "baz").then(function(){
+	console.log("We async now");
+	return asyncClient.end();
+});
+```

index.js

@@ -2,9 +2,27 @@
 var mqtt = require("mqtt");
 var inArray = require("in-array");
 
-var RegularClientPrototype = mqtt.client;
+var RegularClientPrototype = mqtt.MqttClient.prototype;
 
-var ASYNC_METHODS = ["publish", "subscribe", "unsubscribe", "unsubscribe", "end"];
+var ASYNC_METHODS = ["publish",
+	"subscribe",
+	"unsubscribe",
+	"unsubscribe",
+	"end"
+];
+
+var SYNC_METHODS = [
+	"emit",
+	"addListener",
+	"on",
+	"once",
+	"removeListener",
+	"removeAllListeners",
+	"setMaxListeners",
+	"getMaxListeners",
+	"listeners",
+	"listenerCount"
+];
 
 module.exports = {
 	connect: connect,
@@ -32,11 +50,8 @@ AsyncClient.prototype = {
 	}
 };
 
-for (var name in RegularClientPrototype) {
-	if (inArray(ASYNC_METHODS, name))
-		defineAsync(name);
-	else definePassthrough(name);
-}
+ASYNC_METHODS.forEach(defineAsync);
+SYNC_METHODS.forEach(definePassthrough);
 
 function definePassthrough(name) {
 	AsyncClient.prototype[name] = function() {
@@ -56,7 +71,7 @@ function defineAsync(name) {
 
 		return new Promise(function(resolve, reject) {
 			args.push(makeCallback(resolve, reject));
-			client.apply(client, args);
+			client[name].apply(client, args);
 		});
 	};
 }

package.json

@@ -26,5 +26,9 @@
   "dependencies": {
     "in-array": "^0.1.2",
     "mqtt": "^2.3.1"
+  },
+  "devDependencies": {
+    "mqtt-connection": "^3.0.0",
+    "tape": "^4.6.3"
   }
 }

test.js

@@ -0,0 +1,153 @@
+"use strict";
+
+var Server = require("mqtt/test/server");
+
+var AsyncMQTT = require("./");
+var AsyncClient = AsyncMQTT.AsyncClient;
+
+var test = require("tape");
+
+var SERVER_PORT = 1883;
+var SERVER_URL = "mqtt://localhost:" + SERVER_PORT;
+
+var server = buildServer().listen(SERVER_PORT);
+server.unref();
+
+server.on("listening", runTests);
+
+function runTests() {
+	test("Connect should return an instance of AsyncClient", function (t) {
+		t.plan(1);
+		var client = AsyncMQTT.connect(SERVER_URL);
+
+		t.ok(client instanceof AsyncClient, "Connect returned an AsyncClient");
+
+		client.end();
+	});
+
+	test("Should be able to listen on event on client", function (t) {
+		t.plan(1);
+
+		var client = AsyncMQTT.connect(SERVER_URL);
+
+		client.once("connect", function () {
+			t.pass("Connected");
+			client.end();
+		});
+	});
+
+	test("Calling end() should resolve once disconnected", function (t) {
+		t.plan(2);
+
+		var client = AsyncMQTT.connect(SERVER_URL);
+
+		client.on("close", function () {
+			t.pass("Close event occured");
+		});
+
+		client.on("connect", function () {
+			// Wait for connect to emit before ending
+			client.end().then(function(){
+				t.pass("End resolved");
+			});
+		});
+	});
+
+	test("Calling subscribe should resolve once subscribed", function (t) {
+		t.plan(1);
+
+		var client = AsyncMQTT.connect(SERVER_URL);
+
+		client.subscribe("example", {
+			qos: 1
+		}).then(function(){
+			t.pass("Subscribed");
+			client.end();
+		})
+	});
+
+	test("Calling unsubscribe should resolve once completed", function(t){
+		t.plan(1);
+
+		var client = AsyncMQTT.connect(SERVER_URL);
+
+		client.subscribe("example", {
+			qos: 1
+		}).then(function(){
+			return client.unsubscribe("example");
+		}).then(function(){
+			t.pass("Unsunbscribed");
+			return client.end();
+		});
+	});
+
+	test("Calling publish should resolve once completed", function (t) {
+		t.plan(1);
+
+		var client = AsyncMQTT.connect(SERVER_URL);
+
+		client.publish("example", "test", {
+			qos: 1
+		}).then(function(){
+			t.pass("Published");
+			return client.end();
+		});
+	});
+}
+
+// Taken from MQTT.js tests
+function buildServer () {
+  return new Server(function (client) {
+    client.on('connect', function (packet) {
+      if (packet.clientId === 'invalid') {
+        client.connack({returnCode: 2})
+      } else {
+        client.connack({returnCode: 0})
+      }
+    })
+
+    client.on('publish', function (packet) {
+      setImmediate(function () {
+        switch (packet.qos) {
+          case 0:
+            break
+          case 1:
+            client.puback(packet)
+            break
+          case 2:
+            client.pubrec(packet)
+            break
+        }
+      })
+    })
+
+    client.on('pubrel', function (packet) {
+      client.pubcomp(packet)
+    })
+
+    client.on('pubrec', function (packet) {
+      client.pubrel(packet)
+    })
+
+    client.on('pubcomp', function () {
+      // Nothing to be done
+    })
+
+    client.on('subscribe', function (packet) {
+      client.suback({
+        messageId: packet.messageId,
+        granted: packet.subscriptions.map(function (e) {
+          return e.qos
+        })
+      })
+    })
+
+    client.on('unsubscribe', function (packet) {
+      client.unsuback(packet)
+    })
+
+    client.on('pingreq', function () {
+      client.pingresp()
+    })
+  })
+}