Add DebugAdapterNamedPipeServer and DebugAdapterInlineImplementation

packages/debug/src/node/debug-adapter-contribution-registry.ts

@@ -19,7 +19,7 @@ import { ContributionProvider } from '@theia/core';
 import { DebugConfiguration } from '../common/debug-configuration';
 import { DebuggerDescription, DebugError } from '../common/debug-service';
 
-import { DebugAdapterContribution, DebugAdapterExecutable, DebugAdapterSessionFactory } from '../common/debug-model';
+import { DebugAdapterContribution, DebugAdapterExecutable, DebugAdapterSessionFactory } from './debug-model';
 import { IJSONSchema, IJSONSchemaSnippet } from '@theia/core/lib/common/json-schema';
 
 /**

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

@@ -37,9 +37,11 @@ import {
     DebugAdapterSessionFactory,
     DebugAdapterFactory,
     DebugAdapterForkExecutable
-} from '../common/debug-model';
+} from './debug-model';
 import { DebugAdapterSessionImpl } from './debug-adapter-session';
 import { environment } from '@theia/core/shared/@theia/application-package';
+import { StreamCommunicationProvider } from './stream-communication-provider';
+import { Disposable } from '@theia/core/lib/common/disposable';
 
 /**
  * [DebugAdapterFactory](#DebugAdapterFactory) implementation based on
@@ -56,11 +58,9 @@ export class LaunchBasedDebugAdapterFactory implements DebugAdapterFactory {
         const process = this.childProcess(executable);
 
         // FIXME: propagate onError + onExit
-        return {
-            input: process.inputStream,
-            output: process.outputStream,
-            dispose: () => process.kill()
-        };
+        const provider = new StreamCommunicationProvider(process.outputStream, process.inputStream);
+        provider.push(Disposable.create(() => process.kill()));
+        return provider;
     }
 
     private childProcess(executable: DebugAdapterExecutable): RawProcess {
@@ -84,11 +84,10 @@ export class LaunchBasedDebugAdapterFactory implements DebugAdapterFactory {
     connect(debugServerPort: number): CommunicationProvider {
         const socket = net.createConnection(debugServerPort);
         // FIXME: propagate socket.on('error', ...) + socket.on('close', ...)
-        return {
-            input: socket,
-            output: socket,
-            dispose: () => socket.end()
-        };
+
+        const provider = new StreamCommunicationProvider(socket, socket);
+        provider.push(Disposable.create(() => socket.end()));
+        return provider;
     }
 }
 

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

@@ -20,7 +20,7 @@ import { MessagingService } from '@theia/core/lib/node/messaging/messaging-servi
 
 import { DebugAdapterPath } from '../common/debug-service';
 import { DebugConfiguration } from '../common/debug-configuration';
-import { DebugAdapterSession, DebugAdapterSessionFactory, DebugAdapterFactory } from '../common/debug-model';
+import { DebugAdapterSession, DebugAdapterSessionFactory, DebugAdapterFactory } from './debug-model';
 import { DebugAdapterContributionRegistry } from './debug-adapter-contribution-registry';
 
 /**

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

@@ -23,8 +23,8 @@
 
 import {
     CommunicationProvider,
-    DebugAdapterSession,
-} from '../common/debug-model';
+    DebugAdapterSession
+} from './debug-model';
 import { DebugProtocol } from 'vscode-debugprotocol';
 import { IWebSocket } from '@theia/core/shared/vscode-ws-jsonrpc';
 import { DisposableCollection, Disposable } from '@theia/core/lib/common/disposable';
@@ -34,25 +34,23 @@ import { DisposableCollection, Disposable } from '@theia/core/lib/common/disposa
  */
 export class DebugAdapterSessionImpl implements DebugAdapterSession {
 
-    private static TWO_CRLF = '\r\n\r\n';
-    private static CONTENT_LENGTH = 'Content-Length';
-
     private readonly toDispose = new DisposableCollection();
     private channel: IWebSocket | undefined;
-    private contentLength: number;
-    private buffer: Buffer;
 
     constructor(
         readonly id: string,
         protected readonly communicationProvider: CommunicationProvider
     ) {
-        this.contentLength = -1;
-        this.buffer = Buffer.alloc(0);
         this.toDispose.pushAll([
             this.communicationProvider,
             Disposable.create(() => this.write(JSON.stringify({ seq: -1, type: 'request', command: 'disconnect' }))),
             Disposable.create(() => this.write(JSON.stringify({ seq: -1, type: 'request', command: 'terminate' })))
         ]);
+
+        this.communicationProvider.onMessageReceived((message: string) => this.send(message));
+        this.communicationProvider.onClose(() => this.onDebugAdapterExit(1, undefined)); // FIXME pass a proper exit code
+        this.communicationProvider.onError(error => this.onDebugAdapterError(error));
+
     }
 
     async start(channel: IWebSocket): Promise<void> {
@@ -63,10 +61,6 @@ export class DebugAdapterSessionImpl implements DebugAdapterSession {
         this.channel.onMessage((message: string) => this.write(message));
         this.channel.onClose(() => this.channel = undefined);
 
-        this.communicationProvider.output.on('data', (data: Buffer) => this.handleData(data));
-        this.communicationProvider.output.on('close', () => this.onDebugAdapterExit(1, undefined)); // FIXME pass a proper exit code
-        this.communicationProvider.output.on('error', error => this.onDebugAdapterError(error));
-        this.communicationProvider.input.on('error', error => this.onDebugAdapterError(error));
     }
 
     protected onDebugAdapterExit(exitCode: number, signal: string | undefined): void {
@@ -91,57 +85,14 @@ export class DebugAdapterSessionImpl implements DebugAdapterSession {
         this.send(JSON.stringify(event));
     }
 
-    protected handleData(data: Buffer): void {
-        this.buffer = Buffer.concat([this.buffer, data]);
-
-        while (true) {
-            if (this.contentLength >= 0) {
-                if (this.buffer.length >= this.contentLength) {
-                    const message = this.buffer.toString('utf8', 0, this.contentLength);
-                    this.buffer = this.buffer.slice(this.contentLength);
-                    this.contentLength = -1;
-
-                    if (message.length > 0) {
-                        this.send(message);
-                    }
-                    continue; // there may be more complete messages to process
-                }
-            } else {
-                let idx = this.buffer.indexOf(DebugAdapterSessionImpl.CONTENT_LENGTH);
-                if (idx > 0) {
-                    // log unrecognized output
-                    const output = this.buffer.slice(0, idx);
-                    console.log(output.toString('utf-8'));
-
-                    this.buffer = this.buffer.slice(idx);
-                }
-
-                idx = this.buffer.indexOf(DebugAdapterSessionImpl.TWO_CRLF);
-                if (idx !== -1) {
-                    const header = this.buffer.toString('utf8', 0, idx);
-                    const lines = header.split('\r\n');
-                    for (let i = 0; i < lines.length; i++) {
-                        const pair = lines[i].split(/: +/);
-                        if (pair[0] === DebugAdapterSessionImpl.CONTENT_LENGTH) {
-                            this.contentLength = +pair[1];
-                        }
-                    }
-                    this.buffer = this.buffer.slice(idx + DebugAdapterSessionImpl.TWO_CRLF.length);
-                    continue;
-                }
-            }
-            break;
-        }
-    }
-
     protected send(message: string): void {
         if (this.channel) {
             this.channel.send(message);
         }
     }
 
     protected write(message: string): void {
-        this.communicationProvider.input.write(`Content-Length: ${Buffer.byteLength(message, 'utf8')}\r\n\r\n${message}`, 'utf8');
+        this.communicationProvider.send(message);
     }
 
     async stop(): Promise<void> {

packages/debug/src/node/debug-backend-module.ts

@@ -30,7 +30,7 @@ import {
     DebugAdapterContribution,
     DebugAdapterSessionFactory,
     DebugAdapterFactory
-} from '../common/debug-model';
+} from './debug-model';
 import { DebugServiceImpl } from './debug-service-impl';
 import { DebugAdapterContributionRegistry } from './debug-adapter-contribution-registry';
 import { DebugAdapterSessionManager } from './debug-adapter-session-manager';

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

@@ -22,12 +22,12 @@
 // Some entities copied and modified from https://github.com/Microsoft/vscode/blob/master/src/vs/vscode.d.ts
 // Some entities copied and modified from https://github.com/Microsoft/vscode/blob/master/src/vs/workbench/parts/debug/common/debug.ts
 
-import * as stream from 'stream';
 import { WebSocketChannel } from '@theia/core/lib/common/messaging/web-socket-channel';
-import { DebugConfiguration } from './debug-configuration';
+import { DebugConfiguration } from '../common/debug-configuration';
 import { IJSONSchema, IJSONSchemaSnippet } from '@theia/core/lib/common/json-schema';
 import { Disposable } from '@theia/core/lib/common/disposable';
 import { MaybePromise } from '@theia/core/lib/common/types';
+import { Event } from '@theia/core/lib/common/event';
 
 // FIXME: break down this file to debug adapter and debug adapter contribution (see Theia file naming conventions)
 
@@ -96,8 +96,10 @@ export type DebugAdapterExecutable = DebugAdapterSpawnExecutable | DebugAdapterF
  * TODO: the better name is DebugStreamConnection + handling on error and close
  */
 export interface CommunicationProvider extends Disposable {
-    output: stream.Readable;
-    input: stream.Writable;
+    onMessageReceived: Event<string>;
+    onError: Event<Error>;
+    onClose: Event<void>;
+    send(message: string): void;
 }
 
 /**

packages/debug/src/node/inline-communication-provider.ts

@@ -0,0 +1,44 @@
+/********************************************************************************
+ * Copyright (C) 2021 Red Hat, 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 { Emitter, Event } from '@theia/core/lib/common/event';
+import { CommunicationProvider } from './debug-model';
+import * as theia from '@theia/plugin';
+
+/**
+ * A communication provider for using the inline implementation of a debug adapter.
+ */
+export class InlineCommunicationProvider implements CommunicationProvider {
+    private messageReceivedEmitter = new Emitter<string>();
+    onMessageReceived: Event<string> = this.messageReceivedEmitter.event;
+    onError: Event<Error> = Event.None;
+    private closeEmitter = new Emitter<void>();
+    onClose: Event<void> = this.closeEmitter.event;
+
+    constructor(private debugAdapter: theia.DebugAdapter) {
+        this.debugAdapter.onDidSendMessage(msg => {
+            this.messageReceivedEmitter.fire(JSON.stringify(msg));
+        });
+    }
+
+    send(message: string): void {
+        this.debugAdapter.handleMessage(JSON.parse(message));
+    }
+
+    dispose(): void {
+        this.debugAdapter.dispose();
+    }
+}

packages/debug/src/node/stream-communication-provider.ts

@@ -0,0 +1,93 @@
+/********************************************************************************
+ * Copyright (C) 2021 Red Hat, 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 { DisposableCollection } from '@theia/core/lib/common/disposable';
+import { Emitter, Event } from '@theia/core/lib/common/event';
+import * as stream from 'stream';
+import { CommunicationProvider } from './debug-model';
+
+export class StreamCommunicationProvider extends DisposableCollection implements CommunicationProvider {
+    private messageReceivedEmitter = new Emitter<string>();
+    onMessageReceived: Event<string> = this.messageReceivedEmitter.event;
+    private errorEmitter = new Emitter<Error>();
+    onError: Event<Error> = this.errorEmitter.event;
+    private closeEmitter = new Emitter<void>();
+    onClose: Event<void> = this.closeEmitter.event;
+
+    // these constants are for the message header, see: https://microsoft.github.io/debug-adapter-protocol/overview#header-part
+    private static TWO_CRLF = '\r\n\r\n';
+    private static CONTENT_LENGTH = 'Content-Length';
+    private contentLength: number = -1;
+    private buffer: Buffer = Buffer.alloc(0);
+
+    constructor(private fromAdapter: stream.Readable, private toAdapter: stream.Writable) {
+        super();
+
+        this.fromAdapter.on('data', (data: Buffer) => this.handleData(data));
+        this.fromAdapter.on('close', () => this.closeEmitter.fire()); // FIXME pass a proper exit code
+        this.fromAdapter.on('error', error => this.errorEmitter.fire(error));
+        this.toAdapter.on('error', error => this.errorEmitter.fire(error));
+    };
+
+    send(message: string): void {
+        const msg = `${StreamCommunicationProvider.CONTENT_LENGTH}: ${Buffer.byteLength(message, 'utf8')}${StreamCommunicationProvider.TWO_CRLF}${message}`;
+
+        this.toAdapter.write(msg, 'utf8');
+    }
+
+    protected handleData(data: Buffer): void {
+        this.buffer = Buffer.concat([this.buffer, data]);
+
+        while (true) {
+            if (this.contentLength >= 0) {
+                if (this.buffer.length >= this.contentLength) {
+                    const message = this.buffer.toString('utf8', 0, this.contentLength);
+                    this.buffer = this.buffer.slice(this.contentLength);
+                    this.contentLength = -1;
+
+                    if (message.length > 0) {
+                        this.messageReceivedEmitter.fire(message);
+                    }
+                    continue; // there may be more complete messages to process
+                }
+            } else {
+                let idx = this.buffer.indexOf(StreamCommunicationProvider.CONTENT_LENGTH);
+                if (idx > 0) {
+                    // log unrecognized output
+                    const output = this.buffer.slice(0, idx);
+                    console.log(output.toString('utf-8'));
+
+                    this.buffer = this.buffer.slice(idx);
+                }
+
+                idx = this.buffer.indexOf(StreamCommunicationProvider.TWO_CRLF);
+                if (idx !== -1) {
+                    const header = this.buffer.toString('utf8', 0, idx);
+                    const lines = header.split('\r\n');
+                    for (let i = 0; i < lines.length; i++) {
+                        const pair = lines[i].split(/: +/);
+                        if (pair[0] === StreamCommunicationProvider.CONTENT_LENGTH) {
+                            this.contentLength = +pair[1];
+                        }
+                    }
+                    this.buffer = this.buffer.slice(idx + StreamCommunicationProvider.TWO_CRLF.length);
+                    continue;
+                }
+            }
+            break;
+        }
+    }
+}

packages/debug/src/node/vscode/vscode-debug-adapter-contribution.ts

@@ -16,7 +16,7 @@
 
 import * as fs from '@theia/core/shared/fs-extra';
 import * as path from 'path';
-import { DebugAdapterExecutable, DebugAdapterContribution } from '../../common/debug-model';
+import { DebugAdapterExecutable, DebugAdapterContribution } from '../debug-model';
 import { isWindows, isOSX } from '@theia/core/lib/common/os';
 import { IJSONSchema, IJSONSchemaSnippet } from '@theia/core/lib/common/json-schema';
 import { deepClone } from '@theia/core/lib/common/objects';

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

@@ -30,7 +30,7 @@ import { TerminalWidgetOptions, TerminalWidget } from '@theia/terminal/lib/brows
 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';
+import { ContributionProvider } from '@theia/core/lib/common/contribution-provider';
 
 export class PluginDebugSession extends DebugSession {
     constructor(