[Streaming] Add working ws websocket implementation for Node environm…

…ent (#1334)

* add working ws websocket impl

* remove unused constants

libraries/botframework-streaming/package.json

@@ -10,6 +10,7 @@
   "devDependencies": {
     "@types/chai": "^4.1.7",
     "@types/node": "^10.12.18",
+    "@types/ws": "^6.0.3",
     "chai": "^4.2.0",
     "mocha": "^6.2.0",
     "nyc": "^14.1.1",
@@ -22,7 +23,8 @@
   "dependencies": {
     "promise.prototype.finally": "^3.1.0",
     "uuid": "^3.3.2",
-    "watershed": "^0.4.0"
+    "watershed": "^0.4.0",
+    "ws": "^7.1.2"
   },
   "engines": {
     "node": ">12.3"

libraries/botframework-streaming/src/index.ts

@@ -20,4 +20,6 @@ export {
     NodeWebSocketFactoryBase,
     WebSocketClient,
     WebSocketServer,
+    WsNodeWebSocket,
+    WsNodeWebSocketFactory,
 } from './webSocket';

libraries/botframework-streaming/src/interfaces/ISocket.ts

@@ -11,7 +11,7 @@
  * with the WebSocket server or client.
  */
 export interface ISocket {
-    isConnected(): boolean;
+    isConnected: boolean;
     write(buffer: Buffer);
     connect(serverAddress: string): Promise<void>;
     close();

libraries/botframework-streaming/src/webSocket/browserWebSocket.ts

@@ -52,7 +52,7 @@ export class BrowserWebSocket implements ISocket {
     /**
      * True if the socket is currently connected.
      */
-    public isConnected(): boolean {
+    public get isConnected(): boolean {
         return this.webSocket.readyState === 1;
     }
 

libraries/botframework-streaming/src/webSocket/factories/index.ts

@@ -8,3 +8,4 @@
 
 export * from './nodeWebSocketFactory';
 export * from './nodeWebSocketFactoryBase';
+export * from './wsNodeWebSocketFactory';

libraries/botframework-streaming/src/webSocket/factories/nodeWebSocketFactory.ts

@@ -16,7 +16,7 @@ export class NodeWebSocketFactory extends NodeWebSocketFactoryBase {
     constructor() {
         super();
     }
-    
+
     /**
      * Creates a NodeWebSocket instance.
      * @param req 
@@ -26,7 +26,7 @@ export class NodeWebSocketFactory extends NodeWebSocketFactoryBase {
     public createWebSocket(req: IncomingMessage, socket: Socket, head: Buffer): NodeWebSocket {
         const s = new NodeWebSocket();
         s.create(req, socket, head);
-        
+
         return s;
     }
 }

libraries/botframework-streaming/src/webSocket/factories/wsNodeWebSocketFactory.ts

@@ -0,0 +1,27 @@
+/**
+ * @module botframework-streaming
+ */
+/**
+ * Copyright (c) Microsoft Corporation. All rights reserved.
+ * Licensed under the MIT License.
+ */
+
+import { IncomingMessage } from 'http';
+import { Socket } from 'net';
+
+import { WsNodeWebSocket } from '../wsNodeWebSocket';
+
+export class WsNodeWebSocketFactory {    
+    /**
+     * Creates a WsNodeWebSocket instance.
+     * @param req 
+     * @param socket 
+     * @param head 
+     */
+    public async createWebSocket(req: IncomingMessage, socket: Socket, head: Buffer): Promise<WsNodeWebSocket> {
+        const s = new WsNodeWebSocket();
+        await s.create(req, socket, head);
+
+        return s;
+    }
+}

libraries/botframework-streaming/src/webSocket/index.ts

@@ -13,3 +13,4 @@ export * from './nodeWebSocket';
 export * from './webSocketClient';
 export * from './webSocketServer';
 export * from './webSocketTransport';
+export * from './wsNodeWebSocket';

libraries/botframework-streaming/src/webSocket/nodeWebSocket.ts

@@ -41,7 +41,7 @@ export class NodeWebSocket implements ISocket {
     /**
      * True if the socket is currently connected.
      */
-    public isConnected(): boolean {
+    public get isConnected(): boolean {
         return this.connected;
     }
 

libraries/botframework-streaming/src/webSocket/webSocketTransport.ts

@@ -48,7 +48,7 @@ export class WebSocketTransport implements ITransportSender, ITransportReceiver
      * @param buffer The buffered data to send out over the connection.
      */
     public send(buffer: Buffer): number {
-        if (this._socket && this._socket.isConnected()) {
+        if (this._socket && this._socket.isConnected) {
             this._socket.write(buffer);
 
             return buffer.length;
@@ -61,14 +61,14 @@ export class WebSocketTransport implements ITransportSender, ITransportReceiver
      * Returns true if the transport is connected to a socket.
      */
     public isConnected(): boolean {
-        return this._socket.isConnected();
+        return this._socket.isConnected;
     }
 
     /**
      * Close the socket this transport is connected to.
      */
     public close(): void {
-        if (this._socket && this._socket.isConnected()) {
+        if (this._socket && this._socket.isConnected) {
             this._socket.close();
         }
     }

libraries/botframework-streaming/src/webSocket/wsNodeWebSocket.ts

@@ -0,0 +1,139 @@
+/**
+ * @module botframework-streaming
+ */
+/**
+ * Copyright (c) Microsoft Corporation. All rights reserved.
+ * Licensed under the MIT License.
+ */
+
+import { ISocket } from '../interfaces';
+import { IncomingMessage, request } from 'http';
+import { Socket } from 'net';
+import * as WebSocket from 'ws';
+import * as crypto from 'crypto';
+
+const WS_SERVER = new WebSocket.Server({ noServer: true });
+
+// Taken from watershed, these needs to be investigated.
+const NONCE_LENGTH = 16;
+
+export class WsNodeWebSocket implements ISocket {
+    private wsSocket: WebSocket;
+    private connected: boolean;
+
+    /**
+     * Creates a new instance of the [WsNodeWebSocket](xref:botframework-streaming.WsNodeWebSocket) class.
+     *
+     * @param socket The ws socket object to build this connection on.
+     */
+    public constructor(wsSocket?: WebSocket) {
+        this.wsSocket = wsSocket;
+        this.connected = !!wsSocket;
+    }
+
+    /**
+     * Create and set a `ws` WebSocket with an HTTP Request, Socket and Buffer.
+     * @param req IncomingMessage
+     * @param socket Socket
+     * @param head Buffer
+     */
+    public async create(req: IncomingMessage, socket: Socket, head: Buffer): Promise<void> {
+        return new Promise<void>((resolve, reject) => {
+            try {
+                WS_SERVER.handleUpgrade(req, socket, head, (websocket) => {
+                    this.wsSocket = websocket;
+                    this.connected = true;
+                    resolve();
+                });
+            } catch (err) {
+                reject(err);
+            }
+        });
+    }
+
+    /**
+     * True if the socket is currently connected.
+     */
+    public get isConnected(): boolean {
+        return this.connected;
+    }
+
+    /**
+     * Writes a buffer to the socket and sends it.
+     *
+     * @param buffer The buffer of data to send across the connection.
+     */
+    public write(buffer: Buffer): void {
+        this.wsSocket.send(buffer);
+    }
+
+    /**
+     * Connects to the supporting socket using WebSocket protocol.
+     *
+     * @param serverAddress The address the server is listening on.
+     * @param port The port the server is listening on, defaults to 8082.
+     */
+    public async connect(serverAddress, port = 8082): Promise<void> {
+        // Taken from WaterShed, this needs to be investigated.
+        const wskey = crypto.randomBytes(NONCE_LENGTH).toString('base64');
+        const options = {
+            port: port,
+            hostname: serverAddress,
+            headers: {
+                connection: 'upgrade',
+                'Sec-WebSocket-Key': wskey,
+                'Sec-WebSocket-Version': '13'
+            }
+        };
+        const req = request(options);
+        req.end();
+        req.on('upgrade', (res, socket, head): void => {
+            // @types/ws does not contain the signature for completeUpgrade
+            // https://github.com/websockets/ws/blob/0a612364e69fc07624b8010c6873f7766743a8e3/lib/websocket-server.js#L269
+            (WS_SERVER as any).completeUpgrade(wskey, undefined, res, socket, head, (websocket): void => {
+                this.wsSocket = websocket;
+                this.connected = true;
+            });
+        });
+
+        return new Promise<void>((resolve, reject): void => {
+            req.on('close', resolve);
+            req.on('error', reject);
+        });
+    }
+
+    /**
+     * Set the handler for `'data'` and `'message'` events received on the socket.
+     */
+    public setOnMessageHandler(handler: (x: any) => void): void {
+        this.wsSocket.on('data', handler);
+        this.wsSocket.on('message', handler);
+    }
+
+    /**
+     * Close the socket.
+     * @remarks
+     * Optionally pass in a status code and string explaining why the connection is closing.
+     * @param code
+     * @param data
+     */
+    public close(code?: number, data?: string): void {
+        this.connected = false;
+
+        return this.wsSocket.close(code, data);
+    }
+
+    /**
+     * Set the callback to call when encountering socket closures.
+     */
+    public setOnCloseHandler(handler: (x: any) => void): void {
+        this.wsSocket.on('close', handler);
+    }
+
+    /**
+     * Set the callback to call when encountering errors.
+     */
+    public setOnErrorHandler(handler: (x: any) => void): void {
+        this.wsSocket.on('error', (error): void => { if (error) { handler(error); } });
+    }
+}

libraries/botframework-streaming/tests/NodeWebSocket.test.js

@@ -19,7 +19,7 @@ describe('NodeSocket', () => {
 
     it('starts out connected', () => {
         const ns = new NodeWebSocket(new FauxSock);
-        expect(ns.isConnected()).to.be.true;
+        expect(ns.isConnected).to.be.true;
     });
 
     it('writes to the socket', () => {

libraries/botframework-streaming/tests/WebSocket.test.js

@@ -231,7 +231,7 @@ describe('Streaming Extensions WebSocket Library Tests', () => {
         it('knows its connected', () => {
             let bs = new ws.BrowserWebSocket( new FauxSock());
             bs.connect('fakeUrl');
-            expect(bs.isConnected()).to.be.true;
+            expect(bs.isConnected).to.be.true;
         });
 
         it('writes to the socket', () => {

libraries/botframework-streaming/tests/helpers/fauxSock.js

@@ -18,9 +18,32 @@ class FauxSock {
         this.onmessage = undefined;
         this.onerror = undefined;
         this.onclose = undefined;
+
+        // `ws` specific check in WebSocketServer.completeUpgrade
+        this.readable = true;
+        this.writable = true;
+    }
+
+    /* Start of `ws` specific methods. */
+    removeListener(event, handler) {
+        switch (event) {
+            case 'error':
+                return;
+            default:
+                console.error(`FauxSock.removeListener(): Reached default case: ${event}`);
+        }
+    }
+
+    setTimeout(value) {
+        this.timeoutValue = value;
+        return;
+    }
+
+    setNoDelay() {
     }
+    /* End of `ws` specific methods. */
 
-    isConnected() {
+    get isConnected() {
         return this.connected;
     }
 
@@ -75,14 +98,24 @@ class FauxSock {
         if (action === 'close') {
             this.closeHandler = handler;
         }
+        if (action === 'end') {
+            this.endHandler = handler;
+        }
+        // Required for `watershed` WebSockets
         if (action === 'text') {
             this.textHandler = handler;
         }
+        // Required for `watershed` WebSockets
         if (action === 'binary') {
             this.binaryHandler = handler;
         }
-        if (action === 'end') {
-            this.endHandler = handler;
+        // Required for `ws` WebSockets
+        if (action === 'data') {
+            this.dataHandler = handler;
+        }
+        // Required for `ws` WebSockets
+        if (action === 'message') {
+            this._messageHandler = handler;
         }
     };
 

libraries/botframework-streaming/tests/helpers/testRequest.js

@@ -8,11 +8,17 @@ class TestRequest {
         let headers = [];
     }
 
+    setMethod(verb) {
+        this.method = 'GET';
+    }
+
     isUpgradeRequest() {
         return this.upgradeRequestVal;
     }
 
     setIsUpgradeRequest(value) {
+        // `ws` specific check
+        this.method = 'GET';
         this.upgradeRequestVal = value;
     }
 

libraries/botframework-streaming/tests/wsNodeWebSocket.test.js

@@ -0,0 +1,82 @@
+const { WsNodeWebSocket } = require('../');
+const { expect } = require('chai');
+const { FauxSock, TestRequest } = require('./helpers');
+const { randomBytes } = require('crypto');
+
+describe('WsNodeWebSocket', () => {
+    it('creates a new WsNodeWebSocket', () => {
+        const wsSocket = new WsNodeWebSocket(new FauxSock);
+        expect(wsSocket).to.be.instanceOf(WsNodeWebSocket);
+        expect(wsSocket.close()).to.not.throw;
+    });
+
+    it('requires a valid URL', () => {
+        try {
+            const wsSocket = new WsNodeWebSocket(new FauxSock);
+        } catch (error) {
+            expect(error.message).to.equal('Invalid URL: fakeURL');
+        }
+    });
+
+    it('starts out connected', () => {
+        const wsSocket = new WsNodeWebSocket(new FauxSock);
+        expect(wsSocket.isConnected).to.be.true;
+    });
+
+    it('writes to the socket', () => {
+        const wsSocket = new WsNodeWebSocket(new FauxSock);
+        const buff = Buffer.from('hello');
+        expect(wsSocket.write(buff)).to.not.throw;
+    });
+
+    it('attempts to open a connection', () => {
+        const wsSocket = new WsNodeWebSocket(new FauxSock);
+        expect(wsSocket.connect().catch((error) => {
+            expect(error.message).to.equal('connect ECONNREFUSED 127.0.0.1:8082');
+        }));
+    });
+
+    it('can set message handlers on the socket', () => {
+        const sock = new FauxSock();
+        const wsSocket = new WsNodeWebSocket(sock);
+        expect(sock.dataHandler).to.be.undefined;
+        expect(sock._messageHandler).to.be.undefined;
+        expect(wsSocket.setOnMessageHandler(() => { })).to.not.throw;
+        expect(sock.dataHandler).to.not.be.undefined;
+        expect(sock._messageHandler).to.not.be.undefined;
+    });
+
+    it('can set error handler on the socket', () => {
+        const sock = new FauxSock();
+        const wsSocket = new WsNodeWebSocket(sock);
+        expect(sock.errorHandler).to.be.undefined;
+        expect(wsSocket.setOnErrorHandler(() => { })).to.not.throw;
+        expect(sock.errorHandler).to.not.be.undefined;
+    });
+
+    it('can set end handler on the socket', () => {
+        const sock = new FauxSock();
+        const wsSocket = new WsNodeWebSocket(sock);
+        expect(sock.closeHandler).to.be.undefined;
+        expect(wsSocket.setOnCloseHandler(() => { })).to.not.throw;
+        expect(sock.closeHandler).to.not.be.undefined;
+    });
+
+    it('create() should be successful and set a WebSocket', async () => {
+        const sock = new FauxSock();
+        const nodeSocket = new WsNodeWebSocket();
+        const request = new TestRequest();
+
+        // Configure a proper upgrade request for `ws`.
+        request.setIsUpgradeRequest(true);
+        request.headers = { upgrade: 'websocket' };
+        // Use Node.js `crypto` module to calculate a valid 'sec-websocket-key' value. 
+        // The key must pass this RegExp:
+        // https://github.com/websockets/ws/blob/0a612364e69fc07624b8010c6873f7766743a8e3/lib/websocket-server.js#L12
+        request.headers['sec-websocket-key'] = randomBytes(16).toString('base64');
+        request.headers['sec-websocket-version'] = '13';
+        request.headers['sec-websocket-protocol'] = '';
+
+        await nodeSocket.create(request, sock, Buffer.from([]));
+    });
+});