Repository: moscajs/aedes
Pull request #432 (2020-02-18 06:38:48)

Update README.md
- Rearranged sections
- We have aleady got a Sponsorship button on Github, it's not neccessary to emphasis it in README.md
- Added a Code Quality badge
- Added a node minimum version suppport badge for latest release

robertsLando: @gnought I would prefer to keep the backers button on top of readme, I have other sponsored projects and I noticed that the github sponsor button isn't used from most users and I received lot more sponsors when placing the button even on top of readme (seems stupid but maybe it's because not everyone know what that button is for)

Patch:
diff --git a/README.md b/README.md
index 6f6798a..a27c5f3 100644
--- a/README.md
+++ b/README.md
@@ -2,15 +2,17 @@
 # Aedes
 
 ![ci](https://github.com/moscajs/aedes/workflows/ci/badge.svg)
-[![Known Vulnerabilities](https://snyk.io/test/github/moscajs/aedes/badge.svg)](https://snyk.io/test/github/moscajs/aedes)
-[![Coverage Status](https://coveralls.io/repos/moscajs/aedes/badge.svg?branch=master&service=github)](https://coveralls.io/github/moscajs/aedes?branch=master)
 [![js-standard-style](https://img.shields.io/badge/code%20style-standard-brightgreen.svg?style=flat)](http://standardjs.com/)\
+![LGTM Grade](https://img.shields.io/lgtm/grade/javascript/github/moscajs/aedes)
+[![Coverage Status](https://coveralls.io/repos/moscajs/aedes/badge.svg?branch=master&service=github)](https://coveralls.io/github/moscajs/aedes?branch=master)
+[![Known Vulnerabilities](https://snyk.io/test/github/moscajs/aedes/badge.svg)](https://snyk.io/test/github/moscajs/aedes)\
 [![Dependencies Status](https://david-dm.org/moscajs/aedes/status.svg)](https://david-dm.org/moscajs/aedes)
 [![devDependencies Status](https://david-dm.org/moscajs/aedes/dev-status.svg)](https://david-dm.org/moscajs/aedes?type=dev)\
+![node](https://img.shields.io/node/v/aedes)
 [![NPM version](https://img.shields.io/npm/v/aedes.svg?style=flat)](https://www.npmjs.com/aedes)
 [![NPM downloads](https://img.shields.io/npm/dm/aedes.svg?style=flat)](https://www.npmjs.com/aedes)
 
-[![Become a Backer](https://opencollective.com/aedes/tiers/backer.svg?avatarHeight=80&width=600)](https://opencollective.com/aedes)
+[![opencollective](https://opencollective.com/aedes/donate/button.png)](https://opencollective.com/aedes/donate)
 
 Barebone MQTT server that can run on any stream servers
 
@@ -23,15 +25,14 @@ Barebone MQTT server that can run on any stream servers
   - [Middleware Plugins](#middleware-plugins)
     - [Persistence](#persistence)
     - [MQEmitter](#mqemitter)
-  - [Collaborators](#collaborators)
   - [Acknowledgements](#acknowledgements)
-  - [Contribution](#contribution)
-  - [Contributors](#contributors)
-    - [Backers](#backers)
-    - [Sponsors](#sponsors)
   - [Mosca vs Aedes](#mosca-vs-aedes)
     - [Benchmark: Aedes](#benchmark-aedes)
     - [Benchmark: Mosca](#benchmark-mosca)
+  - [Collaborators](#collaborators)
+  - [Contribution](#contribution)
+    - [Backers](#backers)
+    - [Sponsors](#sponsors)
   - [License](#license)
 
 ## Install
@@ -98,47 +99,12 @@ npm install aedes
   a Merkle DAG
 - [mqemitter-aerospike]: Aerospike mqemitter
 
-## Collaborators
-
-- [__Gavin D'mello__](https://github.com/GavinDmello)
-- [__Behrad Zari__](https://github.com/behrad)
-- [__Gnought__](https://github.com/gnought)
-- [__Daniel Lando__](https://github.com/robertsLando)
-
 ## Acknowledgements
 
 This library is born after a lot of discussion with all
 [Mosca](http://www.npmjs.com/mosca) users and how that was deployed in
 production. This addresses your concerns about performance and stability.
 
-## Contribution
-
-[![Help wanted](https://img.shields.io/github/labels/moscajs/aedes/help%20wanted)](https://github.com/moscajs/aedes/labels/help%20wanted)
-[![Contributors](https://img.shields.io/github/contributors/moscajs/aedes)](https://github.com/moscajs/aedes/graphs/contributors)
-
-Want to contribute? Check our list of
-[features/bugs](https://github.com/moscajs/aedes/projects/1)
-
-## Contributors
-
-This project exists thanks to all the people who contribute ❤️
-
-[![Backers](https://opencollective.com/aedes/contributors.svg?width=890&button=false)](https://opencollective.com/aedes#backers)
-
-### Backers
-
-Thank you to all our backers! 🙏 [[Become a backer](https://opencollective.com/aedes#backer)]
-
-[![Backers](https://opencollective.com/aedes/backers.svg?width=890)](https://opencollective.com/aedes#backers)
-
-### Sponsors
-
-Support this project by becoming a sponsor. Your logo will show up here with a link to your website. [[Become a sponsor](https://opencollective.com/aedes#sponsor)]
-
-[![Sponsor 1](https://opencollective.com/aedes/sponsor/0/avatar.svg)](https://opencollective.com/aedes/sponsor/0/website)
-
-[![Sponsor 2](https://opencollective.com/aedes/sponsor/1/avatar.svg)](https://opencollective.com/aedes/sponsor/1/website)
-
 ## Mosca vs Aedes
 
 Example benchmark test with 1000 clients sending 5000 QoS 1 messsages. Used
@@ -178,6 +144,33 @@ Average Bandwidth (msg/sec): 18.926
 Total Bandwidth (msg/sec):   18925.942
 ```
 
+## Collaborators
+
+- [__Gavin D'mello__](https://github.com/GavinDmello)
+- [__Behrad Zari__](https://github.com/behrad)
+- [__Gnought__](https://github.com/gnought)
+- [__Daniel Lando__](https://github.com/robertsLando)
+
+## Contribution
+
+[![Help wanted](https://img.shields.io/github/labels/moscajs/aedes/help%20wanted)](https://github.com/moscajs/aedes/labels/help%20wanted)
+[![Contributors](https://img.shields.io/github/contributors/moscajs/aedes)](https://github.com/moscajs/aedes/graphs/contributors)
+
+Want to contribute? Check our list of
+[features/bugs](https://github.com/moscajs/aedes/projects/1)
+
+### Backers
+
+Thank you to all our backers! :raised_hands:
+
+[![Backers](https://opencollective.com/aedes/backers.svg?avatarHeight=64&width=890&button=false)](https://opencollective.com/aedes#backers)
+
+### Sponsors
+
+Become a sponsor to get your logo on our README on Github
+
+[![Sponsor](https://opencollective.com/aedes/sponsors.svg)](https://opencollective.com/aedes#sponsor)
+
 ## License
 
 Licensed under [MIT](./LICENSE).
diff --git a/aedes.d.ts b/aedes.d.ts
index a33f065..9c39601 100644
--- a/aedes.d.ts
+++ b/aedes.d.ts
@@ -69,6 +69,7 @@ declare namespace aedes {
     authorizeForward?: AuthorizeForwardHandler
     published?: PublishedHandler
     queueLimit?: number
+    maxClientsIdLength?: number
   }
   interface Client extends EventEmitter {
     id: string
diff --git a/aedes.js b/aedes.js
index f71e1b3..2c860c9 100644
--- a/aedes.js
+++ b/aedes.js
@@ -28,7 +28,8 @@ const defaultOptions = {
   published: defaultPublished,
   trustProxy: false,
   trustedProxies: [],
-  queueLimit: 42
+  queueLimit: 42,
+  maxClientsIdLength: 23
 }
 
 function Aedes (opts) {
@@ -44,7 +45,11 @@ function Aedes (opts) {
   this.counter = 0
   this.queueLimit = opts.queueLimit
   this.connectTimeout = opts.connectTimeout
-  this.mq = opts.mq || mqemitter(opts)
+  this.maxClientsIdLength = opts.maxClientsIdLength
+  this.mq = opts.mq || mqemitter({
+    concurrency: opts.concurrency,
+    matchEmptyLevels: true // [MQTT-4.7.1-3]
+  })
   this.handle = function handle (conn, req) {
     conn.setMaxListeners(opts.concurrency * 2)
     // create a new Client instance for a new connection
diff --git a/docs/Aedes.md b/docs/Aedes.md
index 7611285..315685f 100644
--- a/docs/Aedes.md
+++ b/docs/Aedes.md
@@ -23,7 +23,7 @@
   - [aedes.subscribe (topic, deliverfunc, callback)](#aedessubscribe-topic-deliverfunc-callback)
   - [aedes.unsubscribe (topic, deliverfunc, callback)](#aedesunsubscribe-topic-deliverfunc-callback)
   - [aedes.publish (packet, callback)](#aedespublish-packet-callback)
-  - [aedes.close (callback)](#aedesclose-callback)
+  - [aedes.close ([callback])](#aedesclose-callback)
   - [Handler: decodeProtocol (client, buffer)](#handler-decodeprotocol-client-buffer)
   - [Handler: preConnect (client, callback)](#handler-preconnect-client-callback)
   - [Handler: authenticate (client, username, password, callback)](#handler-authenticate-client-username-password-callback)
@@ -39,6 +39,7 @@
   - `concurrency` `<number>` maximum number of concurrent messages delivered by `mq`. __Default__: `100`
   - `persistence` [`<Persistence>`](../README.md#persistence) a middleware stores _QoS > 0, retained, will_ packets and subscriptions. __Default__: `aedes-persistence`
   - `queueLimit` `<number>` maximum number of queued messages before client session is established. If number of queued items exceeds, `connectionError` throws an error `Client queue limit reached`. __Default__: `42`
+  - `maxClientsIdLength` option to override MQTT 3.1.0 clients Id length limit. __Default__: `23`
   - `heartbeatInterval` `<number>` an interval in millisconds at which server beats its health signal in `$SYS/<aedes.id>/heartbeat` topic. __Default__: `60000`
   - `connectTimeout` `<number>` maximum waiting time in milliseconds waiting for a [`CONNECT`][CONNECT] packet. __Default__: `30000`
 - Returns `<Aedes>`
diff --git a/lib/client.js b/lib/client.js
index 6f6f468..c194abb 100644
--- a/lib/client.js
+++ b/lib/client.js
@@ -218,7 +218,7 @@ Client.prototype.subscribe = function (packet, done) {
       subscriptions: packet
     }
   }
-  handleSubscribe(this, packet, done)
+  handleSubscribe(this, packet, false, done)
 }
 
 Client.prototype.unsubscribe = function (packet, done) {
diff --git a/lib/handlers/connect.js b/lib/handlers/connect.js
index b0af2bf..093f26a 100644
--- a/lib/handlers/connect.js
+++ b/lib/handlers/connect.js
@@ -66,7 +66,7 @@ function init (client, packet, done) {
     returnCode = 1
   }
   // MQTT 3.1.0 allows <= 23 client id length
-  if (packet.protocolVersion === 3 && clientId.length > 23) {
+  if (packet.protocolVersion === 3 && clientId.length > client.broker.maxClientsIdLength) {
     returnCode = 2
   }
   if (returnCode > 0) {
@@ -180,7 +180,7 @@ function gotSubs (err, subs, client) {
 
 function restoreSubs (arg, done) {
   if (arg.subs) {
-    handleSubscribe(this.client, { subscriptions: arg.subs, restore: true }, done)
+    handleSubscribe(this.client, { subscriptions: arg.subs }, true, done)
     arg.sessionPresent = !!arg.subs // cast to boolean, [MQTT-3.2.2-2]
     return
   }
diff --git a/lib/handlers/index.js b/lib/handlers/index.js
index 8ec3d8b..1caabfe 100644
--- a/lib/handlers/index.js
+++ b/lib/handlers/index.js
@@ -31,7 +31,7 @@ function handle (client, packet, done) {
       handlePublish(client, packet, done)
       break
     case 'subscribe':
-      handleSubscribe(client, packet, done)
+      handleSubscribe(client, packet, false, done)
       break
     case 'unsubscribe':
       handleUnsubscribe(client, packet, done)
diff --git a/lib/handlers/subscribe.js b/lib/handlers/subscribe.js
index d952cbd..c51ad60 100644
--- a/lib/handlers/subscribe.js
+++ b/lib/handlers/subscribe.js
@@ -1,14 +1,19 @@
 'use strict'
 
-const write = require('../write')
 const fastfall = require('fastfall')
 const Packet = require('aedes-packet')
 const through = require('through2')
-const validateTopic = require('./validations').validateTopic
-const topicActions = fastfall([
+const validateTopic = require('./validations')
+const write = require('../write')
+
+const subscribeTopicActions = fastfall([
   authorize,
   storeSubscriptions,
-  subTopic
+  addSubs
+])
+const restoreTopicActions = fastfall([
+  authorize,
+  addSubs
 ])
 
 function SubAck (packet, granted) {
@@ -22,9 +27,10 @@ function Subscription (qos, func) {
   this.func = func
 }
 
-function SubscribeState (client, packet, finish, granted) {
+function SubscribeState (client, packet, restore, finish, granted) {
   this.client = client
   this.packet = packet
+  this.actions = restore ? restoreTopicActions : subscribeTopicActions
   this.finish = finish
   this.subState = []
 }
@@ -49,19 +55,19 @@ function _dedupe (subs) {
   return ret
 }
 
-function handleSubscribe (client, packet, done) {
+function handleSubscribe (client, packet, restore, done) {
   const subs = packet.subscriptions
   client.broker._parallel(
-    new SubscribeState(client, packet, done),
+    new SubscribeState(client, packet, restore, done),
     doSubscribe,
     subs.length === 1 ? subs : _dedupe(subs),
-    completeSubscribe)
+    restore ? done : completeSubscribe)
 }
 
 function doSubscribe (sub, done) {
   const s = new SubState(this.client, this.packet, sub.qos)
   this.subState.push(s)
-  topicActions.call(s, sub, done)
+  this.actions.call(s, sub, done)
 }
 
 function authorize (sub, done) {
@@ -86,15 +92,11 @@ function blockDollarSignTopics (func) {
 
 function storeSubscriptions (sub, done) {
   if (!sub || typeof sub !== 'object') {
+    this.granted = 128
     return done(null, null)
   }
 
   const packet = this.packet
-
-  if (packet.restore) {
-    return done(null, sub)
-  }
-
   const client = this.client
 
   if (client.clean) {
@@ -106,9 +108,8 @@ function storeSubscriptions (sub, done) {
   })
 }
 
-function subTopic (sub, done) {
+function addSubs (sub, done) {
   if (!sub || typeof sub !== 'object') {
-    this.granted = 128
     return done()
   }
 
@@ -151,52 +152,44 @@ function completeSubscribe (err) {
 
   const packet = this.packet
   const client = this.client
-  const broker = client.broker
-  const granted = this.subState.map(obj => obj.granted)
-  this.subState = []
-
-  const subs = packet.subscriptions
-
-  if (!packet.restore) {
-    broker.emit('subscribe', subs, client)
-  }
 
   if (packet.messageId) {
-    write(client, new SubAck(packet, granted), nop)
-  }
-
-  // negated subscription check
-  if (granted[0] === 128) {
-    return done()
+    // [MQTT-3.9.3-1]
+    write(client,
+      new SubAck(packet, this.subState.map(obj => obj.granted)),
+      done)
   } else {
     done()
   }
+  this.subState = []
+
+  const broker = client.broker
+  const subs = packet.subscriptions
+
+  broker.emit('subscribe', subs, client)
 
   // Conform to MQTT 3.1.1 section 3.1.2.4
   // Restored sessions should not contain any retained message.
   // Retained message should be only fetched from SUBSCRIBE.
-  if (!packet.restore) {
-    const persistence = broker.persistence
-    var topics = []
-    for (var i = 0; i < subs.length; i++) {
-      topics.push(subs[i].topic)
-    }
-    const stream = persistence.createRetainedStreamCombi(topics)
-    stream.pipe(through.obj(function sendRetained (packet, enc, cb) {
-      packet = new Packet({
-        cmd: packet.cmd,
-        qos: packet.qos,
-        topic: packet.topic,
-        payload: packet.payload,
-        retain: true
-      }, broker)
-      // this should not be deduped
-      packet.brokerId = null
-      client.deliverQoS(packet, cb)
-    }))
+
+  const persistence = broker.persistence
+  var topics = []
+  for (var i = 0; i < subs.length; i++) {
+    topics.push(subs[i].topic)
   }
+  const stream = persistence.createRetainedStreamCombi(topics)
+  stream.pipe(through.obj(function sendRetained (packet, enc, cb) {
+    packet = new Packet({
+      cmd: packet.cmd,
+      qos: packet.qos,
+      topic: packet.topic,
+      payload: packet.payload,
+      retain: true
+    }, broker)
+    // this should not be deduped
+    packet.brokerId = null
+    client.deliverQoS(packet, cb)
+  }))
 }
 
-function nop () {}
-
 module.exports = handleSubscribe
diff --git a/lib/handlers/unsubscribe.js b/lib/handlers/unsubscribe.js
index 7ee5a32..fec0edc 100644
--- a/lib/handlers/unsubscribe.js
+++ b/lib/handlers/unsubscribe.js
@@ -1,7 +1,7 @@
 'use strict'
 
 const write = require('../write')
-const validateTopic = require('./validations').validateTopic
+const validateTopic = require('./validations')
 
 function UnsubscribeState (client, packet, finish) {
   this.client = client
diff --git a/lib/handlers/validations.js b/lib/handlers/validations.js
index f9993aa..2fa82e6 100644
--- a/lib/handlers/validations.js
+++ b/lib/handlers/validations.js
@@ -9,13 +9,13 @@ function validateTopic (topic, message) {
   }
   for (var i = 0; i < topic.length; i++) {
     switch (topic.charCodeAt(i)) {
-      case 35:
+      case 35: // #
         var notAtTheEnd = i !== end
         if (notAtTheEnd || slashInPreEnd) {
           return new Error('# is only allowed in ' + message + ' in the last position')
         }
         break
-      case 43:
+      case 43: // +
         var pastChar = i < end - 1 && topic.charCodeAt(i + 1) !== 47
         var preChar = i > 1 && topic.charCodeAt(i - 1) !== 47
         if (pastChar || preChar) {
@@ -26,4 +26,4 @@ function validateTopic (topic, message) {
   }
 }
 
-module.exports.validateTopic = validateTopic
+module.exports = validateTopic
diff --git a/package.json b/package.json
index 296eef4..95220db 100644
--- a/package.json
+++ b/package.json
@@ -71,8 +71,8 @@
   "devDependencies": {
     "@sinonjs/fake-timers": "^6.0.0",
     "@types/node": "^12.12.27",
-    "@typescript-eslint/eslint-plugin": "^2.19.2",
-    "@typescript-eslint/parser": "^2.19.2",
+    "@typescript-eslint/eslint-plugin": "^2.20.0",
+    "@typescript-eslint/parser": "^2.20.0",
     "concat-stream": "^2.0.0",
     "duplexify": "^4.1.1",
     "license-checker": "^25.0.1",
@@ -87,7 +87,7 @@
     "websocket-stream": "^5.5.0"
   },
   "dependencies": {
-    "aedes-packet": "^2.3.0",
+    "aedes-packet": "^2.3.1",
     "aedes-persistence": "^7.2.1",
     "aedes-protocol-decoder": "^1.0.0",
     "bulk-write-stream": "^2.0.1",
@@ -95,11 +95,10 @@
     "fastfall": "^1.5.1",
     "fastparallel": "^2.3.0",
     "fastseries": "^1.7.2",
-    "from2": "^2.3.0",
-    "mqemitter": "^4.0.0",
-    "mqtt-packet": "^6.3.0",
+    "mqemitter": "^4.1.1",
+    "mqtt-packet": "^6.3.1",
     "proxy-protocol-js": "^4.0.3",
-    "readable-stream": "^3.5.0",
+    "readable-stream": "^3.6.0",
     "retimer": "^2.0.0",
     "reusify": "^1.0.4",
     "shortid": "^2.2.15",


Test patch:
diff --git a/test/connect.js b/test/connect.js
index 179f3e1..83b855f 100644
--- a/test/connect.js
+++ b/test/connect.js
@@ -304,6 +304,32 @@ test('reject clients with > 23 clientId length in MQTT 3.1.0', function (t) {
   })
 })
 
+test('connect clients with > 23 clientId length using aedes maxClientsIdLength option in MQTT 3.1.0', function (t) {
+  t.plan(3)
+
+  const broker = aedes({ maxClientsIdLength: 26 })
+  t.tearDown(broker.close.bind(broker))
+
+  const s = setup(broker)
+
+  s.inStream.write({
+    cmd: 'connect',
+    protocolId: 'MQTT',
+    protocolVersion: 3,
+    clean: true,
+    clientId: 'abcdefghijklmnopqrstuvwxyz',
+    keepalive: 0
+  })
+  s.outStream.on('data', function (packet) {
+    t.equal(packet.cmd, 'connack')
+    t.equal(packet.returnCode, 0)
+    t.equal(broker.connectedClients, 1)
+  })
+  broker.on('connectionError', function (client, err) {
+    t.error(err, 'no error')
+  })
+})
+
 test('connect with > 23 clientId length in MQTT 3.1.1', function (t) {
   t.plan(3)
 
diff --git a/test/qos1.js b/test/qos1.js
index 9e1047b..cdedbc8 100644
--- a/test/qos1.js
+++ b/test/qos1.js
@@ -249,6 +249,64 @@ test('restore QoS 1 subscriptions not clean', function (t) {
   })
 })
 
+test('restore multiple QoS 1 subscriptions not clean w/ authorizeSubscribe', function (t) {
+  t.plan(11)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  var subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' })
+  const expected = {
+    cmd: 'publish',
+    topic: 'foo',
+    payload: Buffer.from('bar'),
+    qos: 1,
+    dup: false,
+    length: 10,
+    retain: false
+  }
+  const publisher = connect(setup(broker))
+
+  subscribe(t, subscriber, 'hello', 1, function () {
+    subscribe(t, subscriber, 'foo', 1, function () {
+      subscriber.inStream.end()
+      broker.authorizeSubscribe = function (client, sub, done) {
+        done(null, sub.topic === 'hello' ? 123 : sub)
+      }
+      subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' }, function (connect) {
+        t.equal(connect.sessionPresent, true, 'session present is set to true')
+        publisher.inStream.write({
+          cmd: 'publish',
+          topic: 'hello',
+          payload: 'world',
+          qos: 1,
+          messageId: 42
+        })
+        publisher.inStream.write({
+          cmd: 'publish',
+          topic: 'foo',
+          payload: 'bar',
+          qos: 1,
+          messageId: 48
+        })
+      })
+      publisher.outStream.on('data', function (packet) {
+        t.equal(packet.cmd, 'puback')
+      })
+
+      subscriber.outStream.on('data', function (packet) {
+        subscriber.inStream.write({
+          cmd: 'puback',
+          messageId: packet.messageId
+        })
+        t.notEqual(packet.messageId, 48, 'messageId must differ')
+        delete packet.messageId
+        t.deepEqual(packet, expected, 'packet must match')
+      })
+    })
+  })
+})
+
 test('remove stored subscriptions if connected with clean=true', function (t) {
   t.plan(5)
 
diff --git a/test/topics.js b/test/topics.js
index 9b039f3..7862bda 100644
--- a/test/topics.js
+++ b/test/topics.js
@@ -4,6 +4,26 @@ const { test } = require('tap')
 const { setup, connect, subscribe } = require('./helper')
 const aedes = require('../')
 
+// [MQTT-4.7.1-3]
+test('Single-level wildcard should match empty level', function (t) {
+  t.plan(4)
+
+  var s = connect(setup())
+  t.tearDown(s.broker.close.bind(s.broker))
+
+  subscribe(t, s, 'a/+/b', 0, function () {
+    s.outStream.once('data', function (packet) {
+      t.pass('ok')
+    })
+
+    s.inStream.write({
+      cmd: 'publish',
+      topic: 'a//b',
+      payload: 'world'
+    })
+  })
+})
+
 // [MQTT-4.7.3-1]
 test('publish empty topic', function (t) {
   t.plan(4)
@@ -13,7 +33,6 @@ test('publish empty topic', function (t) {
   subscribe(t, s, '#', 0, function () {
     s.outStream.once('data', function (packet) {
       t.fail('no packet')
-      // t.end()
     })
 
     s.inStream.write({