Merge a00212c into 26473cd

README.md

@@ -252,8 +252,7 @@ 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. A default function 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.
-
+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) {
@@ -517,6 +516,7 @@ The payload will contain the `clientId` of the connected/disconnected client
   - [mqemitter-aerospike](https://github.com/GavinDmello/mqemitter-aerospike): Aerospike mqemitter based on @mcollina 's mqemitter
 - [aedes-logging](https://github.com/moscajs/aedes-logging): Logging module for Aedes, based on Pino
 - [aedes-stats](https://github.com/moscajs/aedes-stats): Stats for Aedes
+- [aedes-protocol-decoder](https://github.com/moscajs/aedes-protocol-decoder): Protocol decoder for Aedes MQTT Broker
 
 ## Collaborators
 

aedes.js

@@ -11,7 +11,6 @@ const Packet = require('aedes-packet')
 const bulk = require('bulk-write-stream')
 const reusify = require('reusify')
 const Client = require('./lib/client')
-const protocolDecoder = require('./lib/protocol-decoder')
 
 module.exports = Aedes
 Aedes.Server = Aedes
@@ -20,7 +19,7 @@ const defaultOptions = {
   concurrency: 100,
   heartbeatInterval: 60000, // 1 minute
   connectTimeout: 30000, // 30 secs
-  decodeProtocol: defaultDecodeProtocol,
+  decodeProtocol: null,
   preConnect: defaultPreConnect,
   authenticate: defaultAuthenticate,
   authorizePublish: defaultAuthorizePublish,
@@ -305,11 +304,6 @@ Aedes.prototype.close = function (cb = noop) {
 
 Aedes.prototype.version = require('./package.json').version
 
-function defaultDecodeProtocol (client, buffer) {
-  const proto = protocolDecoder(client, buffer)
-  return proto
-}
-
 function defaultPreConnect (client, callback) {
   callback(null, true)
 }

lib/client.js

@@ -70,7 +70,7 @@ function Client (broker, conn, req) {
       that._parsingBatch = 0
       var buf = empty
       buf = client.conn.read(null)
-      if (!client.connackSent && client.broker.trustProxy && buf) {
+      if (!client.connackSent && client.broker.decodeProtocol && client.broker.trustProxy && buf) {
         const { data } = client.broker.decodeProtocol(client, buf)
         if (data) {
           client._parser.parse(data)

package.json

@@ -60,6 +60,10 @@
     {
       "name": "Gnought",
       "url": "https://github.com/gnought"
+    },
+    {
+      "name": "Daniel Lando",
+      "url": "https://github.com/robertsLando"
     }
   ],
   "license": "MIT",
@@ -85,6 +89,7 @@
   "dependencies": {
     "aedes-packet": "^2.1.0",
     "aedes-persistence": "^7.2.1",
+    "aedes-protocol-decoder": "^1.0.0",
     "bulk-write-stream": "^2.0.1",
     "end-of-stream": "^1.4.4",
     "fastfall": "^1.5.1",

test/basic.js

@@ -34,6 +34,23 @@ test('publish QoS 0', function (t) {
   })
 })
 
+test('publish empty topic throws error', function (t) {
+  t.plan(1)
+
+  const s = connect(setup())
+  t.tearDown(s.broker.close.bind(s.broker))
+
+  s.inStream.write({
+    cmd: 'publish',
+    topic: '',
+    payload: 'world'
+  })
+
+  s.broker.on('clientError', function (client, err) {
+    t.pass('should emit error')
+  })
+})
+
 test('subscribe QoS 0', function (t) {
   t.plan(4)
 

test/connect.js

@@ -7,6 +7,7 @@ const mqtt = require('mqtt')
 const mqttPacket = require('mqtt-packet')
 const net = require('net')
 const proxyProtocol = require('proxy-protocol-js')
+const { protocolDecoder } = require('aedes-protocol-decoder')
 const { setup, connect, delay } = require('./helper')
 const aedes = require('../')
 
@@ -574,6 +575,7 @@ test('tcp clients have access to the ipAddress from the socket', function (t) {
       }
       done(null, true)
     },
+    decodeProtocol: protocolDecoder,
     trustProxy: true
   })
 
@@ -630,6 +632,7 @@ test('tcp proxied (protocol v1) clients have access to the ipAddress(v4)', funct
       }
       done(null, true)
     },
+    decodeProtocol: protocolDecoder,
     trustProxy: true
   })
 
@@ -687,6 +690,7 @@ test('tcp proxied (protocol v2) clients have access to the ipAddress(v4)', funct
       }
       done(null, true)
     },
+    decodeProtocol: protocolDecoder,
     trustProxy: true
   })
 
@@ -747,6 +751,7 @@ test('tcp proxied (protocol v2) clients have access to the ipAddress(v6)', funct
       }
       done(null, true)
     },
+    decodeProtocol: protocolDecoder,
     trustProxy: true
   })
 
@@ -786,6 +791,7 @@ test('websocket clients have access to the ipAddress from the socket (if no ip h
       }
       done(null, true)
     },
+    decodeProtocol: protocolDecoder,
     trustProxy: true
   })
 
@@ -822,6 +828,7 @@ test('websocket proxied clients have access to the ipAddress from x-real-ip head
       }
       done(null, true)
     },
+    decodeProtocol: protocolDecoder,
     trustProxy: true
   })
 
@@ -864,6 +871,7 @@ test('websocket proxied clients have access to the ipAddress from x-forwarded-fo
       }
       done(null, true)
     },
+    decodeProtocol: protocolDecoder,
     trustProxy: true
   })