Implement WebSocketServer Adapter (#5866)

* Implement WebSocketServerAdapter

* lint

* clean up

src/LiveQuery/ParseWebSocketServer.js

@@ -1,9 +1,7 @@
+import { loadAdapter } from '../Adapters/AdapterLoader';
+import { WSAdapter } from '../Adapters/WebSocketServer/WSAdapter';
 import logger from '../logger';
-
-const typeMap = new Map([['disconnect', 'close']]);
-const getWS = function(){
-  return require('ws');
-};
+import events from 'events';
 
 export class ParseWebSocketServer {
   server: Object;
@@ -11,14 +9,18 @@ export class ParseWebSocketServer {
   constructor(
     server: any,
     onConnect: Function,
-    websocketTimeout: number = 10 * 1000
+    config
   ) {
-    const WebSocketServer = getWS().Server;
-    const wss = new WebSocketServer({ server: server });
-    wss.on('listening', () => {
+    config.server = server;
+    const wss = loadAdapter(
+      config.wssAdapter,
+      WSAdapter,
+      config,
+    );
+    wss.onListen = () => {
       logger.info('Parse LiveQuery Server starts running');
-    });
-    wss.on('connection', ws => {
+    };
+    wss.onConnection = (ws) => {
       onConnect(new ParseWebSocket(ws));
       // Send ping to client periodically
       const pingIntervalId = setInterval(() => {
@@ -27,24 +29,29 @@ export class ParseWebSocketServer {
         } else {
           clearInterval(pingIntervalId);
         }
-      }, websocketTimeout);
-    });
+      }, config.websocketTimeout || 10 * 1000);
+    };
+    wss.start();
     this.server = wss;
   }
+
+  close() {
+    if (this.server && this.server.close) {
+      this.server.close();
+    }
+  }
 }
 
-export class ParseWebSocket {
+export class ParseWebSocket extends events.EventEmitter {
   ws: any;
 
   constructor(ws: any) {
+    super();
+    ws.onmessage = (request) => this.emit('message', request);
+    ws.onclose = () => this.emit('disconnect');
     this.ws = ws;
   }
 
-  on(type: string, callback): void {
-    const wsType = typeMap.has(type) ? typeMap.get(type) : type;
-    this.ws.on(wsType, callback);
-  }
-
   send(message: any): void {
     this.ws.send(message);
   }