Revert #11011

CHANGELOG.md

@@ -16,17 +16,6 @@
 - [callhierarchy] `paths.ts` and `glob.ts` moved to `core/src/common`; `language-selector.ts` moved to `editor/src/common`. Any imports will need to be updated.
 - [electron] removed redundant config option `disallowReloadKeybinding` from `dev-packages/application-package/src/application-props.ts` file and corresponding test [#11099](https://github.com/eclipse-theia/theia/pull/11099)
 
-<a name="breaking_changes_1.26.0">[Breaking Changes:](#breaking_changes_1.26.0)</a>
-
-- [core] Refactored the core messaging API. Replaced `vscode-ws-jsonrpc` with a custom RPC protocol that is better suited for handling binary data and enables message tunneling.
-  This impacts all main concepts of the messaging API. The API no longer exposes a `Connection` object and uses a generic `Channel` implementation instead.
-  - Replaces usage of `vscode-json-rpc`'s `Connection` with the new generic `Channel`. Affects `AbstractConnectionProvider`, `MessagingService`, `IPCConnectionProvider`, `ElectronMessagingService`
-  - `MessagingService`: No longer offers the `listen` and `forward` method. Use `wsChannel` instead.
-  - `RemoteFileSystemServer`: Use `UInt8Array` instead of plain number arrays for all arguments and return type that store binary data
-  - `DebugAdapter`: Replaced the debug-service internal `Channel` implementation with the newly introduced generic `Channel`.
-  [#11011](https://github.com/eclipse-theia/theia/pull/11011) - Contributed on behalf of STMicroelectronics.
-
-
 ## v1.25.0 - 4/28/2022
 
 [1.25.0 Milestone](https://github.com/eclipse-theia/theia/milestone/35)

package.json

@@ -10,8 +10,6 @@
     "**/@types/node": "12"
   },
   "devDependencies": {
-    "@types/chai": "4.3.0",
-    "@types/chai-spies": "1.0.3",
     "@types/chai-string": "^1.4.0",
     "@types/jsdom": "^11.0.4",
     "@types/node": "12",
@@ -22,8 +20,6 @@
     "@typescript-eslint/eslint-plugin": "^4.8.1",
     "@typescript-eslint/eslint-plugin-tslint": "^4.8.1",
     "@typescript-eslint/parser": "^4.8.1",
-    "chai": "4.3.4",
-    "chai-spies": "1.0.0",
     "chai-string": "^1.4.0",
     "chalk": "4.0.0",
     "concurrently": "^3.5.0",

packages/core/README.md

@@ -96,6 +96,7 @@ export class SomeClass {
     - `react-virtualized` (from [`react-virtualized@^9.20.0`](https://www.npmjs.com/package/react-virtualized))
     - `vscode-languageserver-protocol` (from [`vscode-languageserver-protocol@~3.15.3`](https://www.npmjs.com/package/vscode-languageserver-protocol))
     - `vscode-uri` (from [`vscode-uri@^2.1.1`](https://www.npmjs.com/package/vscode-uri))
+    - `vscode-ws-jsonrpc` (from [`vscode-ws-jsonrpc@^0.2.0`](https://www.npmjs.com/package/vscode-ws-jsonrpc))
     - `dompurify` (from [`dompurify@^2.2.9`](https://www.npmjs.com/package/dompurify))
     - `express` (from [`express@^4.16.3`](https://www.npmjs.com/package/express))
     - `lodash.debounce` (from [`lodash.debounce@^4.0.8`](https://www.npmjs.com/package/lodash.debounce))

packages/core/package.json

@@ -67,6 +67,7 @@
     "uuid": "^8.3.2",
     "vscode-languageserver-protocol": "~3.15.3",
     "vscode-uri": "^2.1.1",
+    "vscode-ws-jsonrpc": "^0.2.0",
     "ws": "^7.1.2",
     "yargs": "^15.3.1"
   },
@@ -106,7 +107,8 @@
         "react-dom",
         "react-virtualized",
         "vscode-languageserver-protocol",
-        "vscode-uri"
+        "vscode-uri",
+        "vscode-ws-jsonrpc"
       ],
       "export =": [
         "dompurify as DOMPurify",

packages/core/shared/vscode-ws-jsonrpc/index.d.ts

@@ -0,0 +1 @@
+export * from 'vscode-ws-jsonrpc';

packages/core/shared/vscode-ws-jsonrpc/index.js

@@ -0,0 +1 @@
+module.exports = require('vscode-ws-jsonrpc');

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

@@ -15,11 +15,11 @@
 // *****************************************************************************
 
 import { injectable, interfaces, decorate, unmanaged } from 'inversify';
-import { JsonRpcProxyFactory, JsonRpcProxy, Emitter, Event, Channel } from '../../common';
+import { JsonRpcProxyFactory, JsonRpcProxy, Emitter, Event } from '../../common';
+import { WebSocketChannel } from '../../common/messaging/web-socket-channel';
 import { Endpoint } from '../endpoint';
 import { AbstractConnectionProvider } from '../../common/messaging/abstract-connection-provider';
 import { io, Socket } from 'socket.io-client';
-import { IWebSocket, WebSocketChannel } from '../../common/messaging/web-socket-channel';
 
 decorate(injectable(), JsonRpcProxyFactory);
 decorate(unmanaged(), JsonRpcProxyFactory, 0);
@@ -35,8 +35,6 @@ export interface WebSocketOptions {
 export class WebSocketConnectionProvider extends AbstractConnectionProvider<WebSocketOptions> {
 
     protected readonly onSocketDidOpenEmitter: Emitter<void> = new Emitter();
-    // Socket that is used by the main channel
-    protected socket: Socket;
     get onSocketDidOpen(): Event<void> {
         return this.onSocketDidOpenEmitter.event;
     }
@@ -50,39 +48,31 @@ export class WebSocketConnectionProvider extends AbstractConnectionProvider<WebS
         return container.get(WebSocketConnectionProvider).createProxy<T>(path, arg);
     }
 
-    protected createMainChannel(): Channel {
+    protected readonly socket: Socket;
+
+    constructor() {
+        super();
         const url = this.createWebSocketUrl(WebSocketChannel.wsPath);
         const socket = this.createWebSocket(url);
-        const channel = new WebSocketChannel(this.toIWebSocket(socket));
         socket.on('connect', () => {
             this.fireSocketDidOpen();
         });
-        channel.onClose(() => this.fireSocketDidClose());
+        socket.on('disconnect', reason => {
+            for (const channel of [...this.channels.values()]) {
+                channel.close(undefined, reason);
+            }
+            this.fireSocketDidClose();
+        });
+        socket.on('message', data => {
+            this.handleIncomingRawMessage(data);
+        });
         socket.connect();
         this.socket = socket;
-
-        return channel;
-    }
-
-    protected toIWebSocket(socket: Socket): IWebSocket {
-        return {
-            close: () => {
-                socket.removeAllListeners('disconnect');
-                socket.removeAllListeners('error');
-                socket.removeAllListeners('message');
-                socket.close();
-            },
-            isConnected: () => socket.connected,
-            onClose: cb => socket.on('disconnect', reason => cb(reason)),
-            onError: cb => socket.on('error', reason => cb(reason)),
-            onMessage: cb => socket.on('message', data => cb(data)),
-            send: message => socket.emit('message', message)
-        };
     }
 
-    override async openChannel(path: string, handler: (channel: Channel) => void, options?: WebSocketOptions): Promise<void> {
+    override openChannel(path: string, handler: (channel: WebSocketChannel) => void, options?: WebSocketOptions): void {
         if (this.socket.connected) {
-            return super.openChannel(path, handler, options);
+            super.openChannel(path, handler, options);
         } else {
             const openChannel = () => {
                 this.socket.off('connect', openChannel);
@@ -92,6 +82,14 @@ export class WebSocketConnectionProvider extends AbstractConnectionProvider<WebS
         }
     }
 
+    protected createChannel(id: number): WebSocketChannel {
+        return new WebSocketChannel(id, content => {
+            if (this.socket.connected) {
+                this.socket.send(content);
+            }
+        });
+    }
+
     /**
      * @param path The handler to reach in the backend.
      */
@@ -145,4 +143,3 @@ export class WebSocketConnectionProvider extends AbstractConnectionProvider<WebS
         this.onSocketDidCloseEmitter.fire(undefined);
     }
 }
-

packages/core/src/browser/progress-status-bar-item.ts

@@ -15,7 +15,7 @@
 // *****************************************************************************
 
 import { injectable, inject } from 'inversify';
-import { CancellationToken } from '../../shared/vscode-languageserver-protocol';
+import { CancellationToken } from 'vscode-ws-jsonrpc';
 import { ProgressClient, ProgressMessage, ProgressUpdate } from '../common';
 import { StatusBar, StatusBarAlignment } from './status-bar';
 import { Deferred } from '../common/promise-util';

packages/core/src/common/cancellation.ts

@@ -49,14 +49,6 @@ export namespace CancellationToken {
         isCancellationRequested: true,
         onCancellationRequested: shortcutEvent
     });
-
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    export function is(value: any): value is CancellationToken {
-        const candidate = value as CancellationToken;
-        return candidate && (candidate === CancellationToken.None
-            || candidate === CancellationToken.Cancelled
-            || (typeof candidate.isCancellationRequested === 'boolean' && !!candidate.onCancellationRequested));
-    }
 }
 
 export class CancellationError extends Error {

packages/core/src/common/index.ts

@@ -29,7 +29,6 @@ export * from './contribution-provider';
 export * from './path';
 export * from './logger';
 export * from './messaging';
-export * from './message-rpc';
 export * from './message-service';
 export * from './message-service-protocol';
 export * from './progress-service';

packages/core/src/common/messaging/abstract-connection-provider.ts

@@ -15,10 +15,11 @@
 // *****************************************************************************
 
 import { injectable, interfaces } from 'inversify';
+import { ConsoleLogger, createWebSocketConnection, Logger } from 'vscode-ws-jsonrpc';
 import { Emitter, Event } from '../event';
 import { ConnectionHandler } from './handler';
 import { JsonRpcProxy, JsonRpcProxyFactory } from './proxy-factory';
-import { Channel, ChannelMultiplexer } from '../message-rpc/channel';
+import { WebSocketChannel } from './web-socket-channel';
 
 /**
  * Factor common logic according to `ElectronIpcConnectionProvider` and
@@ -44,6 +45,9 @@ export abstract class AbstractConnectionProvider<AbstractOptions extends object>
         throw new Error('abstract');
     }
 
+    protected channelIdSeq = 0;
+    protected readonly channels = new Map<number, WebSocketChannel>();
+
     protected readonly onIncomingMessageActivityEmitter: Emitter<void> = new Emitter();
     get onIncomingMessageActivity(): Event<void> {
         return this.onIncomingMessageActivityEmitter.event;
@@ -71,39 +75,50 @@ export abstract class AbstractConnectionProvider<AbstractOptions extends object>
         return factory.createProxy();
     }
 
-    protected channelMultiPlexer: ChannelMultiplexer;
-
-    constructor() {
-        this.channelMultiPlexer = this.createMultiplexer();
-    }
-
-    protected createMultiplexer(): ChannelMultiplexer {
-        return new ChannelMultiplexer(this.createMainChannel());
-    }
-
     /**
      * Install a connection handler for the given path.
      */
     listen(handler: ConnectionHandler, options?: AbstractOptions): void {
         this.openChannel(handler.path, channel => {
-            handler.onConnection(channel);
+            const connection = createWebSocketConnection(channel, this.createLogger());
+            connection.onDispose(() => channel.close());
+            handler.onConnection(connection);
         }, options);
     }
 
-    async openChannel(path: string, handler: (channel: Channel) => void, options?: AbstractOptions): Promise<void> {
-        const newChannel = await this.channelMultiPlexer.open(path);
-        newChannel.onClose(() => {
-            const { reconnecting } = { reconnecting: true, ...options };
-            if (reconnecting) {
-                this.openChannel(path, handler, options);
+    openChannel(path: string, handler: (channel: WebSocketChannel) => void, options?: AbstractOptions): void {
+        const id = this.channelIdSeq++;
+        const channel = this.createChannel(id);
+        this.channels.set(id, channel);
+        channel.onClose(() => {
+            if (this.channels.delete(channel.id)) {
+                const { reconnecting } = { reconnecting: true, ...options };
+                if (reconnecting) {
+                    this.openChannel(path, handler, options);
+                }
+            } else {
+                console.error('The ws channel does not exist', channel.id);
             }
         });
-        handler(newChannel);
+        channel.onOpen(() => handler(channel));
+        channel.open(path);
     }
 
-    /**
-     * Create the main connection that is used for multiplexing all channels.
-     */
-    protected abstract createMainChannel(): Channel;
+    protected abstract createChannel(id: number): WebSocketChannel;
+
+    protected handleIncomingRawMessage(data: string): void {
+        const message: WebSocketChannel.Message = JSON.parse(data);
+        const channel = this.channels.get(message.id);
+        if (channel) {
+            channel.handleMessage(message);
+        } else {
+            console.error('The ws channel does not exist', message.id);
+        }
+        this.onIncomingMessageActivityEmitter.fire(undefined);
+    }
+
+    protected createLogger(): Logger {
+        return new ConsoleLogger();
+    }
 
 }

packages/core/src/common/messaging/connection-error-handler.ts

@@ -14,6 +14,7 @@
 // SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
 // *****************************************************************************
 
+import { Message } from 'vscode-ws-jsonrpc';
 import { ILogger } from '../../common';
 
 export interface ResolvedConnectionErrorHandlerOptions {
@@ -50,7 +51,7 @@ export class ConnectionErrorHandler {
         };
     }
 
-    shouldStop(error: Error, count?: number): boolean {
+    shouldStop(error: Error, message?: Message, count?: number): boolean {
         return !count || count > this.options.maxErrors;
     }
 

packages/core/src/common/messaging/handler.ts

@@ -14,11 +14,11 @@
 // SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
 // *****************************************************************************
 
-import { Channel } from '../message-rpc/channel';
+import { MessageConnection } from 'vscode-ws-jsonrpc';
 
 export const ConnectionHandler = Symbol('ConnectionHandler');
 
 export interface ConnectionHandler {
     readonly path: string;
-    onConnection(connection: Channel): void;
+    onConnection(connection: MessageConnection): void;
 }

packages/core/src/common/messaging/proxy-factory.spec.ts

@@ -15,11 +15,21 @@
 // *****************************************************************************
 
 import * as chai from 'chai';
+import { ConsoleLogger } from '../../node/messaging/logger';
 import { JsonRpcProxyFactory, JsonRpcProxy } from './proxy-factory';
-import { ChannelPipe } from '../message-rpc/channel.spec';
+import { createMessageConnection } from 'vscode-jsonrpc/lib/main';
+import * as stream from 'stream';
 
 const expect = chai.expect;
 
+class NoTransform extends stream.Transform {
+
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    override _transform(chunk: any, encoding: string, callback: Function): void {
+        callback(undefined, chunk);
+    }
+}
+
 class TestServer {
     requests: string[] = [];
     doStuff(arg: string): Promise<string> {
@@ -92,12 +102,15 @@ function getSetup(): {
     const server = new TestServer();
 
     const serverProxyFactory = new JsonRpcProxyFactory<TestServer>(client);
-    const pipe = new ChannelPipe();
-    serverProxyFactory.listen(pipe.right);
+    const client2server = new NoTransform();
+    const server2client = new NoTransform();
+    const serverConnection = createMessageConnection(server2client, client2server, new ConsoleLogger());
+    serverProxyFactory.listen(serverConnection);
     const serverProxy = serverProxyFactory.createProxy();
 
     const clientProxyFactory = new JsonRpcProxyFactory<TestClient>(server);
-    clientProxyFactory.listen(pipe.left);
+    const clientConnection = createMessageConnection(client2server, server2client, new ConsoleLogger());
+    clientProxyFactory.listen(clientConnection);
     const clientProxy = clientProxyFactory.createProxy();
     return {
         client,