Merge pull request #18 from SolidLabResearch/14-implement-usage-of-the-same-websocket-channel-if-already-subscribed-to-a-resource

fixes #14

Author: Kush Bisen

src/server/WebSocketServerHandler.ts

@@ -5,13 +5,13 @@ import { SubscribeNotification } from '../service/SubscribeNotification';
 export class WebSocketServerHandler {
 
     public websocket_server: any;
-    public websocket_connections: Map<string, WebSocket>;
+    public websocket_connections: Map<string, WebSocket[]>;
     public subscribe_notification: SubscribeNotification;
 
 
     constructor(websocket_server: WebSocket.server) {
         this.websocket_server = websocket_server;
-        this.websocket_connections = new Map<string, WebSocket>();
+        this.websocket_connections = new Map<string, WebSocket[]>();
         this.subscribe_notification = new SubscribeNotification();
     }
 
@@ -23,16 +23,14 @@ export class WebSocketServerHandler {
 
         this.websocket_server.on('request', (request: any) => {
             const connection = request.accept('solid-stream-notifications-aggregator', request.origin);
-            connection.on('message', (message: any) => {
+            connection.on('message', async(message: any) => {
                 if (message.type === 'utf8') {
                     const message_utf8 = message.utf8Data;
                     const ws_message = JSON.parse(message_utf8);
                     if (Object.keys(ws_message).includes('subscribe')) {
                         console.log(`Received a subscribe message from the client.`);
                         let stream_to_subscribe = ws_message.subscribe;
                         for (let stream of stream_to_subscribe) {
-                            // We first subscribe to the latest inbox of the LDES stream.
-                            this.subscribe_notification.subscribe_inbox(stream);
                             console.log(`Subscribed to the stream: ${stream}`);
                             this.set_connections(stream, connection);
                         }
@@ -41,7 +39,13 @@ export class WebSocketServerHandler {
                         console.log(`Received a new event message from the client.`);
                         let connection = this.websocket_connections.get(ws_message.stream);
                         if (connection !== undefined) {
-                            connection.send(JSON.stringify(ws_message));
+                            for (const [stream, connections] of this.websocket_connections) {
+                                if (stream == ws_message.stream) {
+                                    for (let connection of connections) {
+                                        connection.send(JSON.stringify(ws_message));
+                                    }
+                                }
+                            }
                         }
                     }
                     else if (Object.keys(ws_message).includes('container_location')) {
@@ -59,7 +63,18 @@ export class WebSocketServerHandler {
         });
     }
 
-    public set_connections(subscribed_stream: string, connection: WebSocket): void {
-        this.websocket_connections.set(subscribed_stream, connection);
+    public set_connections(subscribed_stream: string, connection: WebSocket){
+        if (!this.websocket_connections.has(subscribed_stream)) {   
+            this.subscribe_notification.subscribe_inbox(subscribed_stream);                     
+            this.websocket_connections.set(subscribed_stream, [connection]);
+        }
+        else {
+            const connections = this.websocket_connections.get(subscribed_stream);
+            if (connections !== undefined) {
+                connections.push(connection);
+                this.websocket_connections.set(subscribed_stream, connections);
+            }
+
+        }
     }
 }