Ensure that the Debug API is working with the new RPC (#11268)

After migrating to new binary RPC (83d4308) Debugging was no longer working as expected.
- Partly revert/Switch back to use a string-based message channel for Debug communication. 
- Rename the string-based `Channel` used in the Debug API to `DebugChannel` to avoid naming confusion with the default binary message `Channel`s.
- Introduce a `ForwardingDebugChannel` which can be used to wrap an arbitrary binary message `Channel` (i.e. the service websocket channel` in a `DebugChannel`

Fixes #11263

Contributed on behalf of STMicroelectronics

packages/debug/src/browser/debug-session-connection.ts

@@ -18,9 +18,13 @@
 
 import { DebugProtocol } from 'vscode-debugprotocol';
 import { Deferred } from '@theia/core/lib/common/promise-util';
-import { Event, Emitter, DisposableCollection, Disposable, MaybePromise, Channel } from '@theia/core';
+import { Event, Emitter, DisposableCollection, Disposable, MaybePromise } from '@theia/core';
 import { OutputChannel } from '@theia/output/lib/browser/output-channel';
 
+import { DebugChannel } from '../common/debug-service';
+
+export type DebugRequestHandler = (request: DebugProtocol.Request) => MaybePromise<any>;
+
 export interface DebugRequestTypes {
     'attach': [DebugProtocol.AttachRequestArguments, DebugProtocol.AttachResponse]
     'breakpointLocations': [DebugProtocol.BreakpointLocationsArguments, DebugProtocol.BreakpointLocationsResponse]
@@ -112,14 +116,12 @@ const standardDebugEvents = new Set<string>([
     'thread'
 ]);
 
-export type DebugRequestHandler = (request: DebugProtocol.Request) => MaybePromise<any>;
-
 export class DebugSessionConnection implements Disposable {
 
     private sequence = 1;
 
     protected readonly pendingRequests = new Map<number, Deferred<DebugProtocol.Response>>();
-    protected readonly connectionPromise: Promise<Channel>;
+    protected readonly connectionPromise: Promise<DebugChannel>;
 
     protected readonly requestHandlers = new Map<string, DebugRequestHandler>();
 
@@ -139,7 +141,7 @@ export class DebugSessionConnection implements Disposable {
 
     constructor(
         readonly sessionId: string,
-        connectionFactory: (sessionId: string) => Promise<Channel>,
+        connectionFactory: (sessionId: string) => Promise<DebugChannel>,
         protected readonly traceOutputChannel: OutputChannel | undefined
     ) {
         this.connectionPromise = this.createConnection(connectionFactory);
@@ -159,14 +161,14 @@ export class DebugSessionConnection implements Disposable {
         this.toDispose.dispose();
     }
 
-    protected async createConnection(connectionFactory: (sessionId: string) => Promise<Channel>): Promise<Channel> {
+    protected async createConnection(connectionFactory: (sessionId: string) => Promise<DebugChannel>): Promise<DebugChannel> {
         const connection = await connectionFactory(this.sessionId);
         connection.onClose(() => {
             this.isClosed = true;
             this.cancelPendingRequests();
             this.onDidCloseEmitter.fire();
         });
-        connection.onMessage(data => this.handleMessage(data().readString()));
+        connection.onMessage(data => this.handleMessage(data));
         return connection;
     }
 
@@ -245,7 +247,7 @@ export class DebugSessionConnection implements Disposable {
             const dateStr = `${now.toLocaleString(undefined, { hour12: false })}.${now.getMilliseconds()}`;
             this.traceOutputChannel.appendLine(`${this.sessionId.substring(0, 8)} ${dateStr} theia -> adapter: ${JSON.stringify(message, undefined, 4)}`);
         }
-        connection.getWriteBuffer().writeString(messageStr).commit();
+        connection.send(messageStr);
     }
 
     protected handleMessage(data: string): void {

packages/debug/src/browser/debug-session-contribution.ts

@@ -26,11 +26,10 @@ import { DebugSessionOptions } from './debug-session-options';
 import { OutputChannelManager, OutputChannel } from '@theia/output/lib/browser/output-channel';
 import { DebugPreferences } from './debug-preferences';
 import { DebugSessionConnection } from './debug-session-connection';
-import { DebugAdapterPath } from '../common/debug-service';
+import { DebugChannel, DebugAdapterPath, ForwardingDebugChannel } from '../common/debug-service';
 import { ContributionProvider } from '@theia/core/lib/common/contribution-provider';
 import { FileService } from '@theia/filesystem/lib/browser/file-service';
 import { DebugContribution } from './debug-contribution';
-import { Channel } from '@theia/core/lib/common/message-rpc/channel';
 import { WorkspaceService } from '@theia/workspace/lib/browser';
 
 /**
@@ -122,9 +121,9 @@ export class DefaultDebugSessionFactory implements DebugSessionFactory {
     get(sessionId: string, options: DebugSessionOptions, parentSession?: DebugSession): DebugSession {
         const connection = new DebugSessionConnection(
             sessionId,
-            () => new Promise<Channel>(resolve =>
-                this.connectionProvider.openChannel(`${DebugAdapterPath}/${sessionId}`, channel => {
-                    resolve(channel);
+            () => new Promise<DebugChannel>(resolve =>
+                this.connectionProvider.openChannel(`${DebugAdapterPath}/${sessionId}`, wsChannel => {
+                    resolve(new ForwardingDebugChannel(wsChannel));
                 }, { reconnecting: false })
             ),
             this.getTraceOutputChannel());

packages/debug/src/common/debug-service.ts

@@ -16,7 +16,7 @@
 
 /* eslint-disable @typescript-eslint/no-explicit-any */
 
-import { Disposable, Event } from '@theia/core';
+import { Channel, Disposable, Emitter, Event } from '@theia/core';
 import { ApplicationError } from '@theia/core/lib/common/application-error';
 import { IJSONSchema, IJSONSchemaSnippet } from '@theia/core/lib/common/json-schema';
 import { CommandIdVariables } from '@theia/variable-resolver/lib/common/variable-types';
@@ -142,14 +142,45 @@ export namespace DebugError {
 }
 
 /**
- * A closeable channel to send messages over with error/close handling
+ * A closeable channel to send debug protocol messages over with error/close handling
  */
-export interface Channel {
+export interface DebugChannel {
     send(content: string): void;
-    // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    onMessage(cb: (data: any) => void): void;
+    onMessage(cb: (message: string) => void): void;
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     onError(cb: (reason: any) => void): void;
     onClose(cb: (code: number, reason: string) => void): void;
     close(): void;
 }
+
+/**
+ * A {@link DebugChannel} wrapper implementation that sends and receives messages to/from an underlying {@link Channel}.
+ */
+export class ForwardingDebugChannel implements DebugChannel {
+    private onMessageEmitter = new Emitter<string>();
+
+    constructor(private readonly underlyingChannel: Channel) {
+        this.underlyingChannel.onMessage(msg => this.onMessageEmitter.fire(msg().readString()));
+    }
+
+    send(content: string): void {
+        this.underlyingChannel.getWriteBuffer().writeString(content).commit();
+    }
+
+    onMessage(cb: (message: string) => void): void {
+        this.onMessageEmitter.event(cb);
+    }
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    onError(cb: (reason: any) => void): void {
+        this.underlyingChannel.onError(cb);
+    }
+    onClose(cb: (code: number, reason: string) => void): void {
+        this.underlyingChannel.onClose(event => cb(event.code ?? -1, event.reason));
+    }
+
+    close(): void {
+        this.underlyingChannel.close();
+        this.onMessageEmitter.dispose();
+    }
+
+}

packages/debug/src/node/debug-adapter-session-manager.ts

@@ -18,7 +18,7 @@ import { UUID } from '@theia/core/shared/@phosphor/coreutils';
 import { injectable, inject } from '@theia/core/shared/inversify';
 import { MessagingService } from '@theia/core/lib/node/messaging/messaging-service';
 
-import { DebugAdapterPath } from '../common/debug-service';
+import { DebugAdapterPath, ForwardingDebugChannel } from '../common/debug-service';
 import { DebugConfiguration } from '../common/debug-configuration';
 import { DebugAdapterSession, DebugAdapterSessionFactory, DebugAdapterFactory } from './debug-model';
 import { DebugAdapterContributionRegistry } from './debug-adapter-contribution-registry';
@@ -37,13 +37,13 @@ export class DebugAdapterSessionManager implements MessagingService.Contribution
     protected readonly debugAdapterFactory: DebugAdapterFactory;
 
     configure(service: MessagingService): void {
-        service.wsChannel(`${DebugAdapterPath}/:id`, ({ id }: { id: string }, channel) => {
+        service.wsChannel(`${DebugAdapterPath}/:id`, ({ id }: { id: string }, wsChannel) => {
             const session = this.find(id);
             if (!session) {
-                channel.close();
+                wsChannel.close();
                 return;
             }
-            session.start(channel);
+            session.start(new ForwardingDebugChannel(wsChannel));
         });
     }
 

packages/debug/src/node/debug-adapter-session.ts

@@ -26,14 +26,14 @@ import {
     DebugAdapterSession
 } from './debug-model';
 import { DebugProtocol } from 'vscode-debugprotocol';
-import { Channel } from '@theia/core';
+import { DebugChannel } from '../common/debug-service';
 
 /**
  * [DebugAdapterSession](#DebugAdapterSession) implementation.
  */
 export class DebugAdapterSessionImpl implements DebugAdapterSession {
 
-    private channel: Channel | undefined;
+    private channel: DebugChannel | undefined;
     private isClosed: boolean = false;
 
     constructor(
@@ -46,14 +46,14 @@ export class DebugAdapterSessionImpl implements DebugAdapterSession {
 
     }
 
-    async start(channel: Channel): Promise<void> {
+    async start(channel: DebugChannel): Promise<void> {
 
         console.debug(`starting debug adapter session '${this.id}'`);
         if (this.channel) {
             throw new Error('The session has already been started, id: ' + this.id);
         }
         this.channel = channel;
-        this.channel.onMessage(message => this.write(message().readString()));
+        this.channel.onMessage((message: string) => this.write(message));
         this.channel.onClose(() => this.channel = undefined);
 
     }
@@ -80,7 +80,7 @@ export class DebugAdapterSessionImpl implements DebugAdapterSession {
 
     protected send(message: string): void {
         if (this.channel) {
-            this.channel.getWriteBuffer().writeString(message);
+            this.channel.send(message);
         }
     }
 

packages/debug/src/node/debug-model.ts

@@ -25,7 +25,8 @@
 import { DebugConfiguration } from '../common/debug-configuration';
 import { IJSONSchema, IJSONSchemaSnippet } from '@theia/core/lib/common/json-schema';
 import { MaybePromise } from '@theia/core/lib/common/types';
-import { Channel, Event } from '@theia/core';
+import { Event } from '@theia/core';
+import { DebugChannel } from '../common/debug-service';
 
 // FIXME: break down this file to debug adapter and debug adapter contribution (see Theia file naming conventions)
 
@@ -42,7 +43,7 @@ export const DebugAdapterSession = Symbol('DebugAdapterSession');
 export interface DebugAdapterSession {
     id: string;
     parentSession?: DebugAdapterSession;
-    start(channel: Channel): Promise<void>
+    start(channel: DebugChannel): Promise<void>
     stop(): Promise<void>
 }
 

packages/plugin-ext/src/common/connection.ts

@@ -13,38 +13,27 @@
 //
 // SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
 // *****************************************************************************
+import { DebugChannel } from '@theia/debug/lib/common/debug-service';
 import { ConnectionExt, ConnectionMain } from './plugin-api-rpc';
-import { Emitter, Event } from '@theia/core/lib/common/event';
-import { ChannelCloseEvent, MessageProvider } from '@theia/core/lib/common/message-rpc/channel';
-import { WriteBuffer, Channel } from '@theia/core';
-import { Uint8ArrayReadBuffer, Uint8ArrayWriteBuffer } from '@theia/core/lib/common/message-rpc/uint8-array-message-buffer';
+import { Emitter } from '@theia/core/lib/common/event';
 
 /**
  * A channel communicating with a counterpart in a plugin host.
  */
-export class PluginChannel implements Channel {
-    private messageEmitter: Emitter<MessageProvider> = new Emitter();
+export class PluginChannel implements DebugChannel {
+    private messageEmitter: Emitter<string> = new Emitter();
     private errorEmitter: Emitter<unknown> = new Emitter();
-    private closedEmitter: Emitter<ChannelCloseEvent> = new Emitter();
+    private closedEmitter: Emitter<void> = new Emitter();
 
     constructor(
-        readonly id: string,
+        protected readonly id: string,
         protected readonly connection: ConnectionExt | ConnectionMain) { }
 
-    getWriteBuffer(): WriteBuffer {
-        const result = new Uint8ArrayWriteBuffer();
-        result.onCommit(buffer => {
-            this.connection.$sendMessage(this.id, new Uint8ArrayReadBuffer(buffer).readString());
-        });
-
-        return result;
-    }
-
     send(content: string): void {
         this.connection.$sendMessage(this.id, content);
     }
 
-    fireMessageReceived(msg: MessageProvider): void {
+    fireMessageReceived(msg: string): void {
         this.messageEmitter.fire(msg);
     }
 
@@ -53,19 +42,20 @@ export class PluginChannel implements Channel {
     }
 
     fireClosed(): void {
-        this.closedEmitter.fire({ reason: 'Plugin channel has been closed from the extension side' });
+        this.closedEmitter.fire();
     }
 
-    get onMessage(): Event<MessageProvider> {
-        return this.messageEmitter.event;
+    onMessage(cb: (message: string) => void): void {
+        this.messageEmitter.event(cb);
     }
 
-    get onError(): Event<unknown> {
-        return this.errorEmitter.event;
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    onError(cb: (reason: any) => void): void {
+        this.errorEmitter.event(cb);
     }
 
-    get onClose(): Event<ChannelCloseEvent> {
-        return this.closedEmitter.event;
+    onClose(cb: (code: number, reason: string) => void): void {
+        this.closedEmitter.event(() => cb(-1, 'closed'));
     }
 
     close(): void {
@@ -89,10 +79,7 @@ export class ConnectionImpl implements ConnectionMain, ConnectionExt {
      */
     async $sendMessage(id: string, message: string): Promise<void> {
         if (this.connections.has(id)) {
-            const writer = new Uint8ArrayWriteBuffer().writeString(message);
-            const reader = new Uint8ArrayReadBuffer(writer.getCurrentContents());
-            writer.dispose();
-            this.connections.get(id)!.fireMessageReceived(() => reader);
+            this.connections.get(id)!.fireMessageReceived(message);
         } else {
             console.warn(`Received message for unknown connection: ${id}`);
         }

packages/plugin-ext/src/main/browser/debug/plugin-debug-session-factory.ts

@@ -30,8 +30,8 @@ import { TerminalOptionsExt } from '../../../common/plugin-api-rpc';
 import { FileService } from '@theia/filesystem/lib/browser/file-service';
 import { DebugContribution } from '@theia/debug/lib/browser/debug-contribution';
 import { ContributionProvider } from '@theia/core/lib/common/contribution-provider';
-import { Channel } from '@theia/core/lib/common/message-rpc/channel';
 import { WorkspaceService } from '@theia/workspace/lib/browser';
+import { PluginChannel } from '../../../common/connection';
 
 export class PluginDebugSession extends DebugSession {
     constructor(
@@ -71,7 +71,7 @@ export class PluginDebugSessionFactory extends DefaultDebugSessionFactory {
         protected override readonly messages: MessageClient,
         protected override readonly outputChannelManager: OutputChannelManager,
         protected override readonly debugPreferences: DebugPreferences,
-        protected readonly connectionFactory: (sessionId: string) => Promise<Channel>,
+        protected readonly connectionFactory: (sessionId: string) => Promise<PluginChannel>,
         protected override readonly fileService: FileService,
         protected readonly terminalOptionsExt: TerminalOptionsExt | undefined,
         protected override readonly debugContributionProvider: ContributionProvider<DebugContribution>,

packages/plugin-ext/src/plugin/node/debug/plugin-debug-adapter-session.ts

@@ -17,7 +17,7 @@
 import { DebugAdapterSessionImpl } from '@theia/debug/lib/node/debug-adapter-session';
 import * as theia from '@theia/plugin';
 import { DebugAdapter } from '@theia/debug/lib/node/debug-model';
-import { Channel } from '@theia/core/lib/common/message-rpc/channel';
+import { DebugChannel } from '@theia/debug/lib/common/debug-service';
 
 /* eslint-disable @typescript-eslint/no-explicit-any */
 
@@ -43,7 +43,7 @@ export class PluginDebugAdapterSession extends DebugAdapterSessionImpl {
         this.configuration = theiaSession.configuration;
     }
 
-    override async start(channel: Channel): Promise<void> {
+    override async start(channel: DebugChannel): Promise<void> {
         if (this.tracker.onWillStartSession) {
             this.tracker.onWillStartSession();
         }