core: refactor communication layer to use socket.io (#10514)

Author: msujew

CHANGELOG.md

@@ -13,6 +13,8 @@
 - [plugin-ext] function `logMeasurement` of `PluginDeployerImpl` class and browser class `HostedPluginSupport` is replaced by `measure` using the new `Stopwatch` API [#10407](https://github.com/eclipse-theia/theia/pull/10407)
 - [plugin-ext] the constructor of `BackendApplication` class no longer invokes the `initialize` method. Instead, the `@postConstruct configure` method now starts by calling `initialize` [#10407](https://github.com/eclipse-theia/theia/pull/10407)
 - [plugin] Added support for `vscode.window.createStatusBarItem` [#10754](https://github.com/eclipse-theia/theia/pull/10754) - Contributed on behalf of STMicroelectronics
+- [core] Replaced raw WebSocket transport with Socket.io protocol, changed internal APIs accordingly
+- [core] Removed all of our own custom HTTP Polling implementation
 
 ## v1.22.0 - 1/27/2022
 
@@ -89,7 +91,7 @@
 - [plugin] added support for codicon icon references in view containers [#10491](https://github.com/eclipse-theia/theia/pull/10491)
 - [plugin] added support to set theme attributes in webviews [#10493](https://github.com/eclipse-theia/theia/pull/10493)
 - [plugin] fixed running plugin hosts on `electron` for `Windows` [#10518](https://github.com/eclipse-theia/theia/pull/10518)
-- [preferences] updated `AbstractResourcePreferenceProvider` to handle multiple preference settings in the same tick and handle open preference files. 
+- [preferences] updated `AbstractResourcePreferenceProvider` to handle multiple preference settings in the same tick and handle open preference files.
   It will save the file exactly once, and prompt the user if the file is dirty when a programmatic setting is attempted. [#7775](https://github.com/eclipse-theia/theia/pull/7775)
 - [preferences] added support for non-string enum values in schemas [#10511](https://github.com/eclipse-theia/theia/pull/10511)
 - [preferences] added support for rendering markdown descriptions [#10431](https://github.com/eclipse-theia/theia/pull/10431)
@@ -113,11 +115,11 @@
 - [plugin] changed return type of `WebviewThemeDataProvider.getActiveTheme()` to `Theme` instead of `WebviewThemeType` [#10493](https://github.com/eclipse-theia/theia/pull/10493)
 - [plugin] removed the application prop `resolveSystemPlugins`, builtin plugins should now be resolved at build time [#10353](https://github.com/eclipse-theia/theia/pull/10353)
 - [plugin] renamed `WebviewThemeData.activeTheme` to `activeThemeType` [#10493](https://github.com/eclipse-theia/theia/pull/10493)
-- [preferences] removed `PreferenceProvider#pendingChanges` field. It was previously set unreliably and caused race conditions. 
-  If a `PreferenceProvider` needs a mechanism for deferring the resolution of `PreferenceProvider#setPreference`, it should implement its own system. 
+- [preferences] removed `PreferenceProvider#pendingChanges` field. It was previously set unreliably and caused race conditions.
+  If a `PreferenceProvider` needs a mechanism for deferring the resolution of `PreferenceProvider#setPreference`, it should implement its own system.
   See PR for example implementation in `AbstractResourcePreferenceProvider`. [#7775](https://github.com/eclipse-theia/theia/pull/7775)
 - [terminal] removed deprecated `activateTerminal` method in favor of `open`. [#10529](https://github.com/eclipse-theia/theia/pull/10529)
-- [webpack] Source maps for the frontend renamed from `webpack://[namespace]/[resource-filename]...` to `webpack:///[resource-path]?[loaders]` where `resource-path` is the path to 
+- [webpack] Source maps for the frontend renamed from `webpack://[namespace]/[resource-filename]...` to `webpack:///[resource-path]?[loaders]` where `resource-path` is the path to
   the file relative to your application package's root [#10480](https://github.com/eclipse-theia/theia/pull/10480)
 
 ## v1.20.0 - 11/25/2021

doc/Migration.md

@@ -19,6 +19,17 @@ For example:
 }
 ```
 
+### v1.24.0
+
+#### From WebSocket to Socket.io
+
+This is a very important change to how Theia sends and receives messages with its backend.
+
+This new Socket.io protocol will try to establish a WebSocket connection whenever possible, but it may also
+setup HTTP polling. It may even try to connect through HTTP before attempting WebSocket.
+
+Make sure your network configurations support both WebSockets and/or HTTP polling.
+
 ### v1.23.0
 
 #### TypeScript 4.5.5

packages/core/package.json

@@ -63,6 +63,8 @@
     "reflect-metadata": "^0.1.10",
     "route-parser": "^0.0.5",
     "safer-buffer": "^2.1.2",
+    "socket.io": "4.1.0",
+    "socket.io-client": "4.1.0",
     "uuid": "^8.3.2",
     "vscode-languageserver-protocol": "~3.15.3",
     "vscode-uri": "^2.1.1",

packages/core/src/browser/frontend-application-bindings.ts

@@ -15,7 +15,7 @@
  ********************************************************************************/
 
 import { interfaces } from 'inversify';
-import { bindContributionProvider, DefaultResourceProvider, MessageClient, MessageService, MessageServiceFactory, ResourceProvider, ResourceResolver } from '../common';
+import { bindContributionProvider, DefaultResourceProvider, MessageClient, MessageService, ResourceProvider, ResourceResolver } from '../common';
 import {
     bindPreferenceSchemaProvider, PreferenceProvider,
     PreferenceProviderProvider, PreferenceSchemaProvider, PreferenceScope,
@@ -24,7 +24,6 @@ import {
 
 export function bindMessageService(bind: interfaces.Bind): interfaces.BindingWhenOnSyntax<MessageService> {
     bind(MessageClient).toSelf().inSingletonScope();
-    bind(MessageServiceFactory).toFactory(({ container }) => () => container.get(MessageService));
     return bind(MessageService).toSelf().inSingletonScope();
 }
 

packages/core/src/browser/messaging/messaging-frontend-module.ts

@@ -15,9 +15,8 @@
  ********************************************************************************/
 
 import { ContainerModule } from 'inversify';
-import { DEFAULT_HTTP_FALLBACK_OPTIONS, HttpFallbackOptions, WebSocketConnectionProvider } from './ws-connection-provider';
+import { WebSocketConnectionProvider } from './ws-connection-provider';
 
 export const messagingFrontendModule = new ContainerModule(bind => {
-    bind(HttpFallbackOptions).toConstantValue(DEFAULT_HTTP_FALLBACK_OPTIONS);
     bind(WebSocketConnectionProvider).toSelf().inSingletonScope();
 });

packages/core/src/browser/messaging/ws-connection-provider.ts

@@ -14,13 +14,12 @@
  * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
  ********************************************************************************/
 
-import { injectable, interfaces, decorate, unmanaged, inject, optional } from 'inversify';
-import { JsonRpcProxyFactory, JsonRpcProxy, Emitter, Event, MessageService, MessageServiceFactory } from '../../common';
+import { injectable, interfaces, decorate, unmanaged } from 'inversify';
+import { JsonRpcProxyFactory, JsonRpcProxy, Emitter, Event } from '../../common';
 import { WebSocketChannel } from '../../common/messaging/web-socket-channel';
 import { Endpoint } from '../endpoint';
-import ReconnectingWebSocket from 'reconnecting-websocket';
 import { AbstractConnectionProvider } from '../../common/messaging/abstract-connection-provider';
-import { v4 as uuid } from 'uuid';
+import { io, Socket } from 'socket.io-client';
 
 decorate(injectable(), JsonRpcProxyFactory);
 decorate(unmanaged(), JsonRpcProxyFactory, 0);
@@ -32,29 +31,6 @@ export interface WebSocketOptions {
     reconnecting?: boolean;
 }
 
-export const HttpFallbackOptions = Symbol('HttpFallbackOptions');
-
-export interface HttpFallbackOptions {
-    /** Determines whether Theia is allowed to use the http fallback. True by default. */
-    allowed: boolean;
-    /** Number of failed websocket connection attempts before the fallback is triggered. 2 by default. */
-    maxAttempts: number;
-    /** The maximum duration (in ms) after which the http request should timeout. 5000 by default. */
-    pollingTimeout: number;
-    /** The timeout duration (in ms) after a request was answered with an error code. 5000 by default. */
-    errorTimeout: number;
-    /** The minimum timeout duration (in ms) between two http requests. 0 by default. */
-    requestTimeout: number;
-}
-
-export const DEFAULT_HTTP_FALLBACK_OPTIONS: HttpFallbackOptions = {
-    allowed: true,
-    maxAttempts: 2,
-    errorTimeout: 5000,
-    pollingTimeout: 5000,
-    requestTimeout: 0
-};
-
 @injectable()
 export class WebSocketConnectionProvider extends AbstractConnectionProvider<WebSocketOptions> {
 
@@ -68,128 +44,47 @@ export class WebSocketConnectionProvider extends AbstractConnectionProvider<WebS
         return this.onSocketDidCloseEmitter.event;
     }
 
-    protected readonly onHttpFallbackDidActivateEmitter: Emitter<void> = new Emitter();
-    get onHttpFallbackDidActivate(): Event<void> {
-        return this.onHttpFallbackDidActivateEmitter.event;
-    }
-
     static createProxy<T extends object>(container: interfaces.Container, path: string, arg?: object): JsonRpcProxy<T> {
         return container.get(WebSocketConnectionProvider).createProxy<T>(path, arg);
     }
 
-    @inject(MessageServiceFactory)
-    protected readonly messageService: () => MessageService;
-
-    @inject(HttpFallbackOptions) @optional()
-    protected readonly httpFallbackOptions: HttpFallbackOptions | undefined;
-
-    protected readonly socket: ReconnectingWebSocket;
-    protected useHttpFallback = false;
-    protected websocketErrorCounter = 0;
-    protected httpFallbackId = uuid();
-    protected httpFallbackDisconnected = true;
+    protected readonly socket: Socket;
 
     constructor() {
         super();
         const url = this.createWebSocketUrl(WebSocketChannel.wsPath);
         const socket = this.createWebSocket(url);
-        socket.onerror = event => this.handleSocketError(event);
-        socket.onopen = () => {
+        socket.on('connect', () => {
             this.fireSocketDidOpen();
-        };
-        socket.onclose = ({ code, reason }) => {
+        });
+        socket.on('disconnect', reason => {
             for (const channel of [...this.channels.values()]) {
-                channel.close(code, reason);
+                channel.close(undefined, reason);
             }
             this.fireSocketDidClose();
-        };
-        socket.onmessage = ({ data }) => {
+        });
+        socket.on('message', data => {
             this.handleIncomingRawMessage(data);
-        };
+        });
+        socket.connect();
         this.socket = socket;
-        window.addEventListener('offline', () => this.tryReconnect());
-        window.addEventListener('online', () => this.tryReconnect());
-    }
-
-    handleSocketError(event: unknown): void {
-        this.websocketErrorCounter += 1;
-        if (this.httpFallbackOptions?.allowed && this.websocketErrorCounter >= this.httpFallbackOptions?.maxAttempts) {
-            this.useHttpFallback = true;
-            this.socket.close();
-            const httpUrl = this.createHttpWebSocketUrl(WebSocketChannel.wsPath);
-            this.onHttpFallbackDidActivateEmitter.fire(undefined);
-            this.doLongPolling(httpUrl);
-            this.messageService().warn(
-                'Could not establish a websocket connection. The application will be using the HTTP fallback mode. This may affect performance and the behavior of some features.'
-            );
-        }
-        console.error(event);
-    }
-
-    async doLongPolling(url: string): Promise<void> {
-        let timeoutDuration = this.httpFallbackOptions?.requestTimeout || 0;
-        const controller = new AbortController();
-        const pollingId = window.setTimeout(() => controller.abort(), this.httpFallbackOptions?.pollingTimeout);
-        try {
-            const response = await fetch(url, {
-                method: 'POST',
-                headers: {
-                    'Content-Type': 'application/json'
-                },
-                signal: controller.signal,
-                keepalive: true,
-                body: JSON.stringify({ id: this.httpFallbackId, polling: true })
-            });
-            if (response.status === 200) {
-                window.clearTimeout(pollingId);
-                if (this.httpFallbackDisconnected) {
-                    this.fireSocketDidOpen();
-                }
-                const json: string[] = await response.json();
-                if (Array.isArray(json)) {
-                    for (const item of json) {
-                        this.handleIncomingRawMessage(item);
-                    }
-                } else {
-                    throw new Error('Received invalid long polling response.');
-                }
-            } else {
-                timeoutDuration = this.httpFallbackOptions?.errorTimeout || 0;
-                this.httpFallbackDisconnected = true;
-                this.fireSocketDidClose();
-                throw new Error('Response has error code: ' + response.status);
-            }
-        } catch (e) {
-            console.error('Error occurred during long polling', e);
-        }
-        setTimeout(() => this.doLongPolling(url), timeoutDuration);
     }
 
     openChannel(path: string, handler: (channel: WebSocketChannel) => void, options?: WebSocketOptions): void {
-        if (this.useHttpFallback || this.socket.readyState === WebSocket.OPEN) {
+        if (this.socket.connected) {
             super.openChannel(path, handler, options);
         } else {
             const openChannel = () => {
-                this.socket.removeEventListener('open', openChannel);
+                this.socket.off('connect', openChannel);
                 this.openChannel(path, handler, options);
             };
-            this.socket.addEventListener('open', openChannel);
-            this.onHttpFallbackDidActivate(openChannel);
+            this.socket.on('connect', openChannel);
         }
     }
 
     protected createChannel(id: number): WebSocketChannel {
-        const httpUrl = this.createHttpWebSocketUrl(WebSocketChannel.wsPath);
         return new WebSocketChannel(id, content => {
-            if (this.useHttpFallback) {
-                fetch(httpUrl, {
-                    method: 'POST',
-                    headers: {
-                        'Content-Type': 'application/json'
-                    },
-                    body: JSON.stringify({ id: this.httpFallbackId, content })
-                });
-            } else if (this.socket.readyState < WebSocket.CLOSING) {
+            if (this.socket.connected) {
                 this.socket.send(content);
             }
         });
@@ -211,33 +106,25 @@ export class WebSocketConnectionProvider extends AbstractConnectionProvider<WebS
     /**
      * Creates a web socket for the given url
      */
-    protected createWebSocket(url: string): ReconnectingWebSocket {
-        return new ReconnectingWebSocket(url, undefined, {
-            maxReconnectionDelay: 10000,
-            minReconnectionDelay: 1000,
-            reconnectionDelayGrowFactor: 1.3,
-            connectionTimeout: 10000,
-            maxRetries: Infinity,
-            debug: false
+    protected createWebSocket(url: string): Socket {
+        return io(url, {
+            reconnection: true,
+            reconnectionDelay: 1000,
+            reconnectionDelayMax: 10000,
+            reconnectionAttempts: Infinity,
+            extraHeaders: {
+                // Socket.io strips the `origin` header
+                // We need to provide our own for validation
+                'fix-origin': window.location.origin
+            }
         });
     }
 
     protected fireSocketDidOpen(): void {
-        // Once a websocket connection has opened, disable the http fallback
-        if (this.httpFallbackOptions?.allowed) {
-            this.httpFallbackOptions.allowed = false;
-        }
         this.onSocketDidOpenEmitter.fire(undefined);
     }
 
     protected fireSocketDidClose(): void {
         this.onSocketDidCloseEmitter.fire(undefined);
     }
-
-    protected tryReconnect(): void {
-        if (!this.useHttpFallback && this.socket.readyState !== WebSocket.CONNECTING) {
-            this.socket.reconnect();
-        }
-    }
-
 }

packages/core/src/common/message-service.ts

@@ -25,8 +25,6 @@ import {
 } from './message-service-protocol';
 import { CancellationTokenSource } from './cancellation';
 
-export const MessageServiceFactory = Symbol('MessageServiceFactory');
-
 /**
  * Service to log and categorize messages, show progress information and offer actions.
  *

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

@@ -14,7 +14,6 @@
  * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
  ********************************************************************************/
 
-import * as net from 'net';
 import * as http from 'http';
 import { injectable, inject } from 'inversify';
 import { MessagingContribution } from '../../node/messaging/messaging-contribution';
@@ -33,13 +32,10 @@ export class ElectronMessagingContribution extends MessagingContribution {
     /**
      * Only allow token-bearers.
      */
-    protected handleHttpUpgrade(request: http.IncomingMessage, socket: net.Socket, head: Buffer): void {
+    protected async allowConnect(request: http.IncomingMessage): Promise<boolean> {
         if (this.tokenValidator.allowRequest(request)) {
-            super.handleHttpUpgrade(request, socket, head);
-        } else {
-            console.error(`refused a websocket connection: ${request.connection.remoteAddress}`);
-            socket.destroy(); // kill connection, client will take that as a "no".
+            return super.allowConnect(request);
         }
+        return false;
     }
-
 }