Repository: moscajs/aedes
Pull request #408 (2020-02-05 18:05:29)

Refactored Subscribe Handler

gnought: I think it is best to use `connecting` flag https://github.com/moscajs/aedes/pull/409 to replace `restore` property

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/CODE_OF_CONDUCT.md b/CODE_OF_CONDUCT.md
index 380c245..0d5ca1b 100644
--- a/CODE_OF_CONDUCT.md
+++ b/CODE_OF_CONDUCT.md
@@ -2,75 +2,83 @@
 
 ## Our Pledge
 
-In the interest of fostering an open and welcoming environment, we as
-contributors and maintainers pledge to make participation in our project and
-our community a harassment-free experience for everyone, regardless of age, body
-size, disability, ethnicity, sex characteristics, gender identity and expression,
-level of experience, education, socio-economic status, nationality, personal
-appearance, race, religion, or sexual identity and orientation.
+We as members, contributors, and leaders pledge to make participation in our community a harassment-free experience for everyone, regardless of age, body size, visible or invisible disability, ethnicity, sex characteristics, gender identity and expression, level of experience, education, socio-economic status, nationality, personal appearance, race, religion, or sexual identity and orientation.
+
+We pledge to act and interact in ways that contribute to an open, welcoming, diverse, inclusive, and healthy community.
 
 ## Our Standards
 
-Examples of behavior that contributes to creating a positive environment
-include:
+Examples of behavior that contributes to a positive environment for our community include:
 
-* Using welcoming and inclusive language
-* Being respectful of differing viewpoints and experiences
-* Gracefully accepting constructive criticism
-* Focusing on what is best for the community
-* Showing empathy towards other community members
+* Demonstrating empathy and kindness toward other people
+* Being respectful of differing opinions, viewpoints, and experiences
+* Giving and gracefully accepting constructive feedback
+* Accepting responsibility and apologizing to those affected by our mistakes, and learning from the experience
+* Focusing on what is best not just for us as individuals, but for the overall community
 
-Examples of unacceptable behavior by participants include:
+Examples of unacceptable behavior include:
 
-* The use of sexualized language or imagery and unwelcome sexual attention or
-  advances
-* Trolling, insulting/derogatory comments, and personal or political attacks
+* The use of sexualized language or imagery, and sexual attention or
+  advances of any kind
+* Trolling, insulting or derogatory comments, and personal or political attacks
 * Public or private harassment
-* Publishing others' private information, such as a physical or electronic
-  address, without explicit permission
+* Publishing others' private information, such as a physical or email
+  address, without their explicit permission
 * Other conduct which could reasonably be considered inappropriate in a
   professional setting
 
-## Our Responsibilities
+## Enforcement Responsibilities
 
-Project maintainers are responsible for clarifying the standards of acceptable
-behavior and are expected to take appropriate and fair corrective action in
-response to any instances of unacceptable behavior.
+Community leaders are responsible for clarifying and enforcing our standards of acceptable behavior and will take appropriate and fair corrective action in response to any behavior that they deem inappropriate, threatening, offensive, or harmful.
 
-Project maintainers have the right and responsibility to remove, edit, or
-reject comments, commits, code, wiki edits, issues, and other contributions
-that are not aligned to this Code of Conduct, or to ban temporarily or
-permanently any contributor for other behaviors that they deem inappropriate,
-threatening, offensive, or harmful.
+Community leaders have the right and responsibility to remove, edit, or reject comments, commits, code, wiki edits, issues, and other contributions that are not aligned to this Code of Conduct, and will communicate reasons for moderation decisions when appropriate.
 
 ## Scope
 
-This Code of Conduct applies within all project spaces, and it also applies when
-an individual is representing the project or its community in public spaces.
-Examples of representing a project or community include using an official
-project e-mail address, posting via an official social media account, or acting
-as an appointed representative at an online or offline event. Representation of
-a project may be further defined and clarified by project maintainers.
+This Code of Conduct applies within all community spaces, and also applies when an individual is officially representing the community in public spaces. Examples of representing our community include using an official e-mail address, posting via an official social media account, or acting as an appointed representative at an online or offline event.
 
 ## Enforcement
 
-Instances of abusive, harassing, or otherwise unacceptable behavior may be
-reported by contacting the project team at hello@matteocollina.com. All
-complaints will be reviewed and investigated and will result in a response that
-is deemed necessary and appropriate to the circumstances. The project team is
-obligated to maintain confidentiality with regard to the reporter of an incident.
-Further details of specific enforcement policies may be posted separately.
+Instances of abusive, harassing, or otherwise unacceptable behavior may be reported to the community leaders responsible for enforcement at hello@matteocollina.com. All complaints will be reviewed and investigated promptly and fairly.
+
+All community leaders are obligated to respect the privacy and security of the reporter of any incident.
+
+## Enforcement Guidelines
+
+Community leaders will follow these Community Impact Guidelines in determining the consequences for any action they deem in violation of this Code of Conduct:
+
+### 1. Correction
+
+**Community Impact**: Use of inappropriate language or other behavior deemed unprofessional or unwelcome in the community.
+
+**Consequence**: A private, written warning from community leaders, providing clarity around the nature of the violation and an explanation of why the behavior was inappropriate. A public apology may be requested.
+
+### 2. Warning
 
-Project maintainers who do not follow or enforce the Code of Conduct in good
-faith may face temporary or permanent repercussions as determined by other
-members of the project's leadership.
+**Community Impact**: A violation through a single incident or series of actions.
+
+**Consequence**: A warning with consequences for continued behavior. No interaction with the people involved, including unsolicited interaction with those enforcing the Code of Conduct, for a specified period of time. This includes avoiding interactions in community spaces as well as external channels like social media. Violating these terms may lead to a temporary or permanent ban.
+
+### 3. Temporary Ban
+
+**Community Impact**: A serious violation of community standards, including sustained inappropriate behavior.
+
+**Consequence**: A temporary ban from any sort of interaction or public communication with the community for a specified period of time. No public or private interaction with the people involved, including unsolicited interaction with those enforcing the Code of Conduct, is allowed during this period. Violating these terms may lead to a permanent ban.
+
+### 4. Permanent Ban
+
+**Community Impact**: Demonstrating a pattern of violation of community standards, including sustained inappropriate behavior,  harassment of an individual, or aggression toward or disparagement of classes of individuals.
+
+**Consequence**: A permanent ban from any sort of public interaction within the community.
 
 ## Attribution
 
-This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 1.4,
-available at https://www.contributor-covenant.org/version/1/4/code-of-conduct.html
+This Code of Conduct is adapted from the [Contributor Covenant][homepage], version 2.0,
+available at https://www.contributor-covenant.org/version/2/0/code_of_conduct.html.
+
+Community Impact Guidelines were inspired by [Mozilla's code of conduct enforcement ladder](https://github.com/mozilla/diversity).
 
 [homepage]: https://www.contributor-covenant.org
 
-For answers to common questions about this code of conduct, see
-https://www.contributor-covenant.org/faq
+For answers to common questions about this code of conduct, see the FAQ at
+https://www.contributor-covenant.org/faq. Translations are available at https://www.contributor-covenant.org/translations.
diff --git a/README.md b/README.md
index daa9061..6f6798a 100644
--- a/README.md
+++ b/README.md
@@ -1,549 +1,156 @@
+<!-- markdownlint-disable MD013 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)
-[![devDependencies Status](https://david-dm.org/moscajs/aedes/dev-status.svg)](https://david-dm.org/moscajs/aedes?type=dev)
-<br/>
+
+![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)
-[![NPM version](https://img.shields.io/npm/v/aedes.svg?style=flat)](https://www.npmjs.com/package/aedes)
-[![NPM downloads](https://img.shields.io/npm/dm/aedes.svg?style=flat)](https://www.npmjs.com/package/aedes)
-
-[![js-standard-style](https://cdn.rawgit.com/feross/standard/master/badge.svg)](https://github.com/feross/standard)
-
-Barebone MQTT server that can run on any stream server.
-
-| QoS 0 | QoS 1 | QoS 2 | auth | [bridge][bridge_protocol] | $SYS topics | SSL | [dynamic topics][dynamic_topics] | cluster | websockets | plugin system | MQTT 5 |
-| ----- | ----- | ----- | ---- | ------------------------- | ----------- | --- | -------------------------------- | ------- | ---------- | ------------- | ------ |
-| ✔     | ✔     | ✔     | ✔    | ?                         | ✔           | ✔   | ✔                                | ✔       | ✔          | ✔             | ✘      |
-
-* [Install](#install)
-* [Example](#example)
-* [API](#api)
-* [TODO](#todo)
-* [Plugins](#plugins)
-* [Collaborators](#collaborators)
-* [Contributing](#contributing)
-* [Acknowledgements](#acknowledgements)
-* [Mosca Vs Aedes](#mosca-vs-aedes)
-* [License](#license)
+[![js-standard-style](https://img.shields.io/badge/code%20style-standard-brightgreen.svg?style=flat)](http://standardjs.com/)\
+[![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)\
+[![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)
+
+Barebone MQTT server that can run on any stream servers
+
+- [Aedes](#aedes)
+  - [Install](#install)
+  - [API](#api)
+  - [Features](#features)
+  - [Examples](#examples)
+  - [Exensions](#exensions)
+  - [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)
+  - [License](#license)
 
-<a name="install"></a>
 ## Install
-To install aedes, simply use npm:
-
-```
-npm install aedes --save
-```
-
-<a name="example"></a>
-## Example
-
-```js
-var aedes = require('aedes')()
-var server = require('net').createServer(aedes.handle)
-var port = 1883
 
-server.listen(port, function () {
-  console.log('server listening on port', port)
-})
-```
-
-### TLS
-
-```js
-var fs = require('fs')
-var aedes = require('aedes')()
-
-var options = {
-  key: fs.readFileSync('YOUR_TLS_KEY_FILE.pem'),
-  cert: fs.readFileSync('YOUR_TLS_CERT_FILE.pem')
-}
-
-var server = require('tls').createServer(options, aedes.handle)
-
-server.listen(8883, function () {
-  console.log('server started and listening on port 8883')
-})
-```
-
-### WEBSOCKETS
-
-```js
-var aedes = require('./aedes')()
-var server = require('net').createServer(aedes.handle)
-var httpServer = require('http').createServer()
-var ws = require('websocket-stream')
-var port = 1883
-var wsPort = 8888
-
-server.listen(port, function () {
-  console.log('server listening on port', port)
-})
-
-ws.createServer({
-  server: httpServer
-}, aedes.handle)
+To install aedes, simply use npm:
 
-httpServer.listen(wsPort, function () {
-  console.log('websocket server listening on port', wsPort)
-})
+```sh
+npm install aedes
 ```
 
-<a name="api"></a>
 ## API
 
-  * <a href="#constructor"><code><b>aedes()</b></code></a>
-  * <a href="#Server"><code><b>aedes.Server()</b></code></a>
-  * <a href="#handle"><code>instance.<b>handle()</b></code></a>
-  * <a href="#subscribe"><code>instance.<b>subscribe()</b></code></a>
-  * <a href="#publish"><code>instance.<b>publish()</b></code></a>
-  * <a href="#unsubscribe"><code>instance.<b>unsubscribe()</b></code></a>
-  * <a href="#preConnect"><code>instance.<b>preConnect()</b></code></a>
-  * <a href="#authenticate"><code>instance.<b>authenticate()</b></code></a>
-  * <a href="#authorizePublish"><code>instance.<b>authorizePublish()</b></code></a>
-  * <a href="#authorizeSubscribe"><code>instance.<b>authorizeSubscribe()</b></code></a>
-  * <a href="#authorizeForward"><code>instance.<b>authorizeForward()</b></code></a>
-  * <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>
-  * <a href="#clientclose"><code>client.<b>close()</b></code></a>
-
--------------------------------------------------------
-<a name="constructor"></a>
-### aedes([opts])
-
-Creates a new instance of Aedes.
-
-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)
-* `concurrency`: the max number of messages delivered concurrently,
-  defaults to `100`
-* `queueLimit`: the max number of messages queued while client is waiting to connect,
-  defaults to `42`. If the number is exceeded `connectionError` is thrown with error `Client queue limit reached`
-* `heartbeatInterval`: the interval at which the broker heartbeat is
-  emitted, it used by other broker in the cluster, defaults to
-  `60000` milliseconds
-* `connectTimeout`: the max number of milliseconds to wait for the CONNECT
-  packet to arrive, defaults to `30000` milliseconds
-* `id`: id used to identify this broker instance in `$SYS` messages,
-  defaults to `uuidv4()`
-* `decodeProtocol`: function called when a valid buffer is received, see
-  [instance.decodeProtocol()](#decodeProtocol)
-* `preConnect`: function called when a valid CONNECT is received, see
-  [instance.preConnect()](#preConnect)
-* `authenticate`: function used to authenticate clients, see
-  [instance.authenticate()](#authenticate)
-* `authorizePublish`: function used to authorize PUBLISH packets, see
-  [instance.authorizePublish()](#authorizePublish)
-* `authorizeSubscribe`: function used to authorize SUBSCRIBE packets, see
-  [instance.authorizeSubscribe()](#authorizeSubscribe)
-* `authorizeForward`: function used to authorize forwarded packets, see
-  [instance.authorizeForward()](#authorizeForward)
-* `published`: function called when a new packet is published, see
-  [instance.published()](#published)
-
-Events:
-
-* `client`: when a new [Client](#client) successfully connects and register itself to server, [connackSent event will be come after], arguments:
-  1. `client`
-* `clientReady`: when a new [Client](#client) received all its offline messages, it is ready, arguments:
-  1. `client`
-* `clientDisconnect`: when a [Client](#client) disconnects, arguments:
-  1. `client`
-* `clientError`: when a [Client](#client) errors, arguments:
-  1. `client`
-  2. `err`
-* `connectionError` When a [Client](#client) connection errors and there is no clientId attached , arguments:
-  1. `client`
-  2. `err`
-* `keepaliveTimeout`: when a [Client](#client) keepalive times out, arguments:
-  1. `client`
-* `publish`: when a new packet is published, arguments:
-  1. `packet`
-  2. `client`, it will be null if the message is published using
-     [`publish`](#publish). It is by design that the broker heartbeat will be on publish event, in this case `client` is null
-* `ack`: when a packet published to a client is delivered successfully with QoS 1 or QoS 2, arguments:
-  1. `packet`, this will be the original PUBLISH packet in QoS 1, and PUBREL in QoS 2
-  2. `client`
-* `ping`: when a [Client](#client) sends a ping, arguments:
-  1. `packet`
-  2. `client`
-* `subscribe`: when a client sends a SUBSCRIBE, arguments:
-  1. `subscriptions`, as defined in the `subscriptions` property of the
-     [SUBSCRIBE](https://github.com/mqttjs/mqtt-packet#subscribe)
-packet.
-  2. `client`
-* `unsubscribe`: when a client sends a UNSUBSCRIBE, arguments:
-  1. `unsubscriptions`, as defined in the `subscriptions` property of the
-     [UNSUBSCRIBE](https://github.com/mqttjs/mqtt-packet#unsubscribe)
-packet.
-  2. `client`
-* `connackSent`: when a CONNACK packet is sent to a client, arguments:
-  1. `packet`
-  2. `client`
-* `closed`: when the broker is closed
-
--------------------------------------------------------
-<a name="Server"></a>
-### new aedes.Server([opts])
-
-Same as [`aedes(opts)`](#constructor).
-Creates a new instance of Aedes.
-This variant is useful with TypeScript or ES modules.
-
--------------------------------------------------------
-<a name="handle"></a>
-### instance.handle(duplex)
-
-Handle the given duplex as a MQTT connection.
-
-```js
-var aedes = require('./aedes')()
-var server = require('net').createServer(aedes.handle)
-```
-
--------------------------------------------------------
-<a name="subscribe"></a>
-### instance.subscribe(topic, func(packet, cb), done)
+- [Aedes object](./docs/Aedes.md)
+- [Client object](./docs/Client.md)
 
-After `done` is called, every time [publish](#publish) is invoked on the
-instance (and on any other connected instances) with a matching `topic` the `func` function will be called.
+## Features
 
-`func` needs to call `cb` after receiving the message.
+- Full compatible with MQTT 3.1 and 3.1.1
+- Standard TCP Support
+- SSL / TLS
+- WebSocket Support
+- Message Persistence
+- Automatic Reconnect
+- Offline Buffering
+- Backpress-support API
+- High Availability
+- Authenticaion and Authorization
+- `$SYS` support
+- Pluggable middlewares
+- [Dynamic Topics][dynamic_topics] Support
+- MQTT Bridge Support between aedes
+- MQTT 5.0 _(not support yet)_
+- [Bridge Protocol][bridge_protocol] _(not support yet)_
 
-It supports backpressure.
+## Examples
 
--------------------------------------------------------
-<a name="publish"></a>
-### instance.publish(packet, done)
+- [Examples](./docs/Examples.md)
 
-Publish the given packet to subscribed clients and functions. It supports backpressure.
+## Exensions
 
-A packet contains the following properties:
-
-```js
-{
-  cmd: 'publish',
-  qos: 2,
-  topic: 'test',
-  payload: new Buffer('test'),
-  retain: false
-}
-```
+- [aedes-logging]: Logging module for Aedes, based on Pino
+- [aedes-stats]: Stats for Aedes
+- [aedes-protocol-decoder]: Protocol decoder for Aedes MQTT Broker
 
-Only the `topic` property is mandatory.
-Both `topic` and `payload` can be `Buffer` objects instead of strings.
+## Middleware Plugins
 
--------------------------------------------------------
-<a name="unsubscribe"></a>
-### instance.unsubscribe(topic, func(packet, cb), done)
+### Persistence
 
-The reverse of [subscribe](#subscribe).
-------------------------------------------------------
-<a name="decodeProtocol"></a>
-### instance.decodeProtocol(client, buffer)
-
-It will be called when aedes instance `trustProxy` is `true` and that it receives a first valid buffer from client. client object state is in default and its connected state is false. Use `aedes-protocol-decoder` to parse https headers (x-real-ip | x-forwarded-for) and proxy protocol v1 and v2 to retrieve information in client.connDetails. Override to supply custom protocolDecoder logic, if it returns an object with data property, this property will be parsed as an mqtt-packet.
-
-```js
-instance.decodeProtocol = function(client, buffer) {
-  var protocol = yourDecoder(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
-
-```js
-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))
-
-It will be called when a new client connects. Override to supply custom
-authentication logic.
-
-```js
-instance.authenticate = function (client, username, password, callback) {
-  callback(null, username === 'matteo')
-}
-```
-Other return codes can passed as follows :-
-
-```js
-instance.authenticate = function (client, username, password, callback) {
-  var error = new Error('Auth error')
-  error.returnCode = 1
-  callback(error, null)
-}
-```
-The return code values and their responses which can be passed are given below:
+- [aedes-persistence]: In-memory implementation of an Aedes persistence
+- [aedes-persistence-mongodb]: MongoDB persistence for Aedes
+- [aedes-persistence-redis]: Redis persistence for Aedes
+- [aedes-persistence-level]: LevelDB persistence for Aedes
+- [aedes-persistence-nedb]: NeDB persistence for Aedes
 
-*  `1` - Unacceptable protocol version
-*  `2` - Identifier rejected
-*  `3` - Server unavailable
-*  `4` - Bad user name or password
-
--------------------------------------------------------
-<a name="authorizePublish"></a>
-### instance.authorizePublish(client, packet, done(err))
-
-It will be called when a client publishes a message. Override to supply custom
-authorization logic.
-
-```js
-instance.authorizePublish = function (client, packet, callback) {
-  if (packet.topic === 'aaaa') {
-    return callback(new Error('wrong topic'))
-  }
-
-  if (packet.topic === 'bbb') {
-    packet.payload = new Buffer('overwrite packet payload')
-  }
-
-  callback(null)
-}
-```
-
--------------------------------------------------------
-<a name="authorizeSubscribe"></a>
-### instance.authorizeSubscribe(client, pattern, done(err, pattern))
-
-It will be called when a client subscribes to a topic. Override to supply custom
-authorization logic.
-
-```js
-instance.authorizeSubscribe = function (client, sub, callback) {
-  if (sub.topic === 'aaaa') {
-    return callback(new Error('wrong topic'))
-  }
-
-  if (sub.topic === 'bbb') {
-    // overwrites subscription
-    sub.qos = sub.qos + 2
-  }
-
-  callback(null, sub)
-}
-```
+### MQEmitter
 
-To negate a subscription, set the subscription to `null`:
+- [mqemitter]: An opinionated memory Message Queue with an emitter-style API
+- [mqemitter-redis]: Redis-powered mqemitter
+- [mqemitter-mongodb]: Mongodb based mqemitter
+- [mqemitter-child-process]: Share the same mqemitter between a hierarchy of
+  child processes
+- [mqemitter-cs]: Expose a MQEmitter via a simple client/server protocol
+- [mqemitter-p2p]: A P2P implementation of MQEmitter, based on HyperEmitter and
+  a Merkle DAG
+- [mqemitter-aerospike]: Aerospike mqemitter
 
-```js
-instance.authorizeSubscribe = function (client, sub, callback) {
-  if (sub.topic === 'aaaa') {
-    sub = null
-  }
-
-  callback(null, sub)
-}
-```
-
--------------------------------------------------------
-<a name="authorizeForward"></a>
-### instance.authorizeForward(clientId, packet)
-
-It will be called when a client is set to recieve a message. Override to supply custom
-authorization logic.
-
-```js
-instance.authorizeForward = function (client, packet) {
-  if (packet.topic === 'aaaa' && client.id === "I should not see this") {
-    return null
-    // also works with return undefined
-  }
-
-  if (packet.topic === 'bbb') {
-    packet.payload = new Buffer('overwrite packet payload')
-  }
-
-  return packet
-}
-```
-
--------------------------------------------------------
-<a name="published"></a>
-### instance.published(packet, client, done())
-
-It will be called after a message is published.
-`client` will be null for internal messages.
-Override to supply custom authorization logic.
-
--------------------------------------------------------
-<a name="close"></a>
-### instance.close([cb])
-
-Disconnects all clients.
-
-Events:
-
-* `closed`, in case the broker is closed
-
--------------------------------------------------------
-<a name="Client"></a>
-### Client
-
-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
-
-The client's connection stream object.
-
-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.
-
-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.
-
--------------------------------------------------------
-<a name="clientpublish"></a>
-### client#publish(message, [callback])
-
-Publish the given `message` to this client. QoS 1 and 2 are fully
-respected, while the retained flag is not.
-
-`message` is a [PUBLISH](https://github.com/mqttjs/mqtt-packet#publish) packet.
-
-`callback`  will be called when the message has been sent, but not acked.
-
--------------------------------------------------------
-<a name="clientsubscribe"></a>
-### client#subscribe(subscriptions, [callback])
-
-Subscribe the client to the list of topics.
-
-`subscription` can be:
-
-1. a single object in the format `{ topic: topic, qos: qos }`
-2. an array of the above
-3. a full [subscribe
-   packet](https://github.com/mqttjs/mqtt-packet#subscribe),
-specifying a `messageId` will send suback to the client.
-
-`callback`  will be called when the subscription is completed.
-
--------------------------------------------------------
-<a name="clientunsubscribe"></a>
-### client#unsubscribe(topicObjects, [callback])
-
-Unsubscribe the client to the list of topics.
+## Collaborators
 
-The topic objects can be as follows :-
+- [__Gavin D'mello__](https://github.com/GavinDmello)
+- [__Behrad Zari__](https://github.com/behrad)
+- [__Gnought__](https://github.com/gnought)
+- [__Daniel Lando__](https://github.com/robertsLando)
 
-1. a single object in the format `{ topic: topic, qos: qos }`
-2. an array of the above
+## Acknowledgements
 
-`callback`  will be called when the unsubscriptions are completed.
+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.
 
--------------------------------------------------------
-<a name="clientclose"></a>
-### client#close([cb])
+## Contribution
 
-Disconnects the client
+[![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)
 
--------------------------------------------------------
-<a name="clientpresence"></a>
-### client presence
+Want to contribute? Check our list of
+[features/bugs](https://github.com/moscajs/aedes/projects/1)
 
-You can subscribe on the following `$SYS` topics to get client presence:
+## Contributors
 
- - `$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
+This project exists thanks to all the people who contribute ❤️
 
-## Plugins
+[![Backers](https://opencollective.com/aedes/contributors.svg?width=890&button=false)](https://opencollective.com/aedes#backers)
 
-- [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
+### Backers
 
-## Collaborators
+Thank you to all our backers! 🙏 [[Become a backer](https://opencollective.com/aedes#backer)]
 
-* [__Gavin D'mello__](https://github.com/GavinDmello)
-* [__Behrad Zari__](https://github.com/behrad)
-* [__Gnought__](https://github.com/gnought)
+[![Backers](https://opencollective.com/aedes/backers.svg?width=890)](https://opencollective.com/aedes#backers)
 
-## Acknowledgements
+### Sponsors
 
-This library is born after a lot of discussion with all
-[Mosca](http://npm.im/mosca) users and how that was deployed in
-production. This addresses your concerns about performance and
-stability.
+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)]
 
-## Contributing
+[![Sponsor 1](https://opencollective.com/aedes/sponsor/0/avatar.svg)](https://opencollective.com/aedes/sponsor/0/website)
 
-Want to contribute? Check our list of features/bugs [here](https://github.com/moscajs/aedes/projects/1)
+[![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 [mqtt-benchmark](https://github.com/krylovsk/mqtt-benchmark) with command:
+Example benchmark test with 1000 clients sending 5000 QoS 1 messsages. Used
+[mqtt-benchmark] with command:
 
-`mqtt-benchmark --broker tcp://localhost:1883 --clients 1000 --qos 1 --count 5000`
+```sh
+mqtt-benchmark --broker tcp://localhost:1883 --clients 1000 --qos 1 --count 5000
+```
 
-### Aedes
+### Benchmark: Aedes
 
-```
+```sh
 ========= TOTAL (1000) =========
 Total Ratio:                 1.000 (5000000/5000000)
 Total Runtime (sec):         178.495
@@ -556,9 +163,9 @@ Average Bandwidth (msg/sec): 28.115
 Total Bandwidth (msg/sec):   28114.678
 ```
 
-### Mosca
+### Benchmark: [Mosca](http://www.npmjs.com/mosca)
 
-```
+```sh
 ========= TOTAL (1000) =========
 Total Ratio:                 1.000 (5000000/5000000)
 Total Runtime (sec):         264.934
@@ -573,7 +180,25 @@ Total Bandwidth (msg/sec):   18925.942
 
 ## License
 
-MIT
+Licensed under [MIT](./LICENSE).
 
+[bridge_protocol]: https://github.com/mqtt/mqtt.github.io/wiki/bridge_protocol
 [dynamic_topics]: https://github.com/mqtt/mqtt.github.io/wiki/are_topics_dynamic
-[bridge_protocol]: https://github.com/mqtt/mqtt.github.io/wiki/bridge_protocol
\ No newline at end of file
+[mqtt-benchmark]: https://github.com/krylovsk/mqtt-benchmark
+
+[aedes-logging]: https://www.npmjs.com/aedes-logging
+[aedes-stats]: https://www.npmjs.com/aedes-stats
+[aedes-protocol-decoder]: https://www.npmjs.com/aedes-protocol-decoder
+[aedes-persistence]: https://www.npmjs.com/aedes-persistence
+[aedes-persistence-mongodb]: https://www.npmjs.com/aedes-persistence-mongodb
+[aedes-persistence-redis]: https://www.npmjs.com/aedes-persistence-redis
+[aedes-persistence-level]: https://www.npmjs.com/aedes-persistence-level
+[aedes-persistence-nedb]: https://www.npmjs.com/aedes-persistence-nedb
+
+[mqemitter]: https://www.npmjs.com/mqemitter
+[mqemitter-redis]: https://www.npmjs.com/mqemitter-redis
+[mqemitter-mongodb]: https://www.npmjs.com/mqemitter-mongodb
+[mqemitter-child-process]: https://www.npmjs.com/mqemitter-child-process
+[mqemitter-cs]: https://www.npmjs.com/mqemitter-cs
+[mqemitter-p2p]: https://www.npmjs.com/mqemitter-p2p
+[mqemitter-aerospike]: https://www.npmjs.com/mqemitter-aerospike
diff --git a/aedes.d.ts b/aedes.d.ts
index 2e71b75..9c39601 100644
--- a/aedes.d.ts
+++ b/aedes.d.ts
@@ -4,7 +4,8 @@
 
 /// <reference types="node" />
 
-import { IPublishPacket, ISubscribePacket, ISubscription, IUnsubscribePacket } from 'mqtt-packet'
+import { IConnackPacket, IPingreqPacket, IPublishPacket, IPubrelPacket, ISubscribePacket, ISubscription, IUnsubscribePacket } from 'mqtt-packet'
+import { AedesPacket } from 'aedes-packet'
 import { Duplex } from 'stream'
 import { Socket } from 'net'
 import { IncomingMessage } from 'http'
@@ -18,27 +19,42 @@ declare namespace aedes {
     UNNACCEPTABLE_PROTOCOL = 1,
     IDENTIFIER_REJECTED = 2,
     SERVER_UNAVAILABLE = 3,
-    BAD_USERNAME_OR_PASSWORD = 4
+    BAD_USERNAME_OR_PASSWORD = 4,
+    NOT_AUTHORIZED = 5
   }
 
-  type PreConnectHandler = (client: Client, callback: (err: Error | null, success: boolean) => void) => void
+  type Connection = Duplex | Socket
+
+  type Subscription = ISubscription
+  type Subscriptions = { subscriptions: Subscription[] }
+  type SubscribePacket = ISubscribePacket & { cmd: 'subscribe' }
+  type UnsubscribePacket = IUnsubscribePacket & { cmd: 'unsubscribe' }
+
+  type PublishPacket = IPublishPacket & { cmd: 'publish' }
+  type AedesPublishPacket = PublishPacket & AedesPacket
+
+  type ConnackPacket = IConnackPacket & { cmd: 'connack' }
+  type PubrelPacket = IPubrelPacket & { cmd: 'pubrel' }
+  type PingreqPacket = IPingreqPacket & { cmd: 'pingreq' }
+
+  type PreConnectHandler = (client: Client, callback: (error: Error | null, success: boolean) => void) => void
 
   type AuthenticateError = Error & { returnCode: AuthErrorCode }
 
   type AuthenticateHandler = (
     client: Client,
     username: string,
-    password: string,
-    done: (err: AuthenticateError | null, success: boolean | null) => void
+    password: Buffer,
+    done: (error: AuthenticateError | null, success: boolean | null) => void
   ) => void
 
-  type AuthorizePublishHandler = (client: Client, packet: IPublishPacket, callback: (err?: Error | null) => void) => void
+  type AuthorizePublishHandler = (client: Client, packet: PublishPacket, callback: (error?: Error | null) => void) => void
 
-  type AuthorizeSubscribeHandler = (client: Client, subscription: ISubscription, callback: (err: Error | null, subscription?: ISubscription | null) => void) => void
+  type AuthorizeSubscribeHandler = (client: Client, subscription: Subscription, callback: (error: Error | null, subscription?: Subscription | null) => void) => void
 
-  type AuthorizeForwardHandler = (client: Client, packet: IPublishPacket) => IPublishPacket | null | void
+  type AuthorizeForwardHandler = (client: Client, packet: AedesPublishPacket) => AedesPublishPacket | null | void
 
-  type PublishedHandler = (packet: IPublishPacket, client: Client, callback: (err?: Error | null) => void) => void
+  type PublishedHandler = (packet: AedesPublishPacket, client: Client, callback: (error?: Error | null) => void) => void
 
   interface AedesOptions {
     mq?: any
@@ -53,46 +69,58 @@ declare namespace aedes {
     authorizeForward?: AuthorizeForwardHandler
     published?: PublishedHandler
     queueLimit?: number
+    maxClientsIdLength?: number
   }
   interface Client extends EventEmitter {
     id: string
     clean: boolean
-    conn: Socket
+    conn: Connection
     req?: IncomingMessage
+    connecting: boolean
+    connected: boolean
+    closed: boolean
 
-    on (event: 'connected', callback: () => void): this
-    on (event: 'error', callback: (error: Error) => void): this
+    on (event: 'connected', listener: () => void): this
+    on (event: 'error', listener: (error: Error) => void): this
 
-    publish (message: IPublishPacket, callback?: (error?: Error) => void): void
+    publish (message: PublishPacket, callback?: (error?: Error) => void): void
     subscribe (
-      subscriptions: ISubscription | ISubscription[] | ISubscribePacket,
+      subscriptions: Subscriptions | Subscription | Subscription[] | SubscribePacket,
       callback?: (error?: Error) => void
     ): void
-    unsubscribe (topicObjects: ISubscription | ISubscription[], callback?: (error?: Error) => void): void
+    unsubscribe (topicObjects: Subscriptions | Subscription | Subscription[] | UnsubscribePacket, callback?: (error?: Error) => void): void
     close (callback?: () => void): void
   }
 
   interface Aedes extends EventEmitter {
-    handle: (stream: Duplex) => void
+    id: string
+    connectedClients: number
+    closed: boolean
+
+    handle: (stream: Connection) => Client
 
-    on (event: 'closed', callback: () => void): this
-    on (event: 'client' | 'clientReady' | 'clientDisconnect' | 'keepaliveTimeout' | 'connackSent', callback: (client: Client) => void): this
-    on (event: 'clientError' | 'connectionError', callback: (client: Client, error: Error) => void): this
-    on (event: 'ping' | 'publish' | 'ack', callback: (packet: any, client: Client) => void): this
-    on (event: 'subscribe' | 'unsubscribe', callback: (subscriptions: ISubscription | ISubscription[] | ISubscribePacket, client: Client) => void): this
+    on (event: 'closed', listener: () => void): this
+    on (event: 'client' | 'clientReady' | 'clientDisconnect' | 'keepaliveTimeout', listener: (client: Client) => void): this
+    on (event: 'clientError' | 'connectionError', listener: (client: Client, error: Error) => void): this
+    on (event: 'connackSent', listener: (packet: ConnackPacket, client: Client) => void): this
+    on (event: 'ping', listener: (packet: PingreqPacket, client: Client) => void): this
+    on (event: 'publish', listener: (packet: AedesPublishPacket, client: Client) => void): this
+    on (event: 'ack', listener: (packet: PublishPacket | PubrelPacket, client: Client) => void): this
+    on (event: 'subscribe', listener: (subscriptions: Subscription[], client: Client) => void): this
+    on (event: 'unsubscribe', listener: (unsubscriptions: string[], client: Client) => void): this
 
     publish (
-      packet: IPublishPacket & { topic: string | Buffer },
-      callback: () => void
+      packet: PublishPacket,
+      callback: (error?: Error) => void
     ): void
     subscribe (
       topic: string,
-      deliverfunc: (packet: ISubscribePacket, callback: () => void) => void,
+      deliverfunc: (packet: AedesPublishPacket, callback: () => void) => void,
       callback: () => void
     ): void
     unsubscribe (
       topic: string,
-      deliverfunc: (packet: IUnsubscribePacket, callback: () => void) => void,
+      deliverfunc: (packet: AedesPublishPacket, callback: () => void) => void,
       callback: () => void
     ): void
     close (callback?: () => void): void
diff --git a/aedes.js b/aedes.js
index 0bd05f7..2f1f46f 100644
--- a/aedes.js
+++ b/aedes.js
@@ -1,19 +1,19 @@
 'use strict'
 
-const mqemitter = require('mqemitter')
 const EE = require('events').EventEmitter
 const util = require('util')
-const memory = require('aedes-persistence')
 const parallel = require('fastparallel')
 const series = require('fastseries')
 const uuidv4 = require('uuid/v4')
-const Packet = require('aedes-packet')
 const bulk = require('bulk-write-stream')
 const reusify = require('reusify')
+const { pipeline } = require('readable-stream')
+const Packet = require('aedes-packet')
+const memory = require('aedes-persistence')
+const mqemitter = require('mqemitter')
 const Client = require('./lib/client')
 
-module.exports = Aedes
-Aedes.Server = Aedes
+module.exports = Aedes.Server = Aedes
 
 const defaultOptions = {
   concurrency: 100,
@@ -28,7 +28,8 @@ const defaultOptions = {
   published: defaultPublished,
   trustProxy: false,
   trustedProxies: [],
-  queueLimit: 42
+  queueLimit: 42,
+  maxClientsIdLength: 23
 }
 
 function Aedes (opts) {
@@ -44,6 +45,7 @@ function Aedes (opts) {
   this.counter = 0
   this.queueLimit = opts.queueLimit
   this.connectTimeout = opts.connectTimeout
+  this.maxClientsIdLength = opts.maxClientsIdLength
   this.mq = opts.mq || mqemitter(opts)
   this.handle = function handle (conn, req) {
     conn.setMaxListeners(opts.concurrency * 2)
@@ -92,9 +94,15 @@ function Aedes (opts) {
   this._clearWillInterval = setInterval(function () {
     Object.keys(that.brokers).forEach(deleteOldBrokers)
 
-    that.persistence
-      .streamWill(that.brokers)
-      .pipe(bulk.obj(receiveWills))
+    pipeline(
+      that.persistence.streamWill(that.brokers),
+      bulk.obj(receiveWills),
+      function done (err) {
+        if (err) {
+          that.emit('error', err)
+        }
+      }
+    )
   }, opts.heartbeatInterval * 4)
 
   function receiveWills (chunks, done) {
diff --git a/docs/Aedes.md b/docs/Aedes.md
new file mode 100644
index 0000000..315685f
--- /dev/null
+++ b/docs/Aedes.md
@@ -0,0 +1,414 @@
+<!-- markdownlint-disable MD013 MD024 -->
+# Aedes
+
+- [Aedes](#aedes)
+  - [new Aedes([options]) / new Aedes.Server([options])](#new-aedesoptions--new-aedesserveroptions)
+  - [aedes.id](#aedesid)
+  - [aedes.connectedClients](#aedesconnectedclients)
+  - [aedes.closed](#aedesclosed)
+  - [Event: client](#event-client)
+  - [Event: clientReady](#event-clientready)
+  - [Event: clientDisconnect](#event-clientdisconnect)
+  - [Event: clientError](#event-clienterror)
+  - [Event: connectionError](#event-connectionerror)
+  - [Event: keepaliveTimeout](#event-keepalivetimeout)
+  - [Event: publish](#event-publish)
+  - [Event: ack](#event-ack)
+  - [Event: ping](#event-ping)
+  - [Event: subscribe](#event-subscribe)
+  - [Event: unsubscribe](#event-unsubscribe)
+  - [Event: connackSent](#event-connacksent)
+  - [Event: closed](#event-closed)
+  - [aedes.handle (stream)](#aedeshandle-stream)
+  - [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)
+  - [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)
+  - [Handler: authorizePublish (client, packet, callback)](#handler-authorizepublish-client-packet-callback)
+  - [Handler: authorizeSubscribe (client, subscription, callback)](#handler-authorizesubscribe-client-subscription-callback)
+  - [Handler: authorizeForward (client, packet)](#handler-authorizeforward-client-packet)
+  - [Handler: published (packet, client, callback)](#handler-published-packet-client-callback)
+
+## new Aedes([options]) / new Aedes.Server([options])
+
+- options `<object>`
+  - `mq` [`<MQEmitter>`](../README.md#mqemitter) a middleware helps to deliver messages to subscribed clients. __Default__: `mqemitter`
+  - `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>`
+
+Create a new Aedes server.
+
+Aedes is the class and function exposed by this module. It can be created by `Aedes()` or using `new Aedes()`. An variant `aedes.Server` is for TypeScript or ES modules.
+
+## aedes.id
+
+- `<string>` __Default__: `uuidv4()`
+
+Server unique identifier.
+
+## aedes.connectedClients
+
+- `<number>` __Default__: 0
+
+Number of connected clients in server.
+
+## aedes.closed
+
+- `<boolean>` __Default__: false
+
+a read-only flag indicates if server is closed or not.
+
+## Event: client
+
+- `client` [`<Client>`](./Client.md)
+
+Emitted when the `client` registers itself to server. The `client` in not ready yet. Its [`connecting`](./Client.md##clientconnecting) state equals to `true`.
+
+Server publishes a SYS topic `$SYS/<aedes.id>/new/clients` to inform it registers the client into its registration pool. `client.id` is the payload.
+
+## Event: clientReady
+
+- `client` [`<Client>`](./Client.md)
+
+Emitted when the `client` has received all its offline messages and be initialized. The `client` [`connected`](./Client.md##clientconnected) state equals to `true` and is ready for processing incoming messages.
+
+## Event: clientDisconnect
+
+- `client` [`<Client>`](./Client.md)
+
+Emitted when a client disconnects.
+
+Server publishes a SYS topic `$SYS/<aedes.id>/disconnect/clients` to inform it deregisters the client. `client.id` is the payload.
+
+## Event: clientError
+
+- `client` [`<Client>`](./Client.md)
+- `error` `<Error>`
+
+Emitted when an error occurs.
+
+## Event: connectionError
+
+- `client` [`<Client>`](./Client.md)
+- `error` `<Error>`
+
+Emitted when an error occurs. Unlike `clientError` it raises only when `client` is uninitialized.
+
+## Event: keepaliveTimeout
+
+- `client` [`<Client>`](./Client.md)
+
+Emitted when timeout happes in the `client` keepalive.
+
+## Event: publish
+
+- `packet` `<aedes-packet>` & [`PUBLISH`][PUBLISH]
+- `client` [`<Client>`](./Client.md) | `null`
+
+Emitted when servers delivers the `packet` to subscribed `client`. If there are no clients subscribed to the `packet` topic, server still publish the `packet` and emit thie event. `client` is `null` when `packet` is an internal message like aedes heartbeat message and LWT.
+
+> _Note! `packet` belongs `aedes-packet` type. Some properties belongs to aedes internal, any changes on them will break aedes internal flow._
+
+## Event: ack
+
+- `packet` `<object>` [`PUBLISH`][PUBLISH] for QoS 1, [`PUBREL`][PUBREL] for QoS 2
+- `client` [`<Client>`](./Client.md)
+
+Emitted an QoS 1 or 2 acknowledgement when the `packet` successfully delivered to the `client`.
+
+## Event: ping
+
+- `packet` `<object>` [`PINGREQ`][PINGREQ]
+- `client` [`<Client>`](./Client.md)
+
+Emitted when `client` sends a `PINGREQ`.
+
+## Event: subscribe
+
+- `subscriptions` `<object>`
+- `client` [`<Client>`](./Client.md)
+
+Emitted when `client` successfully subscribe the `subscriptions` in server.
+
+`subscriptions` are an array of `{ topic: topic, qos: qos }`
+
+## Event: unsubscribe
+
+- `unsubscriptions` `Array<string>`
+- `client` [`<Client>`](./Client.md)
+
+Emitted when `client` successfully unsubscribe the `subscriptions` in server.
+
+`unsubscriptions` are an array of unsubscribed topics.
+
+## Event: connackSent
+
+- `packet` `<object>` [`CONNACK`][CONNACK]
+- `client` [`<Client>`](./Client.md)
+
+Emitted when server sends an acknowledge to `client`. Please refer to the MQTT specification for the explanation of returnCode object property in `CONNACK`.
+
+## Event: closed
+
+Emitted when server is closed.
+
+## aedes.handle (stream)
+
+- stream: `<net.Socket>` | `<stream.Duplex>`
+- Returns: [`<Client>`](./Client.md)
+
+A connection listener that pipe stream to aedes.
+
+```js
+const aedes = require('./aedes')()
+const server = require('net').createServer(aedes.handle)
+```
+
+## aedes.subscribe (topic, deliverfunc, callback)
+
+- topic: `<string>`
+- deliverfunc: `<Function>` `(packet, cb) => void`
+  - packet: `<aedes-packet>` & [`PUBLISH`][PUBLISH]
+  - cb: `<Function>`
+- callback: `<Function>`
+
+Directly subscribe a `topic` in server side. Bypass [`authorizeSubscribe`](#handler-authorizesubscribe-client-subscription-callback)
+
+The `topic` and `deliverfunc` is a compound key to differentiate the uniqueness of its subscription pool. `topic` could be the one that is existed, in this case `deliverfunc` will be invoked as well as `SUBSCRIBE` does.
+
+`deliverfunc` supports backpressue.
+
+In aedes internal, `deliverfunc` is a function that delivers messages to subscribed clients.
+
+> _Note! `packet` belongs `aedes-packet` type. Some properties belongs to aedes internal, any changes on them will break aedes internal flow._
+
+In general you would find most properities in `packet` is same as what the incoming [`PUBLISH`][PUBLISH] is. For sure `cmd` property in `packet` structure in `deliverfunc` must be `publish`.
+
+> _Note! it requires `deliverfunc` to call `cb` before the function returns, otherwise some subscribed clients with  same `topic` will not receive messages._
+
+`callback` is invoked when server successfully registers the subscription.
+
+## aedes.unsubscribe (topic, deliverfunc, callback)
+
+Reverse of [aedes.subscribe](#aedessubscribe-topic-deliverfunc-callback).
+
+> _Note! the `deliverfunc` should be same as when `aedes.subscribe` does, otherwise the unsubscription will fail._
+
+## aedes.publish (packet, callback)
+
+- `packet` `<object>` [`PUBLISH`][PUBLISH]
+- `callback` `<Function>` `(error) => void`
+  - error `<Error>` | `null`
+
+Directly deliver `packet` on behalf of server to subscribed clients. Bypass [`authorizePublish`](#handler-authorizepublish-client-packet-callback).
+
+`callback` will be invoked with `error` arugments after finish.
+
+## aedes.close ([callback])
+
+- callback: `<Function>`
+
+Close aedes server and disconnects all clients.
+
+`callback` will be invoked when server is closed.
+
+## Handler: decodeProtocol (client, buffer)
+
+- client: [`<Client>`](./Client.md)
+- buffer: `<Buffer>`
+
+Invoked when aedes instance `trustProxy` is `true`
+
+It targets to decode wrapped protocols (e.g. websocket and PROXY) into plain raw mqtt stream.
+
+`aedes-protocol-decoder` is an example to parse https headers (x-real-ip | x-forwarded-for) and proxy protocol v1 and v2 to retrieve information in `client.connDetails`.
+
+```js
+aedes.decodeProtocol = function(client, buffer) {
+  return yourDecoder(client, buffer)
+}
+```
+
+## Handler: preConnect (client, callback)
+
+- client: [`<Client>`](./Client.md)
+- callback: `<Function>` `(error, successful) => void`
+  - error `<Error>` | `null`
+  - successful `<boolean>`
+
+Invoked when server receives a valid [`CONNECT`][CONNECT] packet.
+
+`client` object is in default state. If invoked `callback` with no errors and `successful` be `true`, server will continue to establish a session.
+
+Any `error` will be raised in `connectionError` event.
+
+Some Use Cases:
+
+1. Rate Limit / Throttle by `client.conn.remoteAddress`
+2. Check `aedes.connectedClient` to limit maximum connections
+3. IP blacklisting
+
+```js
+aedes.preConnect = function(client, callback) {
+  callback(null, client.conn.remoteAddress === '::1') {
+}
+```
+
+```js
+aedes.preConnect = function(client, callback) {
+  callback(new Error('connection error'), client.conn.remoteAddress !== '::1') {
+}
+```
+
+## Handler: authenticate (client, username, password, callback)
+
+- client: [`<Client>`](./Client.md)
+- username: `<string>`
+- password: `<Buffer>`
+- callback: `<Function>` `(error, successful) => void`
+  - error `<Error>` | `null`
+  - successful `<boolean>`
+
+Invoked after `preConnect`.
+
+Server parses the [`CONNECT`][CONNECT] packet, initializes `client` object which set `client.id` to match the one in [`CONNECT`][CONNECT] packet and extract `username` and `password` as parameters for user-defined authentication flow.
+
+If invoked `callback` with no errors and `successful` be `true`, server authenticates `client` and continues to setup the client session.
+
+If authenticated, server acknowledges a [`CONNACK`][CONNACK] with `returnCode=0`, otherwise `returnCode=5`. Users could define the value between `2` and `5` by defining a `returnCode` property in `error` object.
+
+```js
+aedes.authenticate = function (client, username, password, callback) {
+  callback(null, username === 'matteo')
+}
+```
+
+```js
+aedes.authenticate = function (client, username, password, callback) {
+  var error = new Error('Auth error')
+  error.returnCode = 4
+  callback(error, null)
+}
+```
+
+Please refer to [Connect Return Code](http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Table_3.1_-) to see their meanings.
+
+## Handler: authorizePublish (client, packet, callback)
+
+- client: [`<Client>`](./Client.md)
+- packet: `<object>` [`PUBLISH`][PUBLISH]
+- callback: `<Function>` `(error) => void`
+  - error `<Error>` | `null`
+
+Invoked when
+
+1. publish LWT to all online clients
+2. incoming client publish
+
+If invoked `callback` with no errors, server authorizes the packet otherwise emits `clientError` with `error`.
+
+```js
+aedes.authorizePublish = function (client, packet, callback) {
+  if (packet.topic === 'aaaa') {
+    return callback(new Error('wrong topic'))
+  }
+  if (packet.topic === 'bbb') {
+    packet.payload = Buffer.from('overwrite packet payload')
+  }
+  callback(null)
+}
+```
+
+## Handler: authorizeSubscribe (client, subscription, callback)
+
+- client: [`<Client>`](./Client.md)
+- subscription: `<object>`
+- callback: `<Function>` `(error) => void`
+  - error `<Error>` | `null`
+  - subscription: `<object>` | `null`
+
+Invoked when
+
+1. restore subscriptions in non-clean session.
+2. incoming client `SUBSCRIBE`
+
+`subscrption` is a dictionary object like `{ topic: hello, qos: 0 }`.
+
+If invoked `callback` with no errors, server authorizes the packet otherwise emits `clientError` with `error`.
+
+In general user should not touch the `subscription` and pass to callback, but server gives an option to change the subscription on-the-fly.
+
+```js
+aedes.authorizeSubscribe = function (client, sub, callback) {
+  if (sub.topic === 'aaaa') {
+    return callback(new Error('wrong topic'))
+  }
+  if (sub.topic === 'bbb') {
+    // overwrites subscription
+    sub.topic = 'foo'
+    sub.qos = 1
+  }
+  callback(null, sub)
+}
+```
+
+To negate a subscription, set the subscription to `null`:
+
+```js
+aedes.authorizeSubscribe = function (client, sub, callback) {
+  // prohibited to subscribe 'aaaa' and suppress error
+  callback(null, sub.topic === 'aaaa' ? null : sub)
+}
+```
+
+## Handler: authorizeForward (client, packet)
+
+- client: [`<Client>`](./Client.md)
+- packet: `<aedes-packet>` & [`PUBLISH`][PUBLISH]
+- Returns: `<aedes-packet>` | `null`
+
+Invoked when
+
+1. aedes sends retained messages when client reconnects
+2. aedes pre-delivers subscribed message to clients
+
+Return `null` will not forward `packet` to clients.
+
+In general user should not touch the `packet` and return it what it is, but server gives an option to change the `packet` on-the-fly and forward it to clients.
+
+> _Note! `packet` belongs `aedes-packet` type. Some properties belongs to aedes internal, any changes on them will break aedes internal flow._
+
+```js
+aedes.authorizeForward = function (client, packet) {
+  if (packet.topic === 'aaaa' && client.id === "I should not see this") {
+    return
+  }
+  if (packet.topic === 'bbb') {
+    packet.payload = new Buffer('overwrite packet payload')
+  }
+  return packet
+}
+```
+
+## Handler: published (packet, client, callback)
+
+- packet: `<aedes-packet>` & [`PUBLISH`][PUBLISH]
+- client: [`<Client>`](./Client.md)
+- callback: `<Function>`
+
+same as [`Event: publish`](#event-publish), but provides a backpressure functionality.
+
+[CONNECT]: https://github.com/mqttjs/mqtt-packet#connect
+[CONNACK]: https://github.com/mqttjs/mqtt-packet#connack
+[SUBSCRIBE]: https://github.com/mqttjs/mqtt-packet#subscribe
+[PINGREQ]: https://github.com/mqttjs/mqtt-packet#pingreq
+[PUBLISH]: https://github.com/mqttjs/mqtt-packet#publish
+[PUBREL]: https://github.com/mqttjs/mqtt-packet#pubrel
diff --git a/docs/Client.md b/docs/Client.md
new file mode 100644
index 0000000..4860071
--- /dev/null
+++ b/docs/Client.md
@@ -0,0 +1,135 @@
+<!-- markdownlint-disable MD013 MD024 -->
+# Client
+
+- [Client](#client)
+  - [new Client(aedes, stream, request)](#new-clientaedes-stream-request)
+  - [client.conn](#clientconn)
+  - [client.req](#clientreq)
+  - [client.connecting](#clientconnecting)
+  - [client.connected](#clientconnected)
+  - [client.closed](#clientclosed)
+  - [client.id](#clientid)
+  - [client.clean](#clientclean)
+  - [Event: connected](#event-connected)
+  - [Event: error](#event-error)
+  - [client.publish (packet, [callback])](#clientpublish-packet-callback)
+  - [client.subscribe (subscriptions, [callback])](#clientsubscribe-subscriptions-callback)
+  - [client.unsubscribe (unsubscriptions, [callback])](#clientunsubscribe-unsubscriptions-callback)
+  - [client.close ([callback])](#clientclose-callback)
+
+## new Client(aedes, stream, request)
+
+- aedes [`<Aedes>`](./Aedes.md)
+- stream: `<net.Socket>` | `<stream.Duplex>`
+- request: `<http.IncomingMessage>`
+- Returns: `<Client>`
+
+## client.conn
+
+- `<net.Socket>` | `<stream.Duplex>`
+
+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`][websocket-stream], it's the `stream` argument passed to the websocket `handle` function in [`websocket-stream #on-the-server`][websocket-stream-doc-on-the-server]].
+
+## client.req
+
+- `<http.IncomingMessage>`
+
+only for [`websocket-stream`][websocket-stream]. It is a HTTP Websocket upgrade request object passed to websocket `handle` function in [`websocket-stream #on-the-server`][websocket-stream-doc-on-the-server]. It gives an option for accessing headers or cookies.
+
+## client.connecting
+
+- `<boolean>` __Default__: `false`
+
+a read-only flag, 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.
+
+## client.connected
+
+- `<boolean>` __Default__: `false`
+
+a read-only flag, it is `true` when `connected` event is emitted, and `false` when client is closed.
+
+## client.closed
+
+- `<boolean>` __Default__: `false`
+
+a read-only flag indicates if client is closed or not.
+
+## client.id
+
+- `<string>` __Default__: `aedes_${shortid()}`
+
+Client unique identifier, specified by CONNECT packet. It is `null` in [`aedes.preConnect`](./Aedes.md#handler-preconnect-client-callback) unless the handler passes.
+
+## client.clean
+
+- `<boolean>` __Default__: `true`
+
+Client clean flag, specified by CONNECT packet.
+
+## Event: connected
+
+Same as aedes [`clientReady`](./Aedes.md#event-clientready) but in client-wise.
+
+## Event: error
+
+- `error` `<Error>`
+
+Emitted when an error occurs.
+
+## client.publish (packet, [callback])
+
+- `packet` `<object>` [`PUBLISH`][PUBLISH]
+- `callback` `<Function>` `(error) => void`
+  - error `<Error>` | `null`
+
+Publish the given `packet` to this client. QoS 1 and 2 are fully supported, while the retained flag is not.
+
+`callback`  will be invoked when the message has been sent, but not acked.
+
+## client.subscribe (subscriptions, [callback])
+
+- `subscriptions` `<object>`
+- `callback` `<Function>` `(error) => void`
+  - error `<Error>` | `null`
+
+Subscribe client to the list of topics.
+
+`subscriptions` can be:
+
+1. a single object in the format `{ topic: topic, qos: qos }`
+2. an array of the above
+3. a full [`SUBSCRIBE`][SUBSCRIBE], specifying a `messageId` will send suback to the client.
+
+`callback`  will be invoked when the subscription is completed.
+
+## client.unsubscribe (unsubscriptions, [callback])
+
+- `unsubscriptions` `<object>`
+- `callback` `<Function>`
+
+Unsubscribe client to the list of topics.
+
+`unsubscriptions` can be:
+
+1. a single object in the format `{ topic: topic, qos: qos }`
+2. an array of the above
+3. a full [`UNSUBSCRIBE`][UNSUBSCRIBE]
+
+`callback`  will be invoked when the unsubscriptions are completed.
+
+## client.close ([callback])
+
+Disconnect client
+
+`callback` will be invoked when client is closed.
+
+[PUBLISH]: https://github.com/mqttjs/mqtt-packet#publish
+[SUBSCRIBE]: https://github.com/mqttjs/mqtt-packet#subscribe
+[UNSUBSCRIBE]: https://github.com/mqttjs/mqtt-packet#unsubscribe
+
+[websocket-stream]: https://www.npmjs.com/websocket-stream
+[websocket-stream-doc-on-the-server]: https://github.com/maxogden/websocket-stream/blob/master/readme.md#on-the-server
diff --git a/docs/Examples.md b/docs/Examples.md
new file mode 100644
index 0000000..b89597b
--- /dev/null
+++ b/docs/Examples.md
@@ -0,0 +1,48 @@
+<!-- markdownlint-disable MD013 MD024 -->
+# Examples
+
+Simple plain MQTT server
+
+```js
+const aedes = require('aedes')()
+const server = require('net').createServer(aedes.handle)
+const port = 1883
+
+server.listen(port, function () {
+  console.log('server started and listening on port ', port)
+})
+```
+
+MQTT over TLS / MQTTS
+
+```js
+const fs = require('fs')
+const aedes = require('aedes')()
+const port = 8883
+
+const options = {
+  key: fs.readFileSync('YOUR_PRIVATE_KEY_FILE.pem'),
+  cert: fs.readFileSync('YOUR_PUBLIC_CERT_FILE.pem')
+}
+
+const server = require('tls').createServer(options, aedes.handle)
+
+server.listen(port, function () {
+  console.log('server started and listening on port ', port)
+})
+```
+
+MQTT server over WebSocket
+
+```js
+const aedes = require('./aedes')()
+const httpServer = require('http').createServer()
+const ws = require('websocket-stream')
+const port = 8888
+
+ws.createServer({ server: httpServer }, aedes.handle)
+
+httpServer.listen(port, function () {
+  console.log('websocket server listening on port ', port)
+})
+```
diff --git a/lib/client.js b/lib/client.js
index 9f6667c..c194abb 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
@@ -216,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) {
@@ -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..093f26a 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]
@@ -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) {
@@ -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()
@@ -119,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)))
   }
 }
 
@@ -176,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
   }
@@ -228,7 +232,6 @@ function emptyQueue (arg, done) {
   pipeline(
     outgoing,
     through.obj(function clearQueue (data, enc, next) {
-      // console.log('clearQueue')
       var packet = new QoSPacket(data, client)
       packet.writeCallback = next
       persistence.outgoingUpdate(client, packet, emptyQueueFilter)
diff --git a/lib/handlers/index.js b/lib/handlers/index.js
index 57c4907..1caabfe 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
@@ -30,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/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)
diff --git a/package.json b/package.json
index c59a44c..296eef4 100644
--- a/package.json
+++ b/package.json
@@ -5,9 +5,10 @@
   "main": "aedes.js",
   "types": "aedes.d.ts",
   "scripts": {
-    "lint": "npm run lint:standard && npm run lint:typescript",
+    "lint": "npm run lint:standard && npm run lint:typescript && npm run lint:markdown",
     "lint:standard": "standard --verbose | snazzy",
     "lint:typescript": "standard --parser @typescript-eslint/parser --plugin @typescript-eslint/eslint-plugin test/types/*.ts aedes.d.ts",
+    "lint:markdown": "markdownlint docs/*.md README.md",
     "unit": "tap --no-esm -J test/*.js",
     "unit:report": "tap --no-esm -J test/*.js --cov --coverage-report=html --coverage-report=cobertura | tee out.tap",
     "typescript": "tsc --project ./test/types/tsconfig.json",
@@ -68,12 +69,14 @@
   ],
   "license": "MIT",
   "devDependencies": {
-    "@types/node": "^12.12.26",
-    "@typescript-eslint/eslint-plugin": "^2.19.0",
-    "@typescript-eslint/parser": "^2.19.0",
+    "@sinonjs/fake-timers": "^6.0.0",
+    "@types/node": "^12.12.27",
+    "@typescript-eslint/eslint-plugin": "^2.19.2",
+    "@typescript-eslint/parser": "^2.19.2",
     "concat-stream": "^2.0.0",
     "duplexify": "^4.1.1",
     "license-checker": "^25.0.1",
+    "markdownlint-cli": "^0.22.0",
     "mqtt": "^3.0.0",
     "mqtt-connection": "^4.0.0",
     "pre-commit": "^1.2.2",
@@ -84,7 +87,7 @@
     "websocket-stream": "^5.5.0"
   },
   "dependencies": {
-    "aedes-packet": "^2.1.0",
+    "aedes-packet": "^2.3.0",
     "aedes-persistence": "^7.2.1",
     "aedes-protocol-decoder": "^1.0.0",
     "bulk-write-stream": "^2.0.1",
@@ -93,7 +96,7 @@
     "fastparallel": "^2.3.0",
     "fastseries": "^1.7.2",
     "from2": "^2.3.0",
-    "mqemitter": "^3.0.0",
+    "mqemitter": "^4.0.0",
     "mqtt-packet": "^6.3.0",
     "proxy-protocol-js": "^4.0.3",
     "readable-stream": "^3.5.0",


Test patch:
diff --git a/test/auth.js b/test/auth.js
index fc512e2..1f18232 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 () {
@@ -361,7 +444,6 @@ test('authorize publish', function (t) {
     t.notOk(Object.prototype.hasOwnProperty.call(packet, 'messageId'), 'should not contain messageId in QoS 0')
     expected.brokerId = s.broker.id
     expected.brokerCounter = s.broker.counter
-    delete expected.dup
     delete expected.length
     t.deepEqual(packet, expected, 'packet matches')
     cb()
@@ -409,7 +491,6 @@ test('authorize waits for authenticate', function (t) {
     t.notOk(Object.prototype.hasOwnProperty.call(packet, 'messageId'), 'should not contain messageId in QoS 0')
     expected.brokerId = s.broker.id
     expected.brokerCounter = s.broker.counter
-    delete expected.dup
     delete expected.length
     t.deepEqual(packet, expected, 'packet matches')
     cb()
@@ -459,7 +540,6 @@ test('authorize publish from configOptions', function (t) {
     t.notOk(Object.prototype.hasOwnProperty.call(packet, 'messageId'), 'should not contain messageId in QoS 0')
     expected.brokerId = s.broker.id
     expected.brokerCounter = s.broker.counter
-    delete expected.dup
     delete expected.length
     t.deepEqual(packet, expected, 'packet matches')
     cb()
@@ -816,7 +896,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 +912,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 +1022,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/basic.js b/test/basic.js
index a8242d8..4c3a54c 100644
--- a/test/basic.js
+++ b/test/basic.js
@@ -5,6 +5,17 @@ const eos = require('end-of-stream')
 const { setup, connect, subscribe, noError } = require('./helper')
 const aedes = require('../')
 
+test('test aedes.Server', function (t) {
+  t.plan(1)
+
+  const broker = new aedes.Server()
+  t.tearDown(broker.close.bind(broker))
+
+  connect(setup(broker), {}, function () {
+    t.pass('connected')
+  })
+})
+
 test('publish QoS 0', function (t) {
   t.plan(2)
 
@@ -16,7 +27,8 @@ test('publish QoS 0', function (t) {
     topic: 'hello',
     payload: Buffer.from('world'),
     qos: 0,
-    retain: false
+    retain: false,
+    dup: false
   }
 
   s.broker.mq.on('hello', function (packet, cb) {
@@ -34,6 +46,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)
 
@@ -262,11 +305,28 @@ test('disconnect', function (t) {
   })
 })
 
+test('disconnect client on wrong cmd', function (t) {
+  t.plan(1)
+
+  const s = noError(connect(setup()), t)
+  t.tearDown(s.broker.close.bind(s.broker))
+
+  s.broker.on('clientDisconnect', function () {
+    t.pass('closed stream')
+  })
+
+  s.broker.on('clientReady', function (c) {
+    // don't use stream write here because it will throw an error on mqtt_packet genetete
+    c._parser.emit('packet', { cmd: 'pippo' })
+  })
+})
+
 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 +339,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 +352,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/client-pub-sub.js b/test/client-pub-sub.js
index cbad57b..e0a5d60 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) {
@@ -37,6 +37,56 @@ test('publish direct to a single client QoS 0', function (t) {
   })
 })
 
+test('publish direct to a single client throws error', function (t) {
+  t.plan(1)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  broker.persistence.outgoingEnqueue = function (sub, packet, done) {
+    done(new Error('Throws error'))
+  }
+
+  broker.on('client', function (client) {
+    client.publish({
+      topic: 'hello',
+      payload: Buffer.from('world'),
+      qos: 1,
+      retain: false
+    }, function (err) {
+      t.pass('Throws error', err.message, 'throws error')
+    })
+  })
+
+  connect(setup(broker), { clean: false })
+})
+
+test('publish direct to a single client throws error 2', function (t) {
+  t.plan(1)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  broker.persistence.outgoingUpdate = function (client, packet, done) {
+    done(new Error('Throws error'), client, packet)
+  }
+
+  broker.on('client', function (client) {
+    client.publish({
+      topic: 'hello',
+      payload: Buffer.from('world'),
+      qos: 1,
+      retain: false
+    }, () => {})
+
+    client.once('error', function (err) {
+      t.pass('Throws error', err.message, 'throws error')
+    })
+  })
+
+  connect(setup(broker), { clean: false })
+})
+
 test('publish direct to a single client QoS 1', function (t) {
   t.plan(2)
 
@@ -75,6 +125,41 @@ test('publish direct to a single client QoS 1', function (t) {
   })
 })
 
+test('publish QoS 2 throws error in pubrel', function (t) {
+  t.plan(2)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  const s = connect(setup(broker))
+
+  broker.on('clientError', function (c, err) {
+    t.pass('throws error')
+  })
+
+  s.outStream.on('data', function (packet) {
+    if (packet.cmd === 'publish') {
+      s.inStream.write({
+        cmd: 'pubrec',
+        messageId: packet.messageId
+      })
+      s.broker.persistence.outgoingUpdate = function (client, pubrel, cb) {
+        cb(new Error('error'))
+      }
+    }
+  })
+
+  broker.on('clientReady', function (client) {
+    client.publish({
+      topic: 'hello',
+      payload: Buffer.from('world'),
+      qos: 2
+    }, function (err) {
+      t.error(err, 'no error')
+    })
+  })
+})
+
 test('publish direct to a single client QoS 2', function (t) {
   t.plan(3)
 
@@ -133,7 +218,8 @@ test('emit a `ack` event on PUBACK for QoS 1 [clean=false]', function (t) {
     topic: 'hello',
     payload: Buffer.from('world'),
     qos: 1,
-    retain: false
+    retain: false,
+    dup: false
   }
 
   broker.on('clientReady', function (client) {
@@ -417,10 +503,10 @@ test('unsubscribe a client', function (t) {
       qos: 0
     }, function (err) {
       t.error(err, 'no error')
-      client.unsubscribe({
+      client.unsubscribe([{
         topic: 'hello',
         qos: 0
-      }, function (err) {
+      }], function (err) {
         t.error(err, 'no error')
       })
     })
@@ -428,6 +514,61 @@ test('unsubscribe a client', function (t) {
   connect(setup(broker))
 })
 
+test('unsubscribe throws error', function (t) {
+  t.plan(2)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  broker.on('client', function (client) {
+    client.subscribe({
+      topic: 'hello',
+      qos: 0
+    }, function (err) {
+      t.error(err, 'no error')
+      broker.unsubscribe = function (topic, func, done) {
+        done(new Error('error'))
+      }
+      client.unsubscribe({
+        topic: 'hello',
+        qos: 0
+      }, function () {
+        t.pass('throws error')
+      })
+    })
+  })
+  connect(setup(broker))
+})
+
+test('unsubscribe throws error 2', function (t) {
+  t.plan(2)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  broker.on('client', function (client) {
+    client.subscribe({
+      topic: 'hello',
+      qos: 2
+    }, function (err) {
+      t.error(err, 'no error')
+      broker.persistence.removeSubscriptions = function (client, unsubscriptions, done) {
+        done(new Error('error'))
+      }
+      client.unsubscribe({
+        unsubscriptions: [{
+          topic: 'hello',
+          qos: 2
+        }],
+        messageId: 42
+      }, function () {
+        t.pass('throws error')
+      })
+    })
+  })
+  connect(setup(broker))
+})
+
 test('subscribe a client programmatically multiple topics', function (t) {
   t.plan(3)
 
@@ -604,3 +745,125 @@ 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,
+    dup: 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,
+    dup: 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/connect.js b/test/connect.js
index d3ac71e..83b855f 100644
--- a/test/connect.js
+++ b/test/connect.js
@@ -178,6 +178,76 @@ test('clients without clientid and clean=true on MQTT 3.1.1 will get a generated
   })
 })
 
+test('client connect error while fetching subscriptions', function (t) {
+  t.plan(1)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  const s = setup(broker)
+
+  broker.persistence.subscriptionsByClient = function (c, cb) {
+    cb(new Error('error'), [], c)
+  }
+
+  s.inStream.write({
+    cmd: 'connect',
+    protocolId: 'MQTT',
+    protocolVersion: 4,
+    clean: false,
+    clientId: 'my-client',
+    keepalive: 0
+  })
+
+  broker.on('clientError', function (client, err) {
+    t.pass('throws error')
+  })
+})
+
+test('client connect clear outgoing', function (t) {
+  t.plan(1)
+
+  var clientId = 'abcde'
+  var brokerId = 'pippo'
+
+  const broker = aedes({ id: brokerId })
+  t.tearDown(broker.close.bind(broker))
+
+  var subs = [{ clientId: clientId }]
+  var packet = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: Buffer.from('world'),
+    qos: 1,
+    brokerId: brokerId,
+    brokerCounter: 2,
+    retain: true,
+    messageId: 42,
+    dup: false
+  }
+
+  broker.persistence.outgoingEnqueueCombi(subs, packet, function () {
+    const s = setup(broker)
+
+    console.log(broker.persistence._outgoing[clientId][0])
+
+    s.inStream.write({
+      cmd: 'connect',
+      protocolId: 'MQTT',
+      protocolVersion: 4,
+      clean: true,
+      clientId: clientId,
+      keepalive: 0
+    })
+
+    broker.on('clientReady', function (client) {
+      broker.persistence.outgoingUpdate(client, packet, function (err) {
+        t.equal('no such packet', err.message, 'packet not found')
+      })
+    })
+  })
+})
+
 test('clients with zero-byte clientid and clean=true on MQTT 3.1.1 will get a generated clientId', function (t) {
   t.plan(4)
 
@@ -234,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)
 
@@ -304,7 +400,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 +572,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 +600,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) {
diff --git a/test/qos1.js b/test/qos1.js
index 48de675..cdedbc8 100644
--- a/test/qos1.js
+++ b/test/qos1.js
@@ -33,6 +33,29 @@ test('publish QoS 1', function (t) {
   })
 })
 
+test('publish QoS 1 throws error', function (t) {
+  t.plan(1)
+
+  const s = connect(setup())
+  t.tearDown(s.broker.close.bind(s.broker))
+
+  s.broker.persistence.subscriptionsByTopic = function (packet, done) {
+    return done(new Error('Throws error'))
+  }
+
+  s.inStream.write({
+    cmd: 'publish',
+    topic: 'hello',
+    payload: 'world',
+    qos: 1,
+    messageId: 42
+  })
+
+  s.broker.on('error', function (err) {
+    t.equal('Throws error', err.message, 'Throws error')
+  })
+})
+
 test('publish QoS 1 and check offline queue', function (t) {
   t.plan(13)
 
@@ -65,7 +88,8 @@ test('publish QoS 1 and check offline queue', function (t) {
     payload: 'world',
     qos: 1,
     messageId: 10,
-    retain: false
+    retain: false,
+    dup: false
   }
   var queue = []
   subscribe(t, subscriber, 'hello', 1, function () {
@@ -225,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/qos2.js b/test/qos2.js
index 789708a..61acaac 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,109 @@ 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,
+      dup: 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)
 
@@ -452,6 +556,7 @@ test('multiple publish and store one', function (t) {
     payload: Buffer.from('world'),
     qos: 2,
     retain: false,
+    dup: false,
     messageId: 42
   }
 
diff --git a/test/topics.js b/test/topics.js
index 71e96ad..9b039f3 100644
--- a/test/topics.js
+++ b/test/topics.js
@@ -74,7 +74,7 @@ test('publish invalid topic with +', function (t) {
   })
 })
 
-;['base/#/sub', 'base/#sub', 'base/sub#', 'base/xyz+/sub', 'base/+xyz/sub'].forEach(function (topic) {
+;['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(1)
 
diff --git a/test/types/index.ts b/test/types/index.ts
index badacb7..fb898cb 100644
--- a/test/types/index.ts
+++ b/test/types/index.ts
@@ -1,9 +1,8 @@
 /* eslint no-unused-vars: 0 */
 /* eslint no-undef: 0 */
 
-import { Server, Client, AuthenticateError } from '../../aedes'
-import { IPublishPacket, ISubscribePacket, ISubscription, IUnsubscribePacket } from 'mqtt-packet'
-import { createServer } from 'net'
+import { Server, Client, AuthenticateError, AedesPublishPacket, PublishPacket, Subscription } from '../../aedes'
+import { createServer, Socket } from 'net'
 
 const broker = Server({
   concurrency: 100,
@@ -13,14 +12,14 @@ const broker = Server({
     if (client.req) {
       callback(new Error('not websocket stream'), false)
     }
-    if (client.conn.remoteAddress === '::1') {
+    if (client.conn instanceof Socket && client.conn.remoteAddress === '::1') {
       callback(null, true)
     } else {
       callback(new Error('connection error'), false)
     }
   },
-  authenticate: (client: Client, username: string, password: string, callback) => {
-    if (username === 'test' && password === 'test') {
+  authenticate: (client: Client, username: string, password: Buffer, callback) => {
+    if (username === 'test' && password === Buffer.from('test')) {
       callback(null, true)
     } else {
       const error = new Error() as AuthenticateError
@@ -29,7 +28,7 @@ const broker = Server({
       callback(error, false)
     }
   },
-  authorizePublish: (client: Client, packet: IPublishPacket, callback) => {
+  authorizePublish: (client: Client, packet: PublishPacket, callback) => {
     if (packet.topic === 'aaaa') {
       return callback(new Error('wrong topic'))
     }
@@ -40,7 +39,7 @@ const broker = Server({
 
     callback(null)
   },
-  authorizeSubscribe: (client: Client, sub: ISubscription, callback) => {
+  authorizeSubscribe: (client: Client, sub: Subscription, callback) => {
     if (sub.topic === 'aaaa') {
       return callback(new Error('wrong topic'))
     }
@@ -52,7 +51,7 @@ const broker = Server({
 
     callback(null, sub)
   },
-  authorizeForward: (client: Client, packet: IPublishPacket) => {
+  authorizeForward: (client: Client, packet: AedesPublishPacket) => {
     if (packet.topic === 'aaaa' && client.id === 'I should not see this') {
       return null
       // also works with return undefined
@@ -90,7 +89,7 @@ broker.on('keepaliveTimeout', client => {
   console.log(`client: ${client.id} timed out`)
 })
 
-broker.on('connackSent', client => {
+broker.on('connackSent', (packet, client) => {
   console.log(`client: ${client.id} connack sent`)
 })
 
@@ -103,15 +102,15 @@ broker.on('connectionError', client => {
 })
 
 broker.on('ping', (packet, client) => {
-  console.log(`client: ${client.id} ping with packet ${packet.id}`)
+  console.log(`client: ${client.id} ping with packet ${packet.cmd}`)
 })
 
 broker.on('publish', (packet, client) => {
-  console.log(`client: ${client.id} published packet ${packet.id}`)
+  console.log(`client: ${client.id} published packet ${packet.cmd}`)
 })
 
 broker.on('ack', (packet, client) => {
-  console.log(`client: ${client.id} ack with packet ${packet.id}`)
+  console.log(`client: ${client.id} ack with packet ${packet.cmd}`)
 })
 
 broker.on('subscribe', (subscriptions, client) => {
@@ -122,17 +121,17 @@ broker.on('unsubscribe', (subscriptions, client) => {
   console.log(`client: ${client.id} subsribe`)
 })
 
-broker.subscribe('aaaa', (packet: ISubscribePacket, cb) => {
+broker.subscribe('aaaa', (packet: AedesPublishPacket, 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: AedesPublishPacket, cb) => {
   console.log('cmd')
-  console.log(packet.unsubscriptions)
+  console.log(packet.cmd)
   cb()
 }, () => {
   console.log('done unsubscribing')
diff --git a/test/will.js b/test/will.js
index cc7385c..513045f 100644
--- a/test/will.js
+++ b/test/will.js
@@ -4,6 +4,7 @@ const { test } = require('tap')
 const memory = require('aedes-persistence')
 const { setup, connect, noError } = require('./helper')
 const aedes = require('../')
+const Faketimers = require('@sinonjs/fake-timers')
 
 function willConnect (s, opts, connected) {
   opts = opts || {}
@@ -113,6 +114,30 @@ test('delivers old will in case of a crash', function (t) {
   })
 })
 
+test('delete old broker', function (t) {
+  t.plan(1)
+
+  var clock = Faketimers.install()
+
+  var heartbeatInterval = 100
+  const broker = aedes({
+    heartbeatInterval: heartbeatInterval
+  })
+  t.tearDown(broker.close.bind(broker))
+
+  var brokerId = 'dummyBroker'
+
+  broker.brokers[brokerId] = Date.now() - heartbeatInterval * 3.5
+
+  setTimeout(() => {
+    t.equal(broker.brokers[brokerId], undefined, 'Broker deleted')
+  }, heartbeatInterval * 4)
+
+  clock.tick(heartbeatInterval * 4)
+
+  clock.uninstall()
+})
+
 test('store the will in the persistence', function (t) {
   t.plan(5)
 
@@ -374,3 +399,129 @@ test('does not store multiple will with same clientid', function (t) {
     })
   })
 })
+
+test('don\'t delivers a will if broker alive', function (t) {
+  const persistence = memory()
+  const will = {
+    topic: 'mywill',
+    payload: Buffer.from('last will'),
+    qos: 0,
+    retain: false
+  }
+
+  var oldBroker = 'broker1'
+
+  persistence.broker = {
+    id: oldBroker
+  }
+
+  persistence.putWill({
+    id: 'myClientId42'
+  }, will, function (err) {
+    t.error(err, 'no error')
+
+    const opts = {
+      persistence: persistence,
+      heartbeatInterval: 10
+    }
+
+    var count = 0
+
+    const broker = aedes(opts)
+    t.tearDown(broker.close.bind(broker))
+
+    var streamWill = persistence.streamWill
+    persistence.streamWill = function () {
+      // don't pass broker.brokers to streamWill
+      return streamWill.call(persistence)
+    }
+
+    broker.mq.on('mywill', function (packet, cb) {
+      t.fail('Will received')
+      cb()
+    })
+
+    broker.mq.on('$SYS/+/heartbeat', function () {
+      t.pass('Heartbeat received')
+      broker.brokers[oldBroker] = Date.now()
+
+      if (++count === 5) {
+        t.end()
+      }
+    })
+  })
+})
+
+test('handle will publish error', function (t) {
+  t.plan(2)
+  const persistence = memory()
+  const will = {
+    topic: 'mywill',
+    payload: Buffer.from('last will'),
+    qos: 0,
+    retain: false
+  }
+
+  persistence.broker = {
+    id: 'broker1'
+  }
+
+  persistence.putWill({
+    id: 'myClientId42'
+  }, will, function (err) {
+    t.error(err, 'no error')
+
+    const opts = {
+      persistence: persistence,
+      heartbeatInterval: 10
+    }
+
+    persistence.delWill = function (client, cb) {
+      cb(new Error('Throws error'))
+    }
+
+    const broker = aedes(opts)
+    t.tearDown(broker.close.bind(broker))
+
+    broker.once('error', function (err) {
+      t.equal('Throws error', err.message, 'throws error')
+    })
+  })
+})
+
+test('handle will publish error 2', function (t) {
+  t.plan(2)
+  const persistence = memory()
+  const will = {
+    topic: 'mywill',
+    payload: Buffer.from('last will'),
+    qos: 0,
+    retain: true
+  }
+
+  persistence.broker = {
+    id: 'broker1'
+  }
+
+  persistence.putWill({
+    id: 'myClientId42'
+  }, will, function (err) {
+    t.error(err, 'no error')
+
+    const opts = {
+      persistence: persistence,
+      heartbeatInterval: 10
+    }
+
+    persistence.storeRetained = function (packet, cb) {
+      cb(new Error('Throws error'))
+    }
+
+    const broker = aedes(opts)
+    t.tearDown(broker.close.bind(broker))
+
+    broker.once('error', function (err) {
+      t.equal('Throws error', err.message, 'throws error')
+    })
+  })
+})