ws-incoming.js

var http   = require('http'),
    https  = require('https'),
    common = require('../common');

function isPromise(x) { return !!x && typeof x.then === 'function'; }

async function runAsyncListenersSequentially(emitter, event, args, timeoutMs) {
  const listeners = emitter ? emitter.listeners(event) : [];
  for (const fn of listeners) {
    if (fn.length > args.length) {
      await new Promise((resolve, reject) => {
        let settled = false;
        let timer = null;
        const done = (err) => {
          if (timer) clearTimeout(timer);
          if (!settled) {
            settled = true;
            err ? reject(err) : resolve();
          }
        };
        try {
          if (timeoutMs) {
            timer = setTimeout(() => done(new Error(event + ' hook timeout')), timeoutMs);
          }
          const ret = fn.call(emitter, ...args, done);
          if (isPromise(ret)) {
            ret.then(() => done(), done);
          }
        } catch (e) {
          done(e);
        }
      });
    } else {
      const ret = fn.call(emitter, ...args);
      if (isPromise(ret)) {
        if (timeoutMs) {
          await Promise.race([
            ret,
            new Promise((_, rej) => setTimeout(() => rej(new Error(event + ' hook timeout')), timeoutMs))
          ]);
        } else {
          await ret;
        }
      }
    }
  }
}

/*!
 * Array of passes.
 *
 * A `pass` is just a function that is executed on `req, socket, options`
 * so that you can easily add new checks while still keeping the base
 * flexible.
 */

/*
 * Websockets Passes
 *
 */

module.exports = {
  /**
   * WebSocket requests must have the `GET` method and
   * the `upgrade:websocket` header
   *
   * @param {ClientRequest} Req Request object
   * @param {Socket} Websocket
   *
   * @api private
   */
  checkMethodAndHeader : function checkMethodAndHeader(req, socket) {
    if (req.method !== 'GET' || !req.headers.upgrade) {
      socket.destroy();
      return true;
    }

    if (req.headers.upgrade.toLowerCase() !== 'websocket') {
      socket.destroy();
      return true;
    }
  },

  /**
   * Sets `x-forwarded-*` headers if specified in config.
   *
   * @param {ClientRequest} Req Request object
   * @param {Socket} Websocket
   * @param {Object} Options Config object passed to the proxy
   *
   * @api private
   */
  XHeaders : function XHeaders(req, socket, options) {
    if(!options.xfwd) return;

    var values = {
      for  : req.connection.remoteAddress || req.socket.remoteAddress,
      port : common.getPort(req),
      proto: common.hasEncryptedConnection(req) ? 'wss' : 'ws'
    };

    ['for', 'port', 'proto'].forEach(function(header) {
      req.headers['x-forwarded-' + header] =
          (req.headers['x-forwarded-' + header] || '') +
          (req.headers['x-forwarded-' + header] ? ',' : '') +
          values[header];
    });
  },

  /**
   * Does the actual proxying. Make the request and upgrade it
   * send the Switching Protocols request and pipe the sockets.
   *
   * @param {ClientRequest} Req Request object
   * @param {Socket} Websocket
   * @param {Object} Options Config object passed to the proxy
   *
   * @api private
   */
  stream : function stream(req, socket, options, head, server, clb) {

    var createHttpHeader = function(line, headers) {
      return Object.keys(headers).reduce(function (head, key) {
        var value = headers[key];

        if (!Array.isArray(value)) {
          head.push(key + ': ' + value);
          return head;
        }

        for (var i = 0; i < value.length; i++) {
          head.push(key + ': ' + value[i]);
        }
        return head;
      }, [line])
          .join('\r\n') + '\r\n\r\n';
    };

    common.setupSocket(socket);

    if (head && head.length) socket.unshift(head);

    var proxyReq = (common.isSSL.test(options.target.protocol) ? https : http).request(
        common.setupOutgoing(options.ssl || {}, options, req)
    );

    // Error Handler
    proxyReq.on('error', onOutgoingError);
    proxyReq.on('response', function (res) {
      // if upgrade event isn't going to happen, close the socket
      if (!res.upgrade) {
        socket.write(createHttpHeader('HTTP/' + res.httpVersion + ' ' + res.statusCode + ' ' + res.statusMessage, res.headers));
        res.pipe(socket);
      }
    });

    proxyReq.on('upgrade', function(proxyRes, proxySocket, proxyHead) {
      proxySocket.on('error', onOutgoingError);

      // Allow us to listen when the websocket has completed
      proxySocket.on('end', function () {
        server && server.emit('close', proxyRes, proxySocket, proxyHead);
      });

      // The pipe below will end proxySocket if socket closes cleanly, but not
      // if it errors (eg, vanishes from the net and starts returning
      // EHOSTUNREACH). We need to do that explicitly.
      socket.on('error', function () {
        proxySocket.end();
      });

      common.setupSocket(proxySocket);

      if (proxyHead && proxyHead.length) proxySocket.unshift(proxyHead);

      //
      // Remark: Handle writing the headers to the socket when switching protocols
      // Also handles when a header is an array
      //
      socket.write(createHttpHeader('HTTP/1.1 101 Switching Protocols', proxyRes.headers));

      proxySocket.pipe(socket).pipe(proxySocket);

      server && server.emit('open', proxySocket);
      server && server.emit('proxySocket', proxySocket);  // DEPRECATED.
    });

    const hookTimeout = options && options.proxyReqWsTimeout; // optional in ms

    (async () => {
      try {
        if (server && typeof server.listenerCount === 'function' && server.listenerCount('proxyReqWs') > 0) {
          await runAsyncListenersSequentially(
              server,
              'proxyReqWs',
              [proxyReq, req, socket, options, head],
              hookTimeout
          );
        }
        proxyReq.end(); // nach Abschluss aller Hooks
      } catch (err) {
        onOutgoingError(err);
      }
    })();

    function onOutgoingError(err) {
      if (clb) {
        clb(err, req, socket);
      } else if (server && typeof server.emit === 'function') {
        server.emit('error', err, req, socket);
      }
      try { socket.end(); } catch (_) {}
    }
  }
};