Add ability to intercept websocket upgrades and raise events.

Signed-off-by: [email protected]

packages/core/src/node/messaging/messaging-backend-module.ts

@@ -20,6 +20,7 @@ import { BackendApplicationContribution } from '../backend-application';
 import { MessagingContribution, MessagingContainer } from './messaging-contribution';
 import { ConnectionContainerModule } from './connection-container-module';
 import { MessagingService } from './messaging-service';
+import { MessagingListener, MessagingListenerContribution } from './messaging-listeners';
 
 export const messagingBackendModule = new ContainerModule(bind => {
     bindContributionProvider(bind, ConnectionContainerModule);
@@ -31,4 +32,6 @@ export const messagingBackendModule = new ContainerModule(bind => {
         return child.get(MessagingService.Identifier);
     }).inSingletonScope();
     bind(BackendApplicationContribution).toService(MessagingContribution);
+    bind(MessagingListener).toSelf().inSingletonScope();
+    bindContributionProvider(bind, MessagingListenerContribution);
 });

packages/core/src/node/messaging/messaging-contribution.ts

@@ -32,6 +32,7 @@ import { ConsoleLogger } from './logger';
 import { ConnectionContainerModule } from './connection-container-module';
 import Route = require('route-parser');
 import { WsRequestValidator } from '../ws-request-validators';
+import { MessagingListener } from './messaging-listeners';
 
 export const MessagingContainer = Symbol('MessagingContainer');
 
@@ -50,6 +51,9 @@ export class MessagingContribution implements BackendApplicationContribution, Me
     @inject(WsRequestValidator)
     protected readonly wsRequestValidator: WsRequestValidator;
 
+    @inject(MessagingListener)
+    protected readonly messagingListener: MessagingListener;
+
     protected webSocketServer: ws.Server | undefined;
     protected readonly wsHandlers = new MessagingContribution.ConnectionHandlers<ws>();
     protected readonly channelHandlers = new MessagingContribution.ConnectionHandlers<WebSocketChannel>();
@@ -122,6 +126,7 @@ export class MessagingContribution implements BackendApplicationContribution, Me
             if (allowed) {
                 this.webSocketServer!.handleUpgrade(request, socket, head, client => {
                     this.webSocketServer!.emit('connection', client, request);
+                    this.messagingListener.onDidWebSocketUpgrade(request, client);
                 });
             } else {
                 console.error(`refused a websocket connection: ${request.connection.remoteAddress}`);

packages/core/src/node/messaging/messaging-listeners.ts

@@ -0,0 +1,52 @@
+/********************************************************************************
+ * Copyright (C) 2021 MayStreet Inc. and others
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0.
+ *
+ * This Source Code may also be made available under the following Secondary
+ * Licenses when the conditions for such availability set forth in the Eclipse
+ * Public License v. 2.0 are satisfied: GNU General Public License, version 2
+ * with the GNU Classpath Exception which is available at
+ * https://www.gnu.org/software/classpath/license.html.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
+ ********************************************************************************/
+
+import { inject, injectable, named } from 'inversify';
+import { ContributionProvider, MaybePromise } from '../../common';
+
+import * as http from 'http';
+import * as ws from 'ws';
+
+/**
+ * Bind components to this symbol to subscribe to WebSocket events.
+ */
+export const MessagingListenerContribution = Symbol('MessagingListenerContribution');
+export interface MessagingListenerContribution {
+    /**
+     * Function invoked when a HTTP connection is upgraded to a websocket.
+     *
+     * @param request The HTTP connection upgrade request received by the server.
+     * @param socket The WebSocket that the connection was upgraded to.
+     */
+    onDidWebSocketUpgrade(request: http.IncomingMessage, socket: ws): MaybePromise<void>;
+}
+
+/**
+ * Handler of Theia messaging system events, dispatching to MessagingListenerContribution instances.
+ */
+@injectable()
+export class MessagingListener {
+
+    @inject(ContributionProvider) @named(MessagingListenerContribution)
+    protected readonly messagingListenerContributions: ContributionProvider<MessagingListenerContribution>;
+
+    /**
+     * Notify all the subscribed `MessagingListenerContribution`s that the Websocket was upgraded.
+     */
+    async onDidWebSocketUpgrade(request: http.IncomingMessage, socket: ws): Promise<void> {
+        await Promise.all(Array.from(this.messagingListenerContributions.getContributions(), async messagingListener => messagingListener.onDidWebSocketUpgrade(request, socket)));
+    }
+}