Merge pull request #29 from kirilknysh/feature/dynamic-num-cache

Add option to enable/disable numbers cache

README.md

@@ -109,6 +109,9 @@ The object must be one of the ones specified by the [packets](#packets)
 section. Emits an `Error` on the stream if a packet cannot be generated.
 On node >= 0.12, this function automatically calls `cork()` on your stream,
 and then it calls `uncork()` on the next tick.
+By default cache for number buffers is enabled.
+It creates a list of buffers for faster write. To disable cache set `mqtt.writeToStream.cacheNumbers = false`.
+Should be set before any `writeToStream` calls.
 
 <a name="parser">
 

benchmarks/generate.js

@@ -1,10 +1,19 @@
+'use strict'
 
 var mqtt = require('../')
 var max = 100000
 var i
+var buf = Buffer.from('test')
+
+// initialize it
+mqtt.generate({
+  cmd: 'publish',
+  topic: 'test',
+  payload: buf
+})
+
 var start = Date.now()
 var time
-var buf = Buffer.from('test')
 
 for (i = 0; i < max; i++) {
   mqtt.generate({

numbers.js

@@ -3,13 +3,23 @@
 var Buffer = require('safe-buffer').Buffer
 var max = 65536
 var cache = {}
-var buffer
 
-for (var i = 0; i < max; i++) {
-  buffer = Buffer.allocUnsafe(2)
+function generateBuffer (i) {
+  var buffer = Buffer.allocUnsafe(2)
   buffer.writeUInt8(i >> 8, 0, true)
   buffer.writeUInt8(i & 0x00FF, 0 + 1, true)
-  cache[i] = buffer
+
+  return buffer
+}
+
+function generateCache () {
+  for (var i = 0; i < max; i++) {
+    cache[i] = generateBuffer(i)
+  }
 }
 
-module.exports = cache
+module.exports = {
+  cache: cache,
+  generateCache: generateCache,
+  generateNumber: generateBuffer
+}

test.js

@@ -116,6 +116,40 @@ function testWriteToStreamError (expected, fixture) {
   })
 }
 
+test('disabled numbers cache', function (t) {
+  var stream = WS()
+  var message = {
+    cmd: 'publish',
+    retain: false,
+    qos: 0,
+    dup: false,
+    length: 10,
+    topic: Buffer.from('test'),
+    payload: Buffer.from('test')
+  }
+  var expected = Buffer.from([
+    48, 10, // Header
+    0, 4, // Topic length
+    116, 101, 115, 116, // Topic (test)
+    116, 101, 115, 116 // Payload (test)
+  ])
+  var written = Buffer.alloc(0)
+
+  stream.write = (chunk) => {
+    written = Buffer.concat([written, chunk])
+  }
+  mqtt.writeToStream.cacheNumbers = false
+
+  mqtt.writeToStream(message, stream)
+
+  t.deepEqual(written, expected, 'written buffer is expected')
+
+  mqtt.writeToStream.cacheNumbers = true
+
+  stream.end()
+  t.end()
+})
+
 testParseGenerate('minimal connect', {
   cmd: 'connect',
   retain: false,

writeToStream.js

@@ -4,15 +4,26 @@ var protocol = require('./constants')
 var Buffer = require('safe-buffer').Buffer
 var empty = Buffer.allocUnsafe(0)
 var zeroBuf = Buffer.from([0])
-var numCache = require('./numbers')
+var numbers = require('./numbers')
 var nextTick = require('process-nextick-args')
 
+var numCache = numbers.cache
+var generateNumber = numbers.generateNumber
+var generateCache = numbers.generateCache
+var writeNumber = writeNumberCached
+var toGenerate = true
+
 function generate (packet, stream) {
   if (stream.cork) {
     stream.cork()
     nextTick(uncork, stream)
   }
 
+  if (toGenerate) {
+    toGenerate = false
+    generateCache()
+  }
+
   switch (packet.cmd) {
     case 'connect':
       return connect(packet, stream)
@@ -41,6 +52,24 @@ function generate (packet, stream) {
       return false
   }
 }
+/**
+ * Controls numbers cache.
+ * Set to "false" to allocate buffers on-the-flight instead of pre-generated cache
+ */
+Object.defineProperty(generate, 'cacheNumbers', {
+  get: function () {
+    return writeNumber === writeNumberCached
+  },
+  set: function (value) {
+    if (value) {
+      if (!numCache || Object.keys(numCache).length === 0) toGenerate = true
+      writeNumber = writeNumberCached
+    } else {
+      toGenerate = false
+      writeNumber = writeNumberGenerated
+    }
+  }
+})
 
 function uncork (stream) {
   stream.uncork()
@@ -519,9 +548,12 @@ function writeString (stream, string) {
  *
  * @api private
  */
-function writeNumber (stream, number) {
+function writeNumberCached (stream, number) {
   return stream.write(numCache[number])
 }
+function writeNumberGenerated (stream, number) {
+  return stream.write(generateNumber(number))
+}
 
 /**
  * writeStringOrBuffer - write a String or Buffer with the its length prefix