node-http-proxy/lib/node-http-proxy/http-proxy.js
function onUpgrade (reverseProxy, proxySocket) {
    if (!reverseProxy) {
      proxySocket.end();
      socket.end();
      return;
    }

    //
    // Any incoming data on this WebSocket to the proxy target
    // will be written to the `reverseProxy` socket.
    //
    proxySocket.on('data', listeners.onIncoming = function (data) {
      if (reverseProxy.incoming.socket.writable) {
        try {
          self.emit('websocket:outgoing', req, socket, head, data);
          var flushed = reverseProxy.incoming.socket.write(data);
          if (!flushed) {
            proxySocket.pause();
            reverseProxy.incoming.socket.once('drain', function () {
              try { proxySocket.resume() }
              catch (er) { console.error("proxySocket.resume error: %s", er.message) }
            });

            //
            // Force the `drain` event in 100ms if it hasn't
            // happened on its own.
            //
            setTimeout(function () {
              reverseProxy.incoming.socket.emit('drain');
            }, 100);
          }
        }
        catch (ex) {
          detach();
        }
      }
    });

    //
    // Any outgoing data on this Websocket from the proxy target
    // will be written to the `proxySocket` socket.
    //
    reverseProxy.incoming.socket.on('data', listeners.onOutgoing = function (data) {
      try {
        self.emit('websocket:incoming', reverseProxy, reverseProxy.incoming, head, data);
        var flushed = proxySocket.write(data);
        if (!flushed) {
          reverseProxy.incoming.socket.pause();
          proxySocket.once('drain', function () {
            try { reverseProxy.incoming.socket.resume() }
            catch (er) { console.error("reverseProxy.incoming.socket.resume error: %s", er.message) }
          });

          //
          // Force the `drain` event in 100ms if it hasn't
          // happened on its own.
          //
          setTimeout(function () {
            proxySocket.emit('drain');
          }, 100);
        }
      }
      catch (ex) {
        detach();
      }
    });

    //
    // Helper function to detach all event listeners
    // from `reverseProxy` and `proxySocket`.
    //
    function detach() {
      proxySocket.destroySoon();
      proxySocket.removeListener('end', listeners.onIncomingClose);
      proxySocket.removeListener('data', listeners.onIncoming);
      reverseProxy.incoming.socket.destroySoon();
      reverseProxy.incoming.socket.removeListener('end', listeners.onOutgoingClose);
      reverseProxy.incoming.socket.removeListener('data', listeners.onOutgoing);
    }

    //
    // If the incoming `proxySocket` socket closes, then
    // detach all event listeners.
    //
    proxySocket.on('end', listeners.onIncomingClose = function() {
      detach();

      // Emit the `end` event now that we have completed proxying
      self.emit('websocket:end', req, socket, head);
    });

    //
    // If the `reverseProxy` socket closes, then detach all
    // event listeners.
    //
    reverseProxy.incoming.socket.on('end', listeners.onOutgoingClose = function() {
      detach();
    });
  }
function onUpgrade (reverseProxy, proxySocket) {
    if (!reverseProxy) {
      proxySocket.end();
      socket.end();
      return;
    }

    //
    // Any incoming data on this WebSocket to the proxy target
    // will be written to the `reverseProxy` socket.
    //
    proxySocket.on('data', listeners.onIncoming = function (data) {
      if (reverseProxy.incoming.socket.writable) {
        try {
          self.emit('websocket:outgoing', req, socket, head, data);
          var flushed = reverseProxy.incoming.socket.write(data);
          if (!flushed) {
            proxySocket.pause();
            reverseProxy.incoming.socket.once('drain', function () {
              try { proxySocket.resume() }
              catch (er) { console.error("proxySocket.resume error: %s", er.message) }
            });

            //
            // Force the `drain` event in 100ms if it hasn't
            // happened on its own.
            //
            setTimeout(function () {
              reverseProxy.incoming.socket.emit('drain');
            }, 100);
          }
        }
        catch (ex) {
          detach();
        }
      }
    });

    //
    // Any outgoing data on this Websocket from the proxy target
    // will be written to the `proxySocket` socket.
    //
    reverseProxy.incoming.socket.on('data', listeners.onOutgoing = function (data) {
      try {
        self.emit('websocket:incoming', reverseProxy, reverseProxy.incoming, head, data);
        var flushed = proxySocket.write(data);
        if (!flushed) {
          reverseProxy.incoming.socket.pause();
          proxySocket.once('drain', function () {
            try { reverseProxy.incoming.socket.resume() }
            catch (er) { console.error("reverseProxy.incoming.socket.resume error: %s", er.message) }
          });

          //
          // Force the `drain` event in 100ms if it hasn't
          // happened on its own.
          //
          setTimeout(function () {
            proxySocket.emit('drain');
          }, 100);
        }
      }
      catch (ex) {
        detach();
      }
    });

    //
    // Helper function to detach all event listeners
    // from `reverseProxy` and `proxySocket`.
    //
    function detach() {
      proxySocket.destroySoon();
      proxySocket.removeListener('end', listeners.onIncomingClose);
      proxySocket.removeListener('data', listeners.onIncoming);
      reverseProxy.incoming.socket.destroySoon();
      reverseProxy.incoming.socket.removeListener('end', listeners.onOutgoingClose);
      reverseProxy.incoming.socket.removeListener('data', listeners.onOutgoing);
    }

    //
    // If the incoming `proxySocket` socket closes, then
    // detach all event listeners.
    //
    proxySocket.on('end', listeners.onIncomingClose = function() {
      detach();

      // Emit the `end` event now that we have completed proxying
      self.emit('websocket:end', req, socket, head);
    });

    //
    // If the `reverseProxy` socket closes, then detach all
    // event listeners.
    //
    reverseProxy.incoming.socket.on('end', listeners.onOutgoingClose = function() {
      detach();
    });
  }
function onUpgrade ( reverseProxy , proxySocket ) { if ( ! reverseProxy ) { proxySocket . end ( ) ; socket . end ( ) ; return ; } // // Any incoming data on this WebSocket to the proxy target // will be written to the `reverseProxy` socket. // proxySocket . on ( 'data' , listeners . onIncoming = function ( data ) { if ( reverseProxy . incoming . socket . writable ) { try { self . emit ( 'websocket:outgoing' , req , socket , head , data ) ; var flushed = reverseProxy . incoming . socket . write ( data ) ; if ( ! flushed ) { proxySocket . pause ( ) ; reverseProxy . incoming . socket . once ( 'drain' , function ( ) { try { proxySocket . resume ( ) } catch ( er ) { console . error ( "proxySocket.resume error: %s" , er . message ) } } ) ; // // Force the `drain` event in 100ms if it hasn't // happened on its own. // setTimeout ( function ( ) { reverseProxy . incoming . socket . emit ( 'drain' ) ; } , 100 ) ; } } catch ( ex ) { detach ( ) ; } } } ) ; // // Any outgoing data on this Websocket from the proxy target // will be written to the `proxySocket` socket. // reverseProxy . incoming . socket . on ( 'data' , listeners . onOutgoing = function ( data ) { try { self . emit ( 'websocket:incoming' , reverseProxy , reverseProxy . incoming , head , data ) ; var flushed = proxySocket . write ( data ) ; if ( ! flushed ) { reverseProxy . incoming . socket . pause ( ) ; proxySocket . once ( 'drain' , function ( ) { try { reverseProxy . incoming . socket . resume ( ) } catch ( er ) { console . error ( "reverseProxy.incoming.socket.resume error: %s" , er . message ) } } ) ; // // Force the `drain` event in 100ms if it hasn't // happened on its own. // setTimeout ( function ( ) { proxySocket . emit ( 'drain' ) ; } , 100 ) ; } } catch ( ex ) { detach ( ) ; } } ) ; // // Helper function to detach all event listeners // from `reverseProxy` and `proxySocket`. // function detach ( ) { proxySocket . destroySoon ( ) ; proxySocket . removeListener ( 'end' , listeners . onIncomingClose ) ; proxySocket . removeListener ( 'data' , listeners . onIncoming ) ; reverseProxy . incoming . socket . destroySoon ( ) ; reverseProxy . incoming . socket . removeListener ( 'end' , listeners . onOutgoingClose ) ; reverseProxy . incoming . socket . removeListener ( 'data' , listeners . onOutgoing ) ; } // // If the incoming `proxySocket` socket closes, then // detach all event listeners. // proxySocket . on ( 'end' , listeners . onIncomingClose = function ( ) { detach ( ) ; // Emit the `end` event now that we have completed proxying self . emit ( 'websocket:end' , req , socket , head ) ; } ) ; // // If the `reverseProxy` socket closes, then detach all // event listeners. // reverseProxy . incoming . socket . on ( 'end' , listeners . onOutgoingClose = function ( ) { detach ( ) ; } ) ; }
On `upgrade` from the Agent socket, listen to the appropriate events.
On upgrade from the Agent socket listen to the appropriate events .