Implemented writev.

Author: mcollina

package.json

@@ -25,7 +25,7 @@
   "dependencies": {
     "duplexify": "^3.2.0",
     "inherits": "^2.0.1",
-    "through2": "^2.0.0",
+    "readable-stream": "^2.2.0",
     "ws": "^2.2.3",
     "xtend": "^4.0.0"
   },

stream.js

@@ -1,18 +1,41 @@
 'use strict'
 
-var through = require('through2')
+var Transform = require('readable-stream').Transform
 var duplexify = require('duplexify')
 var WS = require('ws')
 
 module.exports = WebSocketStream
 
+function buildProxy (options, socketWrite, socketEnd) {
+  var proxy = new Transform({
+    objectMode: options.objectMode
+  })
+
+  proxy._destroyed = false
+  proxy._write = socketWrite
+  proxy._flush = socketEnd
+
+  proxy.destroy = function(err) {
+    if (this._destroyed) return
+      this._destroyed = true
+
+    var self = this
+    process.nextTick(function() {
+      if (err)
+        self.emit('error', err)
+      self.emit('close')
+    })
+  }
+
+  return proxy
+}
+
 function WebSocketStream(target, protocols, options) {
   var stream, socket
 
   var isBrowser = process.title === 'browser'
   var isNative = !!global.WebSocket
   var socketWrite = isBrowser ? socketWriteBrowser : socketWriteNode
-  var proxy = through.obj(socketWrite, socketEnd)
 
   if (protocols && !Array.isArray(protocols) && 'object' === typeof protocols) {
     // accept the "options" Object as the 2nd argument
@@ -26,6 +49,16 @@ function WebSocketStream(target, protocols, options) {
 
   if (!options) options = {}
 
+  if (options.objectMode === undefined) {
+    options.objectMode = !(options.binary === true || options.binary === undefined)
+  }
+
+  var proxy = buildProxy(options, socketWrite, socketEnd)
+
+  if (!options.objectMode) {
+    proxy._writev = writev
+  }
+
   // browser only: sets the maximum socket buffer size before throttling
   var bufferSize = options.browserBufferSize || 1024 * 512
 
@@ -64,7 +97,7 @@ function WebSocketStream(target, protocols, options) {
 
   proxy.on('close', destroy)
 
-  var coerceToBuffer = options.binary || options.binary === undefined
+  var coerceToBuffer = !options.objectMode
 
   function socketWriteNode(chunk, enc, next) {
     // avoid errors, this never happens unless
@@ -130,5 +163,19 @@ function WebSocketStream(target, protocols, options) {
     socket.close()
   }
 
+  // this is to be enabled only if objectMode is false
+  function writev (chunks, cb) {
+    var buffers = new Array(chunks.length)
+    for (var i = 0; i < chunks.length; i++) {
+      if (typeof chunks[i].chunk === 'string') {
+        buffers[i] = new Buffer(chunks[i], 'utf8') // TODO use safe-buffer
+      } else {
+        buffers[i] = chunks[i].chunk
+      }
+    }
+
+    this._write(Buffer.concat(buffers), 'binary', cb)
+  }
+
   return stream
 }

test.js

@@ -8,7 +8,7 @@ var concat = require('concat-stream')
 test('echo server', function(t) {
 
   echo.start(function() {
-    var client = websocket(echo.url, echo.options)
+    var client = websocket(echo.url)
 
     client.on('error', console.error)
 
@@ -29,7 +29,7 @@ test('echo server', function(t) {
 test('emitting not connected errors', function(t) {
 
   echo.start(function() {
-    var client = websocket(echo.url, echo.options)
+    var client = websocket(echo.url)
 
     client.on('error', function() {
       echo.stop(function() {
@@ -225,3 +225,79 @@ test('stream handlers should fire once per connection', function(t) {
     w.end('pizza cats\n')
   })
 })
+
+test('client with writev', function(t) {
+  var server = http.createServer()
+
+  var str = ''
+  var wss = websocket.createServer({
+    server: server
+  }, function (stream) {
+    stream.once('data', function(data) {
+      t.ok(Buffer.isBuffer(data), 'is a buffer')
+      t.equal(data.toString(), 'hello world')
+
+      stream.once('data', function(data) {
+        t.ok(Buffer.isBuffer(data), 'is a buffer')
+        t.equal(data.toString(), str)
+        stream.end()
+        server.close()
+        t.end()
+      })
+    })
+  })
+
+  server.listen(8352, function () {
+    var client = websocket('ws://localhost:8352', {
+      objectMode: false
+    })
+
+    client.on('error', console.error)
+
+    client.once('connect', function () {
+      client.cork()
+      do {
+        str += 'foobar'
+      } while (client.write('foobar'))
+      client.uncork()
+    })
+
+    client.write('hello world')
+  })
+})
+
+test('server with writev', function(t) {
+  var server = http.createServer()
+
+  var str = ''
+  var wss = websocket.createServer({
+    server: server,
+    objectMode: false
+  }, function (stream) {
+    stream.cork()
+    do {
+      str += 'foobar'
+    } while (stream.write('foobar'))
+    stream.uncork()
+  })
+
+  server.listen(8352, function () {
+    var client = websocket('ws://localhost:8352')
+
+    client.on('error', console.error)
+
+    client.once('data', function(data) {
+      t.ok(Buffer.isBuffer(data), 'is a buffer')
+      t.equal(data.toString(), str)
+      client.end()
+      server.close()
+      t.end()
+    })
+  })
+})
+
+test('stop echo', function(t) {
+  echo.stop(function() {
+    t.end()
+  })
+})