WIP: mqtt stream parser

aedes.d.ts

@@ -69,7 +69,6 @@ declare namespace aedes {
     authorizeSubscribe?: AuthorizeSubscribeHandler
     authorizeForward?: AuthorizeForwardHandler
     published?: PublishedHandler
-    queueLimit?: number
     maxClientsIdLength?: number
   }
   interface Client extends EventEmitter {

aedes.js

@@ -28,7 +28,6 @@ const defaultOptions = {
   published: defaultPublished,
   trustProxy: false,
   trustedProxies: [],
-  queueLimit: 42,
   maxClientsIdLength: 23
 }
 
@@ -45,7 +44,6 @@ function Aedes (opts) {
   // +1 when construct a new aedes-packet
   // internal track for last brokerCounter
   this.counter = 0
-  this.queueLimit = opts.queueLimit
   this.connectTimeout = opts.connectTimeout
   this.maxClientsIdLength = opts.maxClientsIdLength
   this.mq = opts.mq || mqemitter({

docs/Aedes.md

@@ -38,7 +38,6 @@
   - `mq` [`<MQEmitter>`](../README.md#mqemitter) middleware used to deliver messages to subscribed clients. In a cluster environment it is used also to share messages between brokers instances. __Default__: `mqemitter`
   - `concurrency` `<number>` maximum number of concurrent messages delivered by `mq`. __Default__: `100`
   - `persistence` [`<Persistence>`](../README.md#persistence) middleware that stores _QoS > 0, retained, will_ packets and _subscriptions_. __Default__: `aedes-persistence` (_in memory_)
-  - `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`
   - `id` `<string>` aedes broker unique identifier. __Default__: `uuidv4()`

lib/client.js

@@ -1,17 +1,16 @@
 'use strict'
 
-const mqtt = require('mqtt-packet')
 const EventEmitter = require('events')
 const util = require('util')
-const eos = require('end-of-stream')
 const Packet = require('aedes-packet')
 const write = require('./write')
 const QoSPacket = require('./qos-packet')
 const handleSubscribe = require('./handlers/subscribe')
 const handleUnsubscribe = require('./handlers/unsubscribe')
 const handle = require('./handlers')
-const { pipeline } = require('readable-stream')
+const { Writable, finished, pipeline } = require('readable-stream')
 const { through } = require('./utils')
+const MqttStreamParser = require('./mqtt-stream-parser')
 
 module.exports = Client
 
@@ -39,7 +38,7 @@ function Client (broker, conn, req) {
 
   this._disconnected = false
   this._authorized = false
-  this._parsingBatch = 1
+  // for QoS > 0
   this._nextId = Math.ceil(Math.random() * 65535)
 
   this.req = req
@@ -51,50 +50,31 @@ function Client (broker, conn, req) {
   this.will = null
   this._will = null
 
-  this._parser = mqtt.parser()
-  this._parser.client = this
-  this._parser._queue = [] // queue packets received before client fires 'connect' event. Prevents memory leaks on 'connect' event
-  this._parser.on('packet', enqueue)
-  this.once('connected', dequeue)
+  const mqttStream = new Writable({
+    objectMode: true,
+    emitClose: false,
+    encoding: 'utf8',
+    write: handlePacket
+  })
 
-  function nextBatch (err) {
-    if (err) {
-      that.emit('error', err)
-      return
-    }
+  function handlePacket (packet, enc, done) {
+    handle(that, packet, done)
+  }
 
-    const client = that
+  const emitError = this.emit.bind(this, 'error')
 
-    if (client._paused) {
-      return
-    }
+  mqttStream.on('error', emitError)
 
-    that._parsingBatch--
-    if (that._parsingBatch <= 0) {
-      that._parsingBatch = 0
-      var buf = client.conn.read(null)
-      if (!client.connackSent && client.broker.decodeProtocol && client.broker.trustProxy && buf) {
-        const { data } = client.broker.decodeProtocol(client, buf)
-        if (data) {
-          client._parser.parse(data)
-        } else {
-          client._parser.parse(buf)
-        }
-      } else if (buf) {
-        client._parser.parse(buf)
-      }
-    }
-  }
-  this._nextBatch = nextBatch
+  this._mqttStream = mqttStream
+  this._mqttParserStream = new MqttStreamParser(this)
 
-  conn.on('readable', nextBatch)
+  conn.pipe(this._mqttParserStream).pipe(mqttStream)
 
   this.on('error', onError)
-  conn.on('error', this.emit.bind(this, 'error'))
-  this._parser.on('error', this.emit.bind(this, 'error'))
+  conn.on('error', emitError)
 
   conn.on('end', this.close.bind(this))
-  this._eos = eos(this.conn, this.close.bind(this))
+  this._eos = finished(this.conn, this.close.bind(this))
 
   this.deliver0 = function deliverQoS0 (_packet, cb) {
     const toForward = dedupe(that, _packet) &&
@@ -261,10 +241,9 @@ Client.prototype.close = function (done) {
 
   this.closed = true
 
-  this._parser.removeAllListeners('packet')
-  conn.removeAllListeners('readable')
-
-  this._parser._queue = null
+  conn.unpipe(this._mqttParserStream)
+  this._mqttParserStream.end()
+  this._mqttStream.end()
 
   if (this._keepaliveTimer) {
     this._keepaliveTimer.clear()
@@ -334,38 +313,11 @@ Client.prototype.close = function (done) {
 }
 
 Client.prototype.pause = function () {
-  this._paused = true
+  this.conn.pause()
 }
 
 Client.prototype.resume = function () {
-  this._paused = false
-  this._nextBatch()
-}
-
-function enqueue (packet) {
-  const client = this.client
-  client._parsingBatch++
-  // already connected or it's the first packet
-  if (client.connackSent || client._parsingBatch === 1) {
-    handle(client, packet, client._nextBatch)
-  } else {
-    if (this._queue.length < client.broker.queueLimit) {
-      this._queue.push(packet)
-    } else {
-      this.emit('error', new Error('Client queue limit reached'))
-    }
-  }
-}
-
-function dequeue () {
-  const q = this._parser._queue
-  if (q) {
-    for (var i = 0, len = q.length; i < len; i++) {
-      handle(this, q[i], this._nextBatch)
-    }
-
-    this._parser._queue = null
-  }
+  this.conn.resume()
 }
 
 Client.prototype.emptyOutgoingQueue = function (done) {

lib/handlers/connect.js

@@ -43,6 +43,7 @@ function handleConnect (client, packet, done) {
   clearTimeout(client._connectTimer)
   client._connectTimer = null
   client.connecting = true
+  client.pause()
   client.broker.preConnect(client, negate)
 
   function negate (err, successful) {
@@ -97,7 +98,6 @@ function init (client, packet, done) {
 
 function authenticate (arg, done) {
   const client = this.client
-  client.pause()
   client.broker.authenticate(
     client,
     this.packet.username,

lib/mqtt-stream-parser.js

@@ -0,0 +1,27 @@
+'use strict'
+const mqtt = require('mqtt-packet')
+const { through } = require('./utils')
+
+module.exports = MqttStreamParser
+
+function MqttStreamParser (client, options) {
+  if (!options) options = {}
+
+  const stream = through(process)
+  const parser = mqtt.parser(options)
+
+  parser.on('packet', push)
+  parser.on('error', (err) => client.emit('error', err))
+  this._parser = parser
+
+  stream.on('error', stream.end.bind(stream))
+
+  function process (chunk, enc, cb) {
+    parser.parse(chunk)
+    cb()
+  }
+  function push (packet) {
+    stream.push(packet)
+  }
+  return stream
+}

test/basic.js

@@ -374,19 +374,22 @@ test('disconnect', function (t) {
   })
 })
 
-test('disconnect client on wrong cmd', function (t) {
+test('disconnect client on unknown cmd', function (t) {
   t.plan(1)
 
-  const s = noError(connect(setup()), t)
-  t.tearDown(s.broker.close.bind(s.broker))
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
 
-  s.broker.on('clientDisconnect', function () {
-    t.pass('closed stream')
+  broker.on('clientDisconnect', function (client) {
+    t.equal(client.id, 'abcde', 'client matches')
   })
 
-  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' })
+  noError(connect(setup(broker), { clientId: 'abcde' }), t)
+
+  broker.on('client', function (client) {
+    client._mqttStream.write({
+      cmd: 'UNKNOWN'
+    })
   })
 })
 

test/client-pub-sub.js

@@ -139,13 +139,13 @@ test('publish QoS 2 throws error in pubrel', function (t) {
 
   s.outStream.on('data', function (packet) {
     if (packet.cmd === 'publish') {
+      s.broker.persistence.outgoingUpdate = function (client, pubrel, cb) {
+        cb(new Error('error'))
+      }
       s.inStream.write({
         cmd: 'pubrec',
         messageId: packet.messageId
       })
-      s.broker.persistence.outgoingUpdate = function (client, pubrel, cb) {
-        cb(new Error('error'))
-      }
     }
   })
 

test/close_socket_by_other_party.js

@@ -137,9 +137,6 @@ test('multiple clients subscribe same topic, and all clients still receive messa
   const server = require('net').createServer(broker.handle)
   const port = 1883
   server.listen(port)
-  broker.on('clientError', function (client, err) {
-    t.error(err)
-  })
 
   var client1, client2
   const _sameTopic = 'hello'
@@ -152,7 +149,7 @@ test('multiple clients subscribe same topic, and all clients still receive messa
 
   client1.subscribe(_sameTopic, { qos: 0, retain: false }, () => {
     t.pass('client1 sub callback')
-    // stimulate closed socket by users
+    // simulate closed socket by users
     client1.stream.destroy()
 
     // client 2
@@ -166,6 +163,7 @@ test('multiple clients subscribe same topic, and all clients still receive messa
 
       // pubClient
       const pubClient = mqtt.connect('mqtt://localhost', { clientId: 'pubClient' })
+
       pubClient.publish(_sameTopic, 'world', { qos: 0, retain: false }, () => {
         t.pass('pubClient publish event')
         pubClient.end()