Fixed support for node v0.10 and v0.12 with corkable stream.

package.json

@@ -32,6 +32,7 @@
     "dev-null": "^0.1.1",
     "faucet": "0.0.1",
     "pre-commit": "^1.1.1",
+    "readable-stream": "^2.0.2",
     "tape": "^4.2.0"
   },
   "dependencies": {

test.js

@@ -1,6 +1,7 @@
 
 var test    = require('tape')
   , mqtt    = require('./')
+  , WS      = require('readable-stream').Writable
 
 function testParseGenerate(name, object, buffer, opts) {
   test(name + ' parse', function(t) {
@@ -851,3 +852,30 @@ testGenerateError('Invalid password', {
   , username: 'username'
   , password: 42
 })
+
+test('support cork', function (t) {
+  t.plan(9)
+
+  var dest = WS()
+    , count = 0
+
+  dest._write = function (chunk, enc, cb) {
+    t.pass('_write called')
+    cb()
+  }
+
+  mqtt.writeToStream({
+      cmd: 'connect'
+    , retain: false
+    , qos: 0
+    , dup: false
+    , length: 18
+    , protocolId: 'MQIsdp'
+    , protocolVersion: 3
+    , clean: false
+    , keepalive: 30
+    , clientId: 'test'
+  }, dest)
+
+  dest.end()
+})

writeToStream.js

@@ -5,11 +5,22 @@ var protocol = require('./constants')
   , empty = new Buffer(0)
   , zeroBuf = new Buffer([0])
   , numCache = require('./numbers')
+  , nextTick = process.nextTick
+
+if (process.version.indexOf('v0.1') === 0) {
+  (function () {
+    nextTick = function tickShim (func, stream) {
+      return function () {
+        return func(stream)
+      }
+    }
+  })()
+}
 
 function generate(packet, stream) {
   if (stream.cork) {
     stream.cork()
-    process.nextTick(uncork, stream)
+    nextTick(uncork, stream)
   }
 
   switch (packet.cmd) {