Merge 1b581cc into cb3c3a6

test/auth.js

@@ -567,7 +567,15 @@ test('negate multiple subscriptions', function (t) {
 })
 
 test('negate subscription with correct persistence', function (t) {
-  t.plan(7)
+  t.plan(6)
+
+  var expected = [{
+    topic: 'hello',
+    qos: 0
+  }, {
+    topic: 'world',
+    qos: 0
+  }]
 
   var broker = aedes()
   broker.authorizeSubscribe = function (client, sub, cb) {
@@ -582,8 +590,9 @@ test('negate subscription with correct persistence', function (t) {
   s.outStream.once('data', function (packet) {
     t.equal(packet.cmd, 'suback')
     t.deepEqual(packet.granted, [128, 0])
-    t.notEqual(broker.persistence._subscriptions.get('abcde'), undefined)
-    t.deepEqual(broker.persistence._subscriptions.get('abcde').size, 2)
+    broker.persistence.subscriptionsByClient(broker.clients['abcde'], function (_, subs, client) {
+      t.deepEqual(subs, expected)
+    })
     t.equal(packet.messageId, 24)
   })
 

test/basic.js

@@ -11,6 +11,8 @@ var noError = helper.noError
 var subscribe = helper.subscribe
 
 test('connect and connack (minimal)', function (t) {
+  t.plan(1)
+
   var s = setup()
 
   s.inStream.write({
@@ -86,6 +88,8 @@ test('second CONNECT Packet sent from a Client as a protocol violation and disco
 })
 
 test('publish QoS 0', function (t) {
+  t.plan(2)
+
   var s = connect(setup())
   var expected = {
     cmd: 'publish',
@@ -112,6 +116,8 @@ test('publish QoS 0', function (t) {
 })
 
 test('subscribe QoS 0', function (t) {
+  t.plan(4)
+
   var s = connect(setup())
   var expected = {
     cmd: 'publish',
@@ -139,6 +145,7 @@ test('subscribe QoS 0', function (t) {
 
 test('does not die badly on connection error', function (t) {
   t.plan(3)
+
   var s = connect(setup())
 
   s.inStream.write({
@@ -228,6 +235,8 @@ test('unsubscribe without subscribe', function (t) {
 })
 
 test('unsubscribe on disconnect for a clean=true client', function (t) {
+  t.plan(6)
+
   var opts = { clean: true }
   var s = noError(connect(setup(), opts), t)
 
@@ -238,10 +247,10 @@ test('unsubscribe on disconnect for a clean=true client', function (t) {
     })
     s.broker.once('unsubscribe', function () {
       t.pass('should emit unsubscribe')
-      t.end()
     })
     s.broker.once('closed', function () {
       t.ok(true)
+      t.end()
     })
     s.broker.publish({
       cmd: 'publish',
@@ -254,6 +263,8 @@ test('unsubscribe on disconnect for a clean=true client', function (t) {
 })
 
 test('unsubscribe on disconnect for a clean=false client', function (t) {
+  t.plan(5)
+
   var opts = { clean: false }
   var s = noError(connect(setup(), opts), t)
 
@@ -280,6 +291,8 @@ test('unsubscribe on disconnect for a clean=false client', function (t) {
 })
 
 test('disconnect', function (t) {
+  t.plan(0)
+
   var s = noError(connect(setup()), t)
 
   s.outStream.on('finish', function () {
@@ -353,6 +366,8 @@ test('broker closes gracefully', function (t) {
 })
 
 test('testing other event', function (t) {
+  t.plan(1)
+
   var broker = aedes()
   var client = setup(broker)
 
@@ -364,6 +379,8 @@ test('testing other event', function (t) {
 })
 
 test('connect without a clientId for MQTT 3.1.1', function (t) {
+  t.plan(1)
+
   var s = setup()
 
   s.inStream.write({
@@ -454,6 +471,8 @@ test('publish to $SYS/broker/new/clients', function (t) {
 })
 
 test('restore QoS 0 subscriptions not clean', function (t) {
+  t.plan(5)
+
   var broker = aedes()
   var publisher
   var subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' })
@@ -490,6 +509,8 @@ test('restore QoS 0 subscriptions not clean', function (t) {
 })
 
 test('do not restore QoS 0 subscriptions when clean', function (t) {
+  t.plan(6)
+
   var broker = aedes()
   var publisher
   var subscriber = connect(setup(broker), { clean: true, clientId: 'abcde' })
@@ -522,6 +543,8 @@ test('do not restore QoS 0 subscriptions when clean', function (t) {
 })
 
 test('double sub does not double deliver', function (t) {
+  t.plan(7)
+
   var s = connect(setup())
   var expected = {
     cmd: 'publish',
@@ -554,6 +577,8 @@ test('double sub does not double deliver', function (t) {
 })
 
 test('overlapping sub does not double deliver', function (t) {
+  t.plan(7)
+
   var s = connect(setup())
   var expected = {
     cmd: 'publish',
@@ -586,6 +611,8 @@ test('overlapping sub does not double deliver', function (t) {
 })
 
 test('publish empty topic', function (t) {
+  t.plan(4)
+
   var s = connect(setup())
 
   subscribe(t, s, '#', 0, function () {
@@ -608,6 +635,8 @@ test('publish empty topic', function (t) {
 })
 
 test('publish invalid topic with #', function (t) {
+  t.plan(3)
+
   var s = connect(setup())
 
   subscribe(t, s, '#', 0, function () {
@@ -629,6 +658,8 @@ test('publish invalid topic with #', function (t) {
 })
 
 test('publish invalid topic with +', function (t) {
+  t.plan(3)
+
   var s = connect(setup())
 
   subscribe(t, s, '#', 0, function () {
@@ -650,6 +681,8 @@ test('publish invalid topic with +', function (t) {
 
 ;['base/#/sub', 'base/#sub', 'base/sub#', 'base/xyz+/sub', 'base/+xyz/sub'].forEach(function (topic) {
   test('subscribe to invalid topic with "' + topic + '"', function (t) {
+    t.plan(0)
+
     var s = connect(setup())
 
     s.broker.on('clientError', function () {
@@ -667,6 +700,8 @@ test('publish invalid topic with +', function (t) {
   })
 
   test('unsubscribe to invalid topic with "' + topic + '"', function (t) {
+    t.plan(0)
+
     var s = connect(setup())
 
     s.broker.on('clientError', function () {

test/client-pub-sub.js

@@ -346,6 +346,7 @@ test('subscribe a client programmatically with full packet', function (t) {
 
 test('get message when client connects', function (t) {
   t.plan(2)
+
   var client1 = 'gav'
   var broker = aedes()
 
@@ -369,6 +370,7 @@ test('get message when client connects', function (t) {
 
 test('get message when client disconnects', function (t) {
   t.plan(2)
+
   var client1 = 'gav'
   var client2 = 'friend'
   var broker = aedes()
@@ -397,7 +399,7 @@ test('get message when client disconnects', function (t) {
 })
 
 test('should not receive a message on negated subscription', function (t) {
-  t.plan(3)
+  t.plan(2)
 
   var broker = aedes()
   broker.authorizeSubscribe = function (client, sub, callback) {
@@ -422,16 +424,8 @@ test('should not receive a message on negated subscription', function (t) {
   })
 
   var s = connect(setup(broker))
-  var receivedPacket = null
   s.outStream.once('data', function (packet) {
-    receivedPacket = packet
+    t.fail('Packet should not be received')
   })
-
-  setTimeout(function () {
-    if (receivedPacket != null) {
-      t.fail('Packet should not be received')
-    } else {
-      t.pass('Message not received')
-    }
-  }, 100)
+  broker.on('closed', t.end.bind(t))
 })

test/events.js

@@ -9,6 +9,7 @@ var subscribe = helper.subscribe
 
 test('publishes an hearbeat', function (t) {
   t.plan(3)
+
   var broker = aedes({
     heartbeatInterval: 10 // ms
   })
@@ -23,6 +24,7 @@ test('publishes an hearbeat', function (t) {
 
 test('does not forward $SYS topics to # subscription', function (t) {
   t.plan(4)
+
   var s = connect(setup())
 
   subscribe(t, s, '#', 0, function () {
@@ -42,6 +44,7 @@ test('does not forward $SYS topics to # subscription', function (t) {
 
 test('does not forward $SYS topics to +/# subscription', function (t) {
   t.plan(4)
+
   var s = connect(setup())
 
   subscribe(t, s, '+/#', 0, function () {
@@ -89,6 +92,7 @@ test('does not store $SYS topics to QoS 1 # subscription', function (t) {
 test('Emit event when receives a ping', function (t) {
   t.plan(6)
   t.timeoutAfter(2000)
+
   var broker = aedes()
 
   broker.on('ping', function (packet, client) {
@@ -98,8 +102,8 @@ test('Emit event when receives a ping', function (t) {
       t.equal(packet.payload, null)
       t.equal(packet.topic, null)
       t.equal(packet.length, 0)
-      broker.close()
       t.pass('ended')
+      broker.close()
     }
   })
 
@@ -112,6 +116,7 @@ test('Emit event when receives a ping', function (t) {
 
 test('Emit event when broker closed', function (t) {
   t.plan(1)
+
   var broker = aedes()
   broker.once('closed', function () {
     t.ok(true)

test/keep-alive.js

@@ -57,7 +57,7 @@ test('supports keep alive disconnections after a pingreq', function (t) {
 
 test('disconnect if a connect does not arrive in time', function (t) {
   t.plan(2)
-  t.timeoutAfter(200)
+  t.timeoutAfter(500)
 
   var s = setup(aedes({
     connectTimeout: 100

test/not-blocking.js

@@ -1,12 +1,14 @@
 'use strict'
 
 var mqtt = require('mqtt')
-var test = require('tape')
+var test = require('tape').test
 var aedes = require('../')
 var net = require('net')
 var port = 4883
 
 test('do not block after a subscription', function (t) {
+  t.plan(3)
+
   var instance = aedes()
   var server = net.createServer(instance.handle)
   var total = 10000
@@ -73,6 +75,8 @@ test('do not block after a subscription', function (t) {
 })
 
 test('do not block with overlapping subscription', function (t) {
+  t.plan(3)
+
   var instance = aedes({ concurrency: 15 })
   var server = net.createServer(instance.handle)
   var total = 10000