Repository: moscajs/aedes
Pull request #409 (2020-02-06 04:13:07)

Introduce connecting and closed metadata
- `connecting`: show we are in CONNECT phase. Please noted that `connackSent` does not mean reset `connecting` to false and set `connected` to true. We still need to procceed the last step, pushing (QoS>0 and !clean) / retain messages to clients
- `connected` should be flagged true when `clientReady`/`connected` is emitted
- `closed`: show client closed state

gnought: WIP
gnought: WIP on tests

Patch:
diff --git a/.github/FUNDING.yml b/.github/FUNDING.yml
new file mode 100644
index 0000000..f8c8c40
--- /dev/null
+++ b/.github/FUNDING.yml
@@ -0,0 +1,1 @@
+open_collective: aedes
diff --git a/README.md b/README.md
index daa9061..46ad46e 100644
--- a/README.md
+++ b/README.md
@@ -1,3 +1,4 @@
+<!-- markdownlint-disable MD024 -->
 # Aedes
 ![](https://github.com/moscajs/aedes/workflows/ci/badge.svg)
 [![Dependencies Status](https://david-dm.org/moscajs/aedes/status.svg)](https://david-dm.org/moscajs/aedes)
@@ -31,8 +32,8 @@ Barebone MQTT server that can run on any stream server.
 ## Install
 To install aedes, simply use npm:
 
-```
-npm install aedes --save
+```sh
+npm install aedes
 ```
 
 <a name="example"></a>
@@ -106,10 +107,6 @@ httpServer.listen(wsPort, function () {
   * <a href="#published"><code>instance.<b>published()</b></code></a>
   * <a href="#close"><code>instance.<b>close()</b></code></a>
   * <a href="#client"><code><b>Client</b></code></a>
-  * <a href="#clientid"><code>client.<b>id</b></code></a>
-  * <a href="#clientclean"><code>client.<b>clean</b></code></a>
-  * <a href="#clientconn"><code>client.<b>conn</b></code></a>
-  * <a href="#clientreq"><code>client.<b>req</b></code></a>
   * <a href="#clientpublish"><code>client.<b>publish()</b></code></a>
   * <a href="#clientsubscribe"><code>client.<b>subscribe()</b></code></a>
   * <a href="#clientunsubscribe"><code>client.<b>unsubscribe()</b></code></a>
@@ -121,7 +118,7 @@ httpServer.listen(wsPort, function () {
 
 Creates a new instance of Aedes.
 
-Options:
+#### Options
 
 * `mq`: an instance of [MQEmitter](http://npm.im/mqemitter), check [plugins](#plugins) for more mqemitters options. Used to share messages between multiple brokers instances (ex: clusters)
 * `persistence`: an instance of [AedesPersistence](http://npm.im/aedes-persistence), check [plugins](#plugins) for more persistence options. It's used to store *QoS > 1*, *retained*, *will* packets and subscriptions in memory or on disk (if not specified default persistence is in memory)
@@ -151,7 +148,11 @@ Options:
 * `published`: function called when a new packet is published, see
   [instance.published()](#published)
 
-Events:
+#### Properties
+
+* `closed`: read-only, shows the aedes closed state
+
+#### Events
 
 * `client`: when a new [Client](#client) successfully connects and register itself to server, [connackSent event will be come after], arguments:
   1. `client`
@@ -248,7 +249,8 @@ Both `topic` and `payload` can be `Buffer` objects instead of strings.
 ### instance.unsubscribe(topic, func(packet, cb), done)
 
 The reverse of [subscribe](#subscribe).
-------------------------------------------------------
+
+-------------------------------------------------------
 <a name="decodeProtocol"></a>
 ### instance.decodeProtocol(client, buffer)
 
@@ -260,12 +262,14 @@ instance.decodeProtocol = function(client, buffer) {
   return protocol
 }
 ```
+
 -------------------------------------------------------
 <a name="preConnect"></a>
 ### instance.preConnect(client, done(err, successful))
 
 It will be called when aedes instance receives a first valid CONNECT packet from client. client object state is in default and its connected state is false. Any values in CONNECT packet (like clientId, clean flag, keepalive) will pass to client object after this call. Override to supply custom preConnect logic.
 Some use cases:
+
 1. Rate Limit / Throttle by `client.conn.remoteAddress`
 2. Check `instance.connectedClient` to limit maximum connections
 3. IP blacklisting
@@ -275,11 +279,13 @@ instance.preConnect = function(client, callback) {
   callback(null, client.conn.remoteAddress === '::1') {
 }
 ```
+
 ```js
 instance.preConnect = function(client, callback) {
   callback(new Error('connection error'), client.conn.remoteAddress !== '::1') {
 }
 ```
+
 -------------------------------------------------------
 <a name="authenticate"></a>
 ### instance.authenticate(client, username, password, done(err, successful))
@@ -292,6 +298,7 @@ instance.authenticate = function (client, username, password, callback) {
   callback(null, username === 'matteo')
 }
 ```
+
 Other return codes can passed as follows :-
 
 ```js
@@ -301,12 +308,13 @@ instance.authenticate = function (client, username, password, callback) {
   callback(error, null)
 }
 ```
+
 The return code values and their responses which can be passed are given below:
 
-*  `1` - Unacceptable protocol version
-*  `2` - Identifier rejected
-*  `3` - Server unavailable
-*  `4` - Bad user name or password
+* `1` - Unacceptable protocol version
+* `2` - Identifier rejected
+* `3` - Server unavailable
+* `4` - Bad user name or password
 
 -------------------------------------------------------
 <a name="authorizePublish"></a>
@@ -409,40 +417,22 @@ Events:
 
 Classes for all connected clients.
 
-Events:
-
-* `error`, in case something bad happended
-
--------------------------------------------------------
-<a name="clientid"></a>
-### client#id
-
-The id of the client, as specified by the CONNECT packet, defaults to 'aedes_' + shortid()
-
--------------------------------------------------------
-<a name="clientclean"></a>
-### client#clean
-
-`true` if the client connected (CONNECT) with `clean: true`, `false`
-otherwise. Check the MQTT spec for what this means.
-
--------------------------------------------------------
-<a name="clientconn"></a>
-### client#conn
+#### Properties
 
-The client's connection stream object.
+* `connecting` read-only, it is true when Client is in CONNECT phase. Aedes emits `connackSent` event will not reset `connecting` to `false` until it received all its offline messagess to the Client
+* `connected`: read-only, it is true when `connected` event is emitted, and false when Client is closed
+* `closed`: read-only, shows Client closed state
+* `id`: Client id, specified by CONNECT packet, defaults to `'aedes_' + shortid()`. It is `null` in `preConnect` hooks unless it returns good
+* `clean`: Client clean flag, specified by CONNECT packet.
+* `conn`: Client connection stream object.
+  * In the case of `net.createServer`, `conn` passed to the `connectionlistener` function by node's [net.createServer](https://nodejs.org/api/net.html#net_net_createserver_options_connectionlistener) API
+  * In the case of [`websocket-stream`](https://www.npmjs.com/package/websocket-stream), it's the `stream` argument passed to the websocket `handle` function in [`websocket-stream` documentation](https://github.com/maxogden/websocket-stream/blob/e2a51644bb35132d7aa477ae1a27ff083fedbf08/readme.md#on-the-server)
+* `req`: only for [`websocket-stream`](https://www.npmjs.com/package/websocket-stream). It is a HTTP Websocket upgrade request object passed to websocket `handle` function in [`websocket-stream` documentation](https://github.com/maxogden/websocket-stream/blob/e2a51644bb35132d7aa477ae1a27ff083fedbf08/readme.md#on-the-server). It gives an option for accessing headers or cookies
 
-In the case of `net.createServer` brokers, it's the connection passed to the `connectionlistener` function by node's [net.createServer](https://nodejs.org/api/net.html#net_net_createserver_options_connectionlistener) API.
+#### Events
 
-In the case of [websocket-stream](https://www.npmjs.com/package/websocket-stream) brokers, it's the `stream` argument passed to the `handle` function described in [that library's documentation](https://github.com/maxogden/websocket-stream/blob/e2a51644bb35132d7aa477ae1a27ff083fedbf08/readme.md#on-the-server).
-
--------------------------------------------------------
-<a name="clientreq"></a>
-### client#req
-
-The HTTP Websocket upgrade request object passed to websocket broker's `handle` function by the [`websocket-stream` library](https://github.com/maxogden/websocket-stream/blob/e2a51644bb35132d7aa477ae1a27ff083fedbf08/readme.md#on-the-server).
-
-If your clients are connecting to aedes via websocket and you need access to headers or cookies, you can get them here. **NOTE:** this property is only present for websocket connections.
+* `connected`, this is the same as aedes `clientReady` but in client-wise
+* `error`, in case something bad happended
 
 -------------------------------------------------------
 <a name="clientpublish"></a>
@@ -496,27 +486,27 @@ Disconnects the client
 
 You can subscribe on the following `$SYS` topics to get client presence:
 
- - `$SYS/+/new/clients` - will inform about new clients connections
- - `$SYS/+/disconnect/clients` - will inform about client disconnections.
+* `$SYS/+/new/clients` - will inform about new clients connections
+* `$SYS/+/disconnect/clients` - will inform about client disconnections.
 The payload will contain the `clientId` of the connected/disconnected client
 
 ## Plugins
 
-- [aedes-persistence](https://github.com/moscajs/aedes-persistence): In-memory implementation of an Aedes persistence
-  - [aedes-persistence-mongodb](https://github.com/moscajs/aedes-persistence-mongodb): MongoDB persistence for Aedes
-  - [aedes-persistence-redis](https://github.com/moscajs/aedes-persistence-redis): Redis persistence for Aedes
-  - [aedes-persistence-level](https://github.com/moscajs/aedes-persistence-level): LevelDB persistence for Aedes
-  - [aedes-persistence-nedb](https://github.com/ovhemert/aedes-persistence-nedb#readme): NeDB persistence for Aedes
-- [mqemitter](https://github.com/mcollina/mqemitter): An Opinionated Message Queue with an emitter-style API
-  - [mqemitter-redis](https://github.com/mcollina/mqemitter-redis): Redis-powered mqemitter
-  - [mqemitter-mongodb](https://github.com/mcollina/mqemitter-mongodb): Mongodb based mqemitter
-  - [mqemitter-child-process](https://github.com/mcollina/mqemitter-child-process): Share the same mqemitter between a hierarchy of child processes
-  - [mqemitter-cs](https://github.com/mcollina/mqemitter-cs): Expose a MQEmitter via a simple client/server protocol
-  - [mqemitter-p2p](https://github.com/mcollina/mqemitter-p2p): A P2P implementation of MQEmitter, based on HyperEmitter and a Merkle DAG
-  - [mqemitter-aerospike](https://github.com/GavinDmello/mqemitter-aerospike): Aerospike mqemitter based on @mcollina 's mqemitter
-- [aedes-logging](https://github.com/moscajs/aedes-logging): Logging module for Aedes, based on Pino
-- [aedes-stats](https://github.com/moscajs/aedes-stats): Stats for Aedes
-- [aedes-protocol-decoder](https://github.com/moscajs/aedes-protocol-decoder): Protocol decoder for Aedes MQTT Broker
+* [aedes-persistence](https://github.com/moscajs/aedes-persistence): In-memory implementation of an Aedes persistence
+  * [aedes-persistence-mongodb](https://github.com/moscajs/aedes-persistence-mongodb): MongoDB persistence for Aedes
+  * [aedes-persistence-redis](https://github.com/moscajs/aedes-persistence-redis): Redis persistence for Aedes
+  * [aedes-persistence-level](https://github.com/moscajs/aedes-persistence-level): LevelDB persistence for Aedes
+  * [aedes-persistence-nedb](https://github.com/ovhemert/aedes-persistence-nedb#readme): NeDB persistence for Aedes
+* [mqemitter](https://github.com/mcollina/mqemitter): An Opinionated Message Queue with an emitter-style API
+  * [mqemitter-redis](https://github.com/mcollina/mqemitter-redis): Redis-powered mqemitter
+  * [mqemitter-mongodb](https://github.com/mcollina/mqemitter-mongodb): Mongodb based mqemitter
+  * [mqemitter-child-process](https://github.com/mcollina/mqemitter-child-process): Share the same mqemitter between a hierarchy of child processes
+  * [mqemitter-cs](https://github.com/mcollina/mqemitter-cs): Expose a MQEmitter via a simple client/server protocol
+  * [mqemitter-p2p](https://github.com/mcollina/mqemitter-p2p): A P2P implementation of MQEmitter, based on HyperEmitter and a Merkle DAG
+  * [mqemitter-aerospike](https://github.com/GavinDmello/mqemitter-aerospike): Aerospike mqemitter based on @mcollina 's mqemitter
+* [aedes-logging](https://github.com/moscajs/aedes-logging): Logging module for Aedes, based on Pino
+* [aedes-stats](https://github.com/moscajs/aedes-stats): Stats for Aedes
+* [aedes-protocol-decoder](https://github.com/moscajs/aedes-protocol-decoder): Protocol decoder for Aedes MQTT Broker
 
 ## Collaborators
 
@@ -543,7 +533,7 @@ Example benchmark test with 1000 clients sending 5000 QoS 1 messsages. Used [mqt
 
 ### Aedes
 
-```
+```sh
 ========= TOTAL (1000) =========
 Total Ratio:                 1.000 (5000000/5000000)
 Total Runtime (sec):         178.495
@@ -558,7 +548,7 @@ Total Bandwidth (msg/sec):   28114.678
 
 ### Mosca
 
-```
+```sh
 ========= TOTAL (1000) =========
 Total Ratio:                 1.000 (5000000/5000000)
 Total Runtime (sec):         264.934
diff --git a/lib/client.js b/lib/client.js
index 9f6667c..6f6f468 100644
--- a/lib/client.js
+++ b/lib/client.js
@@ -18,6 +18,8 @@ function Client (broker, conn, req) {
   const that = this
 
   // metadata
+  this.closed = false
+  this.connecting = false
   this.connected = false
   this.connackSent = false
   this.errored = false
@@ -232,17 +234,18 @@ Client.prototype.unsubscribe = function (packet, done) {
 }
 
 Client.prototype.close = function (done) {
-  const that = this
-
-  if (this._eos === nop) {
+  if (this.closed) {
     if (typeof done === 'function') {
       done()
     }
     return
   }
 
+  const that = this
   const conn = this.conn
 
+  this.closed = true
+
   this._parser.removeAllListeners('packet')
   conn.removeAllListeners('readable')
 
@@ -262,17 +265,13 @@ Client.prototype.close = function (done) {
   this._eos()
   this._eos = nop
 
-  if (this.connected) {
-    handleUnsubscribe(
-      this,
-      {
-        close: true,
-        unsubscriptions: Object.keys(this.subscriptions)
-      },
-      finish)
-  } else {
-    finish()
-  }
+  handleUnsubscribe(
+    this,
+    {
+      close: true,
+      unsubscriptions: Object.keys(this.subscriptions)
+    },
+    finish)
 
   function finish () {
     if (!that._disconnected && that.will) {
@@ -297,11 +296,12 @@ Client.prototype.close = function (done) {
     that.will = null // this function might be called twice
     that._will = null
 
+    that.connected = false
+    that.connecting = false
+
     conn.removeAllListeners('error')
     conn.on('error', nop)
 
-    that.connected = false
-
     if (that.broker.clients[that.id] && that._authorized) {
       that.broker.unregisterClient(that)
     }
diff --git a/lib/handlers/connect.js b/lib/handlers/connect.js
index 82a8e9e..d8586c4 100644
--- a/lib/handlers/connect.js
+++ b/lib/handlers/connect.js
@@ -42,7 +42,7 @@ const errorMessages = [
 function handleConnect (client, packet, done) {
   clearTimeout(client._connectTimer)
   client._connectTimer = null
-
+  client.connecting = true
   client.broker.preConnect(client, negate)
 
   function negate (err, successful) {
@@ -50,6 +50,7 @@ function handleConnect (client, packet, done) {
       setImmediate(init, client, packet, done)
       return
     }
+    client.connecting = false
     if (err) {
       client.broker.emit('connectionError', client, err)
     }
@@ -58,7 +59,6 @@ function handleConnect (client, packet, done) {
 }
 
 function init (client, packet, done) {
-  client.connected = true
   const clientId = packet.clientId
   var returnCode = 0
   // [MQTT-3.1.2-2]
@@ -88,7 +88,9 @@ function init (client, packet, done) {
     connectActions,
     { returnCode: 0, sessionPresent: false }, // [MQTT-3.1.4-4], [MQTT-3.2.2-4]
     function (err) {
+      this.client.connecting = false
       if (!err) {
+        this.client.connected = true
         this.client.broker.emit('clientReady', client)
         this.client.emit('connected')
       }
@@ -106,7 +108,7 @@ function authenticate (arg, done) {
     negate)
 
   function negate (err, successful) {
-    if (!client.connected || client.broker.closed) {
+    if (client.closed || client.broker.closed) {
       // a hack, sometimes client.close() or broker.close() happened
       // before authenticate() comes back
       // we stop here for not to register it and deregister it in write()
diff --git a/lib/handlers/index.js b/lib/handlers/index.js
index 57c4907..8ec3d8b 100644
--- a/lib/handlers/index.js
+++ b/lib/handlers/index.js
@@ -11,7 +11,7 @@ const handlePing = require('./ping')
 
 function handle (client, packet, done) {
   if (packet.cmd === 'connect') {
-    if (client.connected || !client._connectTimer) {
+    if (client.connecting || client.connected) {
       // [MQTT-3.1.0-2]
       finish(client.conn, packet, done)
       return
@@ -19,7 +19,8 @@ function handle (client, packet, done) {
     handleConnect(client, packet, done)
     return
   }
-  if (!client.connected) {
+
+  if (!client.connecting && !client.connected) {
     // [MQTT-3.1.0-1]
     finish(client.conn, packet, done)
     return
diff --git a/lib/qos-packet.js b/lib/qos-packet.js
index 090d6dc..ea8ea6b 100644
--- a/lib/qos-packet.js
+++ b/lib/qos-packet.js
@@ -10,7 +10,7 @@ function QoSPacket (original, client) {
 
   if (!original.messageId) {
     this.messageId = client._nextId
-    if (client._nextId === 65535) {
+    if (client._nextId >= 65535) {
       client._nextId = 1
     } else {
       client._nextId++
diff --git a/lib/write.js b/lib/write.js
index f62996e..f262623 100644
--- a/lib/write.js
+++ b/lib/write.js
@@ -3,7 +3,7 @@
 const mqtt = require('mqtt-packet')
 
 function write (client, packet, done) {
-  if (client.conn.writable && client.connected) {
+  if (client.conn.writable && (client.connecting || client.connected)) {
     const result = mqtt.writeToStream(packet, client.conn)
     if (!result && !client.errored && done) {
       client.conn.once('drain', done)


Test patch:
diff --git a/test/basic.js b/test/basic.js
index a8242d8..a4d512e 100644
--- a/test/basic.js
+++ b/test/basic.js
@@ -34,6 +34,37 @@ test('publish QoS 0', function (t) {
   })
 })
 
+test('messageId shoud reset to 1 if it reached 65535', function (t) {
+  t.plan(7)
+
+  const s = connect(setup())
+  t.tearDown(s.broker.close.bind(s.broker))
+
+  const publishPacket = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: 'world',
+    qos: 1,
+    messageId: 42
+  }
+  var count = 0
+  s.broker.on('clientReady', function (client) {
+    subscribe(t, s, 'hello', 1, function () {
+      client._nextId = 65535
+      s.outStream.on('data', function (packet) {
+        if (packet.cmd === 'puback') {
+          t.equal(packet.messageId, 42)
+        }
+        if (packet.cmd === 'publish') {
+          t.equal(packet.messageId, count++ === 0 ? 65535 : 1)
+        }
+      })
+      s.inStream.write(publishPacket)
+      s.inStream.write(publishPacket)
+    })
+  })
+})
+
 test('publish empty topic throws error', function (t) {
   t.plan(1)
 
@@ -266,7 +297,8 @@ test('client closes', function (t) {
   t.plan(5)
 
   const broker = aedes()
-  const client = noError(connect(setup(broker), { clientId: 'abcde' }, function () {
+  const client = noError(connect(setup(broker), { clientId: 'abcde' }))
+  broker.on('clientReady', function () {
     const brokerClient = broker.clients.abcde
     t.equal(brokerClient.connected, true, 'client connected')
     eos(client.conn, t.pass.bind(t, 'client closes'))
@@ -279,11 +311,11 @@ test('client closes', function (t) {
         t.error(err, 'no error')
       })
     })
-  }))
+  })
 })
 
 test('broker closes', function (t) {
-  t.plan(3)
+  t.plan(4)
 
   const broker = aedes()
   const client = noError(connect(setup(broker), {
@@ -292,6 +324,7 @@ test('broker closes', function (t) {
     eos(client.conn, t.pass.bind(t, 'client closes'))
     broker.close(function (err) {
       t.error(err, 'no error')
+      t.ok(broker.closed)
       t.equal(broker.clients.abcde, undefined, 'client instance is removed')
     })
   }))
diff --git a/test/connect.js b/test/connect.js
index d3ac71e..ec69ed1 100644
--- a/test/connect.js
+++ b/test/connect.js
@@ -304,7 +304,8 @@ test('second CONNECT Packet sent from a Client as a protocol violation and disco
   broker.on('clientError', function (client, err) {
     t.equal(err.message, 'Invalid protocol')
   })
-  const s = connect(setup(broker), { clientId: 'abcde' }, function () {
+  const s = connect(setup(broker), { clientId: 'abcde' })
+  s.broker.on('clientReady', function () {
     t.ok(broker.clients.abcde.connected)
     // destory client when there is a 2nd cmd:connect, even the clientId is dfferent
     s.inStream.write(packet)
@@ -475,15 +476,18 @@ test('Test queue limit', function (t) {
   })
 })
 
-;[[0, null, false], [0, null, true], [1, new Error('connection banned'), false], [1, new Error('connection banned'), true]].forEach(function (ele, idx) {
-  const plan = ele[0]
-  const err = ele[1]
-  const ok = ele[2]
-  test('preConnect handler', function (t) {
+;[['fail with no error msg', 2, null, false], ['succeed with no error msg', 8, null, true], ['fail with error msg', 5, new Error('connection banned'), false], ['succeed with error msg', 5, new Error('connection banned'), true]].forEach(function (ele, idx) {
+  const title = ele[0]
+  const plan = ele[1]
+  const err = ele[2]
+  const ok = ele[3]
+  test('preConnect handler - ' + title, function (t) {
     t.plan(plan)
 
     const broker = aedes({
       preConnect: function (client, done) {
+        t.ok(client.connecting)
+        t.notOk(client.connected)
         return done(err, ok)
       }
     })
@@ -500,17 +504,26 @@ test('Test queue limit', function (t) {
       keepalive: 0
     })
     broker.on('client', function (client) {
-      if (ok) {
-        t.pass('connect ok')
+      if (ok && !err) {
+        t.ok(client.connecting)
+        t.notOk(client.connected)
+        t.pass('register client ok')
       } else {
         t.fail('no reach here')
       }
     })
+    broker.on('clientReady', function (client) {
+      t.notOk(client.connecting)
+      t.ok(client.connected)
+      t.pass('connect ok')
+    })
     broker.on('clientError', function (client, err) {
       t.fail('no client error')
     })
     broker.on('connectionError', function (client, err) {
       if (err) {
+        t.notOk(client.connecting)
+        t.notOk(client.connected)
         t.equal(err.message, 'connection banned')
       } else {
         t.fail('no connection error')
diff --git a/test/events.js b/test/events.js
index dc39e82..70549b4 100644
--- a/test/events.js
+++ b/test/events.js
@@ -123,15 +123,18 @@ test('Emit event when broker closed', function (t) {
   broker.close()
 })
 
-test('Emit closed event one only when dobule broker.close()', function (t) {
-  t.plan(1)
+test('Emit closed event one only when double broker.close()', function (t) {
+  t.plan(4)
 
   const broker = aedes()
   broker.on('closed', function () {
     t.pass('closed')
   })
+  t.notOk(broker.closed)
   broker.close()
+  t.ok(broker.closed)
   broker.close()
+  t.ok(broker.closed)
 })
 
 test('Test backpressure aedes published function', function (t) {