File Explorer

1'use strict'2 3var Transform = require('readable-stream').Transform4var duplexify = require('duplexify')5var WS = require('isomorphic-ws')6var Buffer = require('safe-buffer').Buffer7 8module.exports = WebSocketStream9 10function buildProxy (options, socketWrite, socketEnd) {11  var proxy = new Transform({12    objectMode: options.objectMode13  })14 15  proxy._write = socketWrite16  proxy._flush = socketEnd17 18  return proxy19}20 21function WebSocketStream(target, protocols, options) {22  var stream, socket23 24  var isBrowser = process.title === 'browser'25  var isNative = !!global.WebSocket26  var socketWrite = isBrowser ? socketWriteBrowser : socketWriteNode27 28  if (protocols && !Array.isArray(protocols) && 'object' === typeof protocols) {29    // accept the "options" Object as the 2nd argument30    options = protocols31    protocols = undefined32 33    if (typeof options.protocol === 'string' || Array.isArray(options.protocol)) {34      protocols = options.protocol;35    }36  }37 38  if (!options) options = {}39 40  if (options.objectMode === undefined) {41    options.objectMode = !(options.binary === true || options.binary === undefined)42  }43 44  var proxy = buildProxy(options, socketWrite, socketEnd)45 46  if (!options.objectMode) {47    proxy._writev = writev48  }49 50  // browser only: sets the maximum socket buffer size before throttling51  var bufferSize = options.browserBufferSize || 1024 * 51252 53  // browser only: how long to wait when throttling54  var bufferTimeout = options.browserBufferTimeout || 100055 56  // use existing WebSocket object that was passed in57  if (typeof target === 'object') {58    socket = target59  // otherwise make a new one60  } else {61    // special constructor treatment for native websockets in browsers, see62    // https://github.com/maxogden/websocket-stream/issues/8263    if (isNative && isBrowser) {64      socket = new WS(target, protocols)65    } else {66      socket = new WS(target, protocols, options)67    }68 69    socket.binaryType = 'arraybuffer'70  }71 72  // according to https://github.com/baygeldin/ws-streamify/issues/173  // Nodejs WebSocketServer cause memory leak74  // Handlers like onerror, onclose, onmessage and onopen are accessible via setter/getter75  // And setter first of all fires removeAllListeners, that doesnt make inner array of clients on WebSocketServer cleared ever76  var eventListenerSupport = ('undefined' === typeof socket.addEventListener)77 78  // was already open when passed in79  if (socket.readyState === socket.OPEN) {80    stream = proxy81  } else {82    stream = duplexify(undefined, undefined, options)83    if (!options.objectMode) {84      stream._writev = writev85    }86 87    if (eventListenerSupport) {88       socket.addEventListener('open', onopen)89    } else {90       socket.onopen = onopen91    }92  }93 94  stream.socket = socket95 96  if (eventListenerSupport) {97     socket.addEventListener('close', onclose)98     socket.addEventListener('error', onerror)99     socket.addEventListener('message', onmessage)100  } else {101     socket.onclose = onclose102     socket.onerror = onerror103     socket.onmessage = onmessage104  }105 106  proxy.on('close', destroy)107 108  var coerceToBuffer = !options.objectMode109 110  function socketWriteNode(chunk, enc, next) {111    // avoid errors, this never happens unless112    // destroy() is called113    if (socket.readyState !== socket.OPEN) {114      next()115      return116    }117 118    if (coerceToBuffer && typeof chunk === 'string') {119      chunk = Buffer.from(chunk, 'utf8')120    }121    socket.send(chunk, next)122  }123 124  function socketWriteBrowser(chunk, enc, next) {125    if (socket.bufferedAmount > bufferSize) {126      setTimeout(socketWriteBrowser, bufferTimeout, chunk, enc, next)127      return128    }129 130    if (coerceToBuffer && typeof chunk === 'string') {131      chunk = Buffer.from(chunk, 'utf8')132    }133 134    try {135      socket.send(chunk)136    } catch(err) {137      return next(err)138    }139 140    next()141  }142 143  function socketEnd(done) {144    socket.close()145    done()146  }147 148  function onopen() {149    stream.setReadable(proxy)150    stream.setWritable(proxy)151    stream.emit('connect')152  }153 154  function onclose(closeDetails) {155    stream.emit('ws-close', closeDetails);156    stream.end()157    stream.destroy()158  }159 160  function onerror(err) {161    stream.destroy(err)162  }163 164  function onmessage(event) {165    var data = event.data166    if (data instanceof ArrayBuffer) data = Buffer.from(data)167    else data = Buffer.from(data, 'utf8')168    proxy.push(data)169  }170 171  function destroy() {172    socket.close()173  }174 175  // this is to be enabled only if objectMode is false176  function writev (chunks, cb) {177    var buffers = new Array(chunks.length)178    for (var i = 0; i < chunks.length; i++) {179      if (typeof chunks[i].chunk === 'string') {180        buffers[i] = Buffer.from(chunks[i], 'utf8')181      } else {182        buffers[i] = chunks[i].chunk183      }184    }185 186    this._write(Buffer.concat(buffers), 'binary', cb)187  }188 189  return stream190}191