Backport http-party/node-http-proxy#1362

.travis.yml

@@ -1,16 +1,5 @@
-sudo: false
 language: node_js
 node_js:
-  - "4"
-  - "6"
   - "8"
 script:
   - npm test
-after_success:
-  - bash <(curl -s https://codecov.io/bash)
-matrix:
-  fast_finish: true
-notifications:
-  email:
-    - travis@nodejitsu.com
-  irc: "irc.freenode.org#nodejitsu"

README.md

@@ -395,6 +395,27 @@ proxyServer.listen(8015);
 
     };
     ```
+*  **wsInterceptClientMsg**: Is a handler which is called when a websocket message is intercepted on its way to the server from the client. It takes two arguments: `data` - is a websocket message and `options` in which exists field `req` - websocket request object and `flags` (fin, mask, compress, binary). If falsy value is returned then nothing will be sended to the client.
+     ```
+      const proxy = new HttpProxy({
+        ...
+        wsInterceptClientMsg: (data, options) {
+          return typeof data === 'string ? data.toUpperCase() : data;
+        }
+        ...
+      })
+    ```
+*  **wsInterceptServerMsg**: Is a handler which is called when a websocket message is intercepted on its way to the client from the server. It takes two arguments: `data` - is a websocket message and `options` in which exist fields `req` - websocket request object and `flags` (fin, mask, compress, binary). If falsy value is returned then nothing will be sended to the target server.
+    ```
+      const proxy = new HttpProxy({
+        ...
+        wsInterceptServerMsg: (data, options) {
+          return typeof data === 'string ? data.toUpperCase() : data;
+        }
+        ...
+      })
+    ```
+
 
 **NOTE:**
 `options.ws` and `options.ssl` are optional.
@@ -466,6 +487,39 @@ proxy.on('close', function (res, socket, head) {
 
 **[Back to top](#table-of-contents)**
 
+### Listening for websocket proxy request events
+
+* `clientSenderInited`: This event is emitted after websocket sender from client to server is initialized.
+* `serverSenderInited`: This event is emitted after websocket sender from server to client is initialized.
+* `wsClientMsg`: This event is emitted after webscoket message is sended from the client to the server.
+* `wsServerMsg`: This event is emitted after websocket message is sended from the server to the client.
+
+```js
+httpProxy.createServer({
+  target: 'ws://localhost:9014',
+  ws: true
+}).listen(8014);
+
+proxyServer.on('upgrade', function (req, socket, head) {
+  proxy.ws(req, socket, head);
+});
+
+proxyServer.on('proxyReqWs', (proxyReq, req, socket, options, head) => {
+  proxyReq.on('clientSenderInited', (sender) => {
+    sender.send('hello from client');
+  });
+
+  proxyReq.on('serverSenderInited', (sender) => {
+    sender.send('hello from server');
+  });
+
+  proxyReq.on('wsClientMsg', console.log);
+  proxyReq.on('wsServerMsg', console.log);
+});
+```
+
+**[Back to top](#table-of-contents)**
+
 ### Shutdown
 
 * When testing or running server within another program it may be necessary to close the proxy.

lib/http-proxy/passes/ws-incoming.js

@@ -1,6 +1,9 @@
-var http   = require('http'),
-    https  = require('https'),
-    common = require('../common');
+'use strict';
+
+const http = require('http');
+const https = require('https');
+const common = require('../common');
+const WsInterceptor = require('../ws/interceptor');
 
 /*!
  * Array of passes.
@@ -128,10 +131,14 @@ module.exports = {
       // 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 () {
+      socket.on('error', function (err) {
+        console.error('ERROR: socket closed with:', err);
         proxySocket.end();
       });
 
+      socket.on('close', () => proxySocket.end());
+      socket.on('end', () => socket.end());
+
       common.setupSocket(proxySocket);
 
       if (proxyHead && proxyHead.length) proxySocket.unshift(proxyHead);
@@ -142,7 +149,9 @@ module.exports = {
       //
       socket.write(createHttpHeader('HTTP/1.1 101 Switching Protocols', proxyRes.headers));
 
-      proxySocket.pipe(socket).pipe(proxySocket);
+      const wsInterceptor = WsInterceptor.create({socket, options, req, proxyReq, proxyRes, proxySocket});
+
+      wsInterceptor.startDataTransfer();
 
       server.emit('open', proxySocket);
       server.emit('proxySocket', proxySocket);  //DEPRECATED.

lib/http-proxy/ws/interceptor.js

@@ -0,0 +1,116 @@
+'use strict';
+
+const PerMessageDeflate = require('ws/lib/PerMessageDeflate');
+const Extensions = require('ws/lib/Extensions');
+const Receiver = require('ws/lib/Receiver');
+const Sender = require('ws/lib/Sender');
+
+const acceptExtensions = ({extensions, isServer}) => {
+    const {extensionName} = PerMessageDeflate;
+    const extension = extensions[extensionName];
+
+    if (!extension) {
+        return {};
+    }
+
+    const perMessageDeflate = new PerMessageDeflate({}, isServer);
+    perMessageDeflate.accept(extension);
+
+    return {[extensionName]: perMessageDeflate};
+};
+
+module.exports = class Interceptor {
+    static create(opts = {}) {
+        return new this(opts);
+    }
+
+    constructor({socket, options, req, proxyReq, proxyRes, proxySocket}) {
+        this._socket = socket;
+        this._options = options;
+        this._req = req;
+        this._proxyReq = proxyReq;
+        this._proxyRes = proxyRes;
+        this._proxySocket = proxySocket;
+        this._isClientSocketOpen = true;
+        this._isServerSocketOpen = true;
+
+        this._configure();
+    }
+
+    _configure() {
+        this._proxySocket.on('close', () => this._isClientSocketOpen = false);
+        this._socket.on('close', () => this._isServerSocketOpen = false);
+
+        const secWsExtensions = this._proxyRes.headers['sec-websocket-extensions'];
+        const extensions = Extensions.parse(secWsExtensions);
+        this._isCompressed = secWsExtensions && secWsExtensions.includes('permessage-deflate');
+
+        // need both versions of extensions for each side of the proxy connection
+        this._clientExtensions = this._isCompressed ? acceptExtensions({extensions, isServer: false}) : null;
+        this._serverExtensions = this._isCompressed ? acceptExtensions({extensions, isServer: true}) : null;
+    }
+
+    _getDataSender({sender, dataSendCond, event, options}) {
+        return ({data, binary = false}) => {
+            const opts = Object.assign({fin: true, compress: this._isCompressed, binary}, options);
+
+            dataSendCond() && sender.send(data, opts);
+
+            this._proxyReq.emit(event, {data, binary});
+        };
+    }
+
+    _getMsgHandler({interceptor, dataSender, binary}) {
+        return (data, flags) => {
+            if (typeof interceptor !== 'function') {
+                dataSender({data});
+                return;
+            }
+
+            const modifiedData = interceptor(data, {req: this._req, flags});
+
+            // if interceptor does not return data then nothing will be sended to the server
+            if (modifiedData) {
+                dataSender({data: modifiedData, binary});
+            }
+        }
+    }
+
+    _interceptClientMessages() {
+        const receiver = new Receiver(this._clientExtensions);
+        const sender = new Sender(this._proxySocket, this._serverExtensions);
+        this._proxyReq.emit('clientSenderInited', sender);
+
+        // frame must be masked when send from client to server - https://tools.ietf.org/html/rfc6455#section-5.3
+        const options = {mask: true};
+        const dataSendCond = () => this._isClientSocketOpen;
+        const dataSender = this._getDataSender({sender, dataSendCond, event: 'wsClientMsg', options});
+
+        receiver.ontext = this._getMsgHandler({interceptor: this._options.wsInterceptClientMsg, dataSender, binary: false});
+        receiver.onbinary = this._getMsgHandler({interceptor: this._options.wsInterceptClientMsg, dataSender, binary: true});
+        receiver.onclose = (code, msg, {masked: mask}) => this._isClientSocketOpen && sender.close(code, msg, mask);
+
+        this._socket.on('data', (data) => receiver.add(data));
+    }
+
+    _interceptServerMessages() {
+        const receiver = new Receiver(this._serverExtensions);
+        const sender = new Sender(this._socket, this._clientExtensions);
+        this._proxyReq.emit('serverSenderInited', sender);
+
+        const options = {mask: false};
+        const dataSendCond = () => this._isServerSocketOpen;
+        const dataSender = this._getDataSender({sender, dataSendCond, event: 'wsServerMsg', options});
+
+        receiver.ontext = this._getMsgHandler({interceptor: this._options.wsInterceptServerMsg, dataSender, binary: false});
+        receiver.onbinary = this._getMsgHandler({interceptor: this._options.wsInterceptServerMsg, dataSender, binary: true});
+        receiver.onclose = (code, msg, {masked: mask}) => this._isServerSocketOpen && sender.close(code, msg, mask);
+
+        this._proxySocket.on('data', (data) => receiver.add(data));
+    }
+
+    startDataTransfer() {
+        this._interceptClientMessages();
+        this._interceptServerMessages();
+    }
+};