Repository: moscajs/aedes
Pull request #418 (2020-02-08 14:04:21)

Added test on broker.subscribe & broker.unsubscribe function
- added tests
- fixed typings


Patch:
diff --git a/aedes.d.ts b/aedes.d.ts
index 2e71b75..8f56b25 100644
--- a/aedes.d.ts
+++ b/aedes.d.ts
@@ -87,12 +87,12 @@ declare namespace aedes {
     ): void
     subscribe (
       topic: string,
-      deliverfunc: (packet: ISubscribePacket, callback: () => void) => void,
+      deliverfunc: (packet: IPublishPacket, callback: () => void) => void,
       callback: () => void
     ): void
     unsubscribe (
       topic: string,
-      deliverfunc: (packet: IUnsubscribePacket, callback: () => void) => void,
+      deliverfunc: (packet: IPublishPacket, callback: () => void) => void,
       callback: () => void
     ): void
     close (callback?: () => void): void
diff --git a/lib/handlers/connect.js b/lib/handlers/connect.js
index d8586c4..7580c9c 100644
--- a/lib/handlers/connect.js
+++ b/lib/handlers/connect.js
@@ -121,21 +121,23 @@ function authenticate (arg, done) {
 
     if (err) {
       const errCode = err.returnCode
-      if (errCode && (errCode >= 1 && errCode <= 3)) {
+      if (errCode && (errCode >= 2 && errCode <= 5)) {
         arg.returnCode = errCode
       } else {
-        // If errorCode is 4 or not a number
-        arg.returnCode = 4
+        arg.returnCode = 5
+      }
+      if (!err.message) {
+        err.message = errorMessages[arg.returnCode]
       }
     } else {
       arg.returnCode = 5
+      err = new Error(errorMessages[arg.returnCode])
     }
-    var error = new Error(errorMessages[arg.returnCode])
-    error.errorCode = arg.returnCode
+    err.errorCode = arg.returnCode
     arg.client = client
     doConnack(arg,
       // [MQTT-3.2.2-5]
-      client.close.bind(client, done.bind(this, error)))
+      client.close.bind(client, done.bind(this, err)))
   }
 }
 


Test patch:
diff --git a/test/auth.js b/test/auth.js
index fc512e2..72c1cf2 100644
--- a/test/auth.js
+++ b/test/auth.js
@@ -3,13 +3,13 @@
 const { test } = require('tap')
 const eos = require('end-of-stream')
 const Client = require('../lib/client')
-const { setup, connect, subscribe, subscribeMultiple } = require('./helper')
+const { setup, connect, noError, subscribe, subscribeMultiple } = require('./helper')
 const aedes = require('../')
 
 test('authenticate successfully a client with username and password', function (t) {
   t.plan(4)
 
-  const s = setup()
+  const s = noError(setup())
   t.tearDown(s.broker.close.bind(s.broker))
 
   s.broker.authenticate = function (client, username, password, cb) {
@@ -46,7 +46,7 @@ test('authenticate successfully a client with username and password', function (
 })
 
 test('authenticate unsuccessfully a client with username and password', function (t) {
-  t.plan(5)
+  t.plan(6)
 
   const s = setup()
   t.tearDown(s.broker.close.bind(s.broker))
@@ -58,6 +58,10 @@ test('authenticate unsuccessfully a client with username and password', function
     cb(null, false)
   }
 
+  s.broker.on('clientError', function (client, err) {
+    t.equal(err.errorCode, 5)
+  })
+
   s.broker.on('clientReady', function (client) {
     t.fail('client should not ready')
   })
@@ -73,7 +77,7 @@ test('authenticate unsuccessfully a client with username and password', function
       topic: null,
       payload: null,
       sessionPresent: false
-    }, 'unsuccessful connack')
+    }, 'unsuccessful connack, unauthorized')
   })
 
   eos(s.outStream, function () {
@@ -93,7 +97,7 @@ test('authenticate unsuccessfully a client with username and password', function
 })
 
 test('authenticate errors', function (t) {
-  t.plan(5)
+  t.plan(7)
 
   const s = setup()
   t.tearDown(s.broker.close.bind(s.broker))
@@ -105,6 +109,11 @@ test('authenticate errors', function (t) {
     cb(new Error('this should happen!'))
   }
 
+  s.broker.on('clientError', function (client, err) {
+    t.equal(err.message, 'this should happen!')
+    t.equal(err.errorCode, 5)
+  })
+
   s.broker.on('clientReady', function (client) {
     t.fail('client should not ready')
   })
@@ -112,7 +121,7 @@ test('authenticate errors', function (t) {
   s.outStream.on('data', function (packet) {
     t.deepEqual(packet, {
       cmd: 'connack',
-      returnCode: 4,
+      returnCode: 5,
       length: 2,
       qos: 0,
       retain: false,
@@ -120,7 +129,7 @@ test('authenticate errors', function (t) {
       topic: null,
       payload: null,
       sessionPresent: false
-    }, 'unsuccessful connack')
+    }, 'unsuccessful connack, unauthorized')
   })
 
   eos(s.outStream, function () {
@@ -140,7 +149,7 @@ test('authenticate errors', function (t) {
 })
 
 test('authentication error when return code 1 (unacceptable protocol version) is passed', function (t) {
-  t.plan(5)
+  t.plan(7)
 
   const s = setup()
   t.tearDown(s.broker.close.bind(s.broker))
@@ -154,6 +163,11 @@ test('authentication error when return code 1 (unacceptable protocol version) is
     cb(error, null)
   }
 
+  s.broker.on('clientError', function (client, err) {
+    t.equal(err.message, 'Auth error')
+    t.equal(err.errorCode, 5)
+  })
+
   s.broker.on('clientReady', function (client) {
     t.fail('client should not ready')
   })
@@ -161,7 +175,7 @@ test('authentication error when return code 1 (unacceptable protocol version) is
   s.outStream.on('data', function (packet) {
     t.deepEqual(packet, {
       cmd: 'connack',
-      returnCode: 1,
+      returnCode: 5,
       length: 2,
       qos: 0,
       retain: false,
@@ -169,7 +183,7 @@ test('authentication error when return code 1 (unacceptable protocol version) is
       topic: null,
       payload: null,
       sessionPresent: false
-    }, 'unsuccessful connack,unacceptable protocol version')
+    }, 'unsuccessful connack, unauthorized')
   })
 
   eos(s.outStream, function () {
@@ -189,7 +203,7 @@ test('authentication error when return code 1 (unacceptable protocol version) is
 })
 
 test('authentication error when return code 2 (identifier rejected) is passed', function (t) {
-  t.plan(5)
+  t.plan(7)
 
   const s = setup()
   t.tearDown(s.broker.close.bind(s.broker))
@@ -203,6 +217,11 @@ test('authentication error when return code 2 (identifier rejected) is passed',
     cb(error, null)
   }
 
+  s.broker.on('clientError', function (client, err) {
+    t.equal(err.message, 'Auth error')
+    t.equal(err.errorCode, 2)
+  })
+
   s.broker.on('clientReady', function (client) {
     t.fail('client should not ready')
   })
@@ -238,7 +257,7 @@ test('authentication error when return code 2 (identifier rejected) is passed',
 })
 
 test('authentication error when return code 3 (Server unavailable) is passed', function (t) {
-  t.plan(5)
+  t.plan(7)
 
   const s = setup()
   t.tearDown(s.broker.close.bind(s.broker))
@@ -252,6 +271,11 @@ test('authentication error when return code 3 (Server unavailable) is passed', f
     cb(error, null)
   }
 
+  s.broker.on('clientError', function (client, err) {
+    t.equal(err.message, 'Auth error')
+    t.equal(err.errorCode, 3)
+  })
+
   s.broker.on('clientReady', function (client) {
     t.fail('client should not ready')
   })
@@ -286,8 +310,62 @@ test('authentication error when return code 3 (Server unavailable) is passed', f
   })
 })
 
+test('authentication error when return code 4 (bad user or password) is passed', function (t) {
+  t.plan(7)
+
+  const s = setup()
+  t.tearDown(s.broker.close.bind(s.broker))
+
+  s.broker.authenticate = function (client, username, password, cb) {
+    t.ok(client instanceof Client, 'client is there')
+    t.equal(username, 'my username', 'username is there')
+    t.deepEqual(password, Buffer.from('my pass'), 'password is there')
+    var error = new Error('Auth error')
+    error.returnCode = 4
+    cb(error, null)
+  }
+
+  s.broker.on('clientError', function (client, err) {
+    t.equal(err.message, 'Auth error')
+    t.equal(err.errorCode, 4)
+  })
+
+  s.broker.on('clientReady', function (client) {
+    t.fail('client should not ready')
+  })
+
+  s.outStream.on('data', function (packet) {
+    t.deepEqual(packet, {
+      cmd: 'connack',
+      returnCode: 4,
+      length: 2,
+      qos: 0,
+      retain: false,
+      dup: false,
+      topic: null,
+      payload: null,
+      sessionPresent: false
+    }, 'unsuccessful connack, bad username or password')
+  })
+
+  eos(s.outStream, function () {
+    t.equal(s.broker.connectedClients, 0, 'no connected clients')
+  })
+
+  s.inStream.write({
+    cmd: 'connect',
+    protocolId: 'MQTT',
+    protocolVersion: 4,
+    clean: true,
+    clientId: 'my-client',
+    username: 'my username',
+    password: 'my pass',
+    keepalive: 0
+  })
+})
+
 test('authentication error when non numeric return code is passed', function (t) {
-  t.plan(5)
+  t.plan(7)
 
   const s = setup()
   t.tearDown(s.broker.close.bind(s.broker))
@@ -301,6 +379,11 @@ test('authentication error when non numeric return code is passed', function (t)
     cb(error, null)
   }
 
+  s.broker.on('clientError', function (client, err) {
+    t.equal(err.message, 'Non numeric error codes')
+    t.equal(err.errorCode, 5)
+  })
+
   s.broker.on('clientReady', function (client) {
     t.fail('client should not ready')
   })
@@ -308,7 +391,7 @@ test('authentication error when non numeric return code is passed', function (t)
   s.outStream.on('data', function (packet) {
     t.deepEqual(packet, {
       cmd: 'connack',
-      returnCode: 4,
+      returnCode: 5,
       length: 2,
       qos: 0,
       retain: false,
@@ -316,7 +399,7 @@ test('authentication error when non numeric return code is passed', function (t)
       topic: null,
       payload: null,
       sessionPresent: false
-    }, 'unsuccessful connack, bad user name or password')
+    }, 'unsuccessful connack, unauthorized')
   })
 
   eos(s.outStream, function () {
@@ -816,7 +899,7 @@ test('failed authentication does not disconnect other client with same clientId'
 })
 
 test('unauthorized connection should not unregister the correct one with same clientId', function (t) {
-  t.plan(2)
+  t.plan(4)
 
   const broker = aedes({
     authenticate: function (client, username, password, callback) {
@@ -832,6 +915,8 @@ test('unauthorized connection should not unregister the correct one with same cl
   t.tearDown(broker.close.bind(broker))
 
   broker.on('clientError', function (client, err) {
+    t.equal(err.message, 'bad user name or password')
+    t.equal(err.errorCode, 4)
     t.equal(broker.connectedClients, 1, 'my-client still connected')
   })
 
@@ -940,37 +1025,39 @@ test('change a topic name inside authorizeForward method in QoS 1 mode', functio
   })
 })
 
-test('prevent publish in QoS1 mode', function (t) {
-  t.plan(2)
+;[true, false].forEach(function (cleanSession) {
+  test(`unauthorized forward publish in QoS 1 mode [clean=${cleanSession}]`, function (t) {
+    t.plan(2)
 
-  const broker = aedes({
-    authorizeForward: function (client, packet) {
-      return null
-    }
-  })
-  t.tearDown(broker.close.bind(broker))
-
-  broker.on('client', function (client) {
-    client.subscribe({
-      topic: 'hello',
-      qos: 1
-    }, function (err) {
-      t.error(err, 'no error')
+    const broker = aedes({
+      authorizeForward: function (client, packet) {
+        return null
+      }
+    })
+    t.tearDown(broker.close.bind(broker))
 
-      broker.publish({
+    broker.on('client', function (client) {
+      client.subscribe({
         topic: 'hello',
-        payload: Buffer.from('world'),
         qos: 1
       }, function (err) {
         t.error(err, 'no error')
+
+        broker.publish({
+          topic: 'hello',
+          payload: Buffer.from('world'),
+          qos: 1
+        }, function (err) {
+          t.error(err, 'no error')
+        })
       })
     })
-  })
 
-  const s = connect(setup(broker))
+    const s = connect(setup(broker), { clean: cleanSession })
 
-  s.outStream.once('data', function (packet) {
-    t.fail('Should have not recieved this packet')
+    s.outStream.once('data', function (packet) {
+      t.fail('Should have not recieved this packet')
+    })
   })
 })
 
diff --git a/test/client-pub-sub.js b/test/client-pub-sub.js
index cbad57b..9e1d725 100644
--- a/test/client-pub-sub.js
+++ b/test/client-pub-sub.js
@@ -1,7 +1,7 @@
 'use strict'
 
 const { test } = require('tap')
-const { setup, connect } = require('./helper')
+const { setup, connect, subscribe, noError } = require('./helper')
 const aedes = require('../')
 
 test('publish direct to a single client QoS 0', function (t) {
@@ -604,3 +604,123 @@ test('should not receive a message on negated subscription', function (t) {
     t.fail('Packet should not be received')
   })
 })
+
+test('programmatically add custom subscribe', function (t) {
+  t.plan(6)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  const s = connect(setup(broker))
+  const expected = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: Buffer.from('world'),
+    qos: 0,
+    retain: false,
+    length: 12,
+    dup: false
+  }
+  var deliverP = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: Buffer.from('world'),
+    qos: 0,
+    retain: false
+  }
+  subscribe(t, s, 'hello', 0, function () {
+    broker.subscribe('hello', deliver, function () {
+      t.pass('subscribed')
+    })
+    s.outStream.on('data', function (packet) {
+      t.deepEqual(packet, expected, 'packet matches')
+    })
+    s.inStream.write({
+      cmd: 'publish',
+      topic: 'hello',
+      payload: 'world',
+      qos: 0,
+      messageId: 42
+    })
+  })
+  function deliver (packet, cb) {
+    deliverP.brokerId = s.broker.id
+    deliverP.brokerCounter = s.broker.counter
+    t.deepEqual(packet, deliverP, 'packet matches')
+    cb()
+  }
+})
+
+test('custom function in broker.subscribe', function (t) {
+  t.plan(4)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  const s = setup(broker)
+  var expected = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: Buffer.from('world'),
+    qos: 1,
+    retain: false,
+    messageId: undefined
+  }
+  connect(s, {}, function () {
+    broker.subscribe('hello', deliver, function () {
+      t.pass('subscribed')
+    })
+    s.inStream.write({
+      cmd: 'publish',
+      topic: 'hello',
+      payload: 'world',
+      qos: 1,
+      messageId: 42
+    })
+  })
+  broker.on('publish', function (packet, client) {
+    if (client) {
+      t.equal(packet.topic, 'hello')
+      t.equal(packet.messageId, 42)
+    }
+  })
+  function deliver (packet, cb) {
+    expected.brokerId = s.broker.id
+    expected.brokerCounter = s.broker.counter
+    t.deepEqual(packet, expected, 'packet matches')
+    cb()
+  }
+})
+
+test('custom function in broker.unsubscribe', function (t) {
+  t.plan(3)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  const s = noError(setup(broker))
+  connect(s, {}, function () {
+    broker.subscribe('hello', deliver, function () {
+      t.pass('subscribed')
+      broker.unsubscribe('hello', deliver, function () {
+        t.pass('unsubscribe')
+        s.inStream.write({
+          cmd: 'publish',
+          topic: 'hello',
+          payload: 'word',
+          qos: 1,
+          messageId: 42
+        })
+      })
+    })
+  })
+  broker.on('publish', function (packet, client) {
+    if (client) {
+      t.pass('publish')
+    }
+  })
+  function deliver (packet, cb) {
+    t.fail('shoudl not be called')
+    cb()
+  }
+})
diff --git a/test/qos2.js b/test/qos2.js
index 789708a..b9f06f0 100644
--- a/test/qos2.js
+++ b/test/qos2.js
@@ -1,6 +1,7 @@
 'use strict'
 
 const { test } = require('tap')
+const concat = require('concat-stream')
 const { setup, connect, subscribe } = require('./helper')
 const aedes = require('../')
 
@@ -189,6 +190,108 @@ test('call published method with client with QoS 2', function (t) {
   })
 })
 
+;[true, false].forEach(function (cleanSession) {
+  test(`authorized forward publish packets in QoS 2 [clean=${cleanSession}]`, function (t) {
+    t.plan(9)
+
+    const broker = aedes()
+    t.tearDown(broker.close.bind(broker))
+
+    const opts = { clean: cleanSession }
+    const publisher = connect(setup(broker))
+    const subscriber = connect(setup(broker), { ...opts, clientId: 'abcde' })
+    const forwarded = {
+      cmd: 'publish',
+      topic: 'hello',
+      payload: Buffer.from('world'),
+      qos: 2,
+      retain: false,
+      messageId: undefined
+    }
+    const expected = {
+      cmd: 'publish',
+      topic: 'hello',
+      payload: Buffer.from('world'),
+      qos: 2,
+      retain: false,
+      length: 14,
+      dup: false
+    }
+    broker.authorizeForward = function (client, packet) {
+      forwarded.brokerId = broker.id
+      forwarded.brokerCounter = broker.counter
+      t.deepEqual(packet, forwarded, 'forwarded packet must match')
+      return packet
+    }
+
+    subscribe(t, subscriber, 'hello', 2, function () {
+      subscriber.outStream.once('data', function (packet) {
+        t.notEqual(packet.messageId, 42)
+        delete packet.messageId
+        t.deepEqual(packet, expected, 'packet must match')
+      })
+
+      publish(t, publisher, {
+        cmd: 'publish',
+        topic: 'hello',
+        payload: Buffer.from('world'),
+        qos: 2,
+        retain: false,
+        messageId: 42,
+        dup: false
+      }, function () {
+        const stream = broker.persistence.outgoingStream({ id: 'abcde' })
+        stream.pipe(concat(function (list) {
+          if (cleanSession) {
+            t.equal(list.length, 0, 'should have empty item in queue')
+          } else {
+            t.equal(list.length, 1, 'should have one item in queue')
+          }
+        }))
+      })
+    })
+  })
+})
+
+;[true, false].forEach(function (cleanSession) {
+  test(`unauthorized forward publish packets in QoS 2 [clean=${cleanSession}]`, function (t) {
+    t.plan(6)
+
+    const broker = aedes()
+    t.tearDown(broker.close.bind(broker))
+
+    const opts = { clean: cleanSession }
+
+    const publisher = connect(setup(broker))
+    const subscriber = connect(setup(broker), { ...opts, clientId: 'abcde' })
+
+    broker.authorizeForward = function (client, packet) {
+
+    }
+
+    subscribe(t, subscriber, 'hello', 2, function () {
+      subscriber.outStream.once('data', function (packet) {
+        t.fail('should not receive any packets')
+      })
+
+      publish(t, publisher, {
+        cmd: 'publish',
+        topic: 'hello',
+        payload: Buffer.from('world'),
+        qos: 2,
+        retain: false,
+        messageId: 42,
+        dup: false
+      }, function () {
+        const stream = broker.persistence.outgoingStream({ id: 'abcde' })
+        stream.pipe(concat(function (list) {
+          t.equal(list.length, 0, 'should empty in queue')
+        }))
+      })
+    })
+  })
+})
+
 test('subscribe QoS 0, but publish QoS 2', function (t) {
   t.plan(6)
 
diff --git a/test/types/index.ts b/test/types/index.ts
index badacb7..428a207 100644
--- a/test/types/index.ts
+++ b/test/types/index.ts
@@ -2,7 +2,7 @@
 /* eslint no-undef: 0 */
 
 import { Server, Client, AuthenticateError } from '../../aedes'
-import { IPublishPacket, ISubscribePacket, ISubscription, IUnsubscribePacket } from 'mqtt-packet'
+import { IPublishPacket, ISubscription } from 'mqtt-packet'
 import { createServer } from 'net'
 
 const broker = Server({
@@ -122,17 +122,17 @@ broker.on('unsubscribe', (subscriptions, client) => {
   console.log(`client: ${client.id} subsribe`)
 })
 
-broker.subscribe('aaaa', (packet: ISubscribePacket, cb) => {
+broker.subscribe('aaaa', (packet: IPublishPacket, cb) => {
   console.log('cmd')
-  console.log(packet.subscriptions)
+  console.log(packet.cmd)
   cb()
 }, () => {
   console.log('done subscribing')
 })
 
-broker.unsubscribe('aaaa', (packet: IUnsubscribePacket, cb) => {
+broker.unsubscribe('aaaa', (packet: IPublishPacket, cb) => {
   console.log('cmd')
-  console.log(packet.unsubscriptions)
+  console.log(packet.cmd)
   cb()
 }, () => {
   console.log('done unsubscribing')