spb/admin-ka

1/* eslint no-unused-vars: ["error", { "varsIgnorePattern": "^WebSocket$" }] */2'use strict';34const WebSocket = require('./websocket');5const { Duplex } = require('stream');67/**8 * Emits the `'close'` event on a stream.9 *10 * @param {Duplex} stream The stream.11 * @private12 */13function emitClose(stream) {14  stream.emit('close');15}1617/**18 * The listener of the `'end'` event.19 *20 * @private21 */22function duplexOnEnd() {23  if (!this.destroyed && this._writableState.finished) {24    this.destroy();25  }26}2728/**29 * The listener of the `'error'` event.30 *31 * @param {Error} err The error32 * @private33 */34function duplexOnError(err) {35  this.removeListener('error', duplexOnError);36  this.destroy();37  if (this.listenerCount('error') === 0) {38    // Do not suppress the throwing behavior.39    this.emit('error', err);40  }41}4243/**44 * Wraps a `WebSocket` in a duplex stream.45 *46 * @param {WebSocket} ws The `WebSocket` to wrap47 * @param {Object} [options] The options for the `Duplex` constructor48 * @return {Duplex} The duplex stream49 * @public50 */51function createWebSocketStream(ws, options) {52  let terminateOnDestroy = true;5354  const duplex = new Duplex({55    ...options,56    autoDestroy: false,57    emitClose: false,58    objectMode: false,59    writableObjectMode: false60  });6162  ws.on('message', function message(msg, isBinary) {63    const data =64      !isBinary && duplex._readableState.objectMode ? msg.toString() : msg;6566    if (!duplex.push(data)) ws.pause();67  });6869  ws.once('error', function error(err) {70    if (duplex.destroyed) return;7172    // Prevent `ws.terminate()` from being called by `duplex._destroy()`.73    //74    // - If the `'error'` event is emitted before the `'open'` event, then75    //   `ws.terminate()` is a noop as no socket is assigned.76    // - Otherwise, the error is re-emitted by the listener of the `'error'`77    //   event of the `Receiver` object. The listener already closes the78    //   connection by calling `ws.close()`. This allows a close frame to be79    //   sent to the other peer. If `ws.terminate()` is called right after this,80    //   then the close frame might not be sent.81    terminateOnDestroy = false;82    duplex.destroy(err);83  });8485  ws.once('close', function close() {86    if (duplex.destroyed) return;8788    duplex.push(null);89  });9091  duplex._destroy = function (err, callback) {92    if (ws.readyState === ws.CLOSED) {93      callback(err);94      process.nextTick(emitClose, duplex);95      return;96    }9798    let called = false;99100    ws.once('error', function error(err) {101      called = true;102      callback(err);103    });104105    ws.once('close', function close() {106      if (!called) callback(err);107      process.nextTick(emitClose, duplex);108    });109110    if (terminateOnDestroy) ws.terminate();111  };112113  duplex._final = function (callback) {114    if (ws.readyState === ws.CONNECTING) {115      ws.once('open', function open() {116        duplex._final(callback);117      });118      return;119    }120121    // If the value of the `_socket` property is `null` it means that `ws` is a122    // client websocket and the handshake failed. In fact, when this happens, a123    // socket is never assigned to the websocket. Wait for the `'error'` event124    // that will be emitted by the websocket.125    if (ws._socket === null) return;126127    if (ws._socket._writableState.finished) {128      callback();129      if (duplex._readableState.endEmitted) duplex.destroy();130    } else {131      ws._socket.once('finish', function finish() {132        // `duplex` is not destroyed here because the `'end'` event will be133        // emitted on `duplex` after this `'finish'` event. The EOF signaling134        // `null` chunk is, in fact, pushed when the websocket emits `'close'`.135        callback();136      });137      ws.close();138    }139  };140141  duplex._read = function () {142    if (ws.isPaused) ws.resume();143  };144145  duplex._write = function (chunk, encoding, callback) {146    if (ws.readyState === ws.CONNECTING) {147      ws.once('open', function open() {148        duplex._write(chunk, encoding, callback);149      });150      return;151    }152153    ws.send(chunk, callback);154  };155156  duplex.on('end', duplexOnEnd);157  duplex.on('error', duplexOnError);158  return duplex;159}160161module.exports = createWebSocketStream;162