[dds] Add a basic DAP server that can run simple Dart scripts

Change-Id: I0f10b81b0b9ad3875727f606dd1a5a44798486b1
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/201263
Reviewed-by: Brian Wilkerson <brianwilkerson@google.com>
Reviewed-by: Ben Konyi <bkonyi@google.com>
Commit-Queue: Ben Konyi <bkonyi@google.com>

pkg/dds/lib/src/dap/adapters/dart.dart

@@ -0,0 +1,283 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+
+import '../base_debug_adapter.dart';
+import '../logging.dart';
+import '../protocol_generated.dart';
+import '../protocol_stream.dart';
+
+/// A base DAP Debug Adapter implementation for running and debugging Dart-based
+/// applications (including Flutter and Tests).
+///
+/// This class implements all functionality common to Dart, Flutter and Test
+/// debug sessions, including things like breakpoints and expression eval.
+///
+/// Sub-classes should handle the launching/attaching of apps and any custom
+/// behaviour (such as Flutter's Hot Reload). This is generally done by overriding
+/// `fooImpl` methods that are called during the handling of a `fooRequest` from
+/// the client.
+///
+/// A DebugAdapter instance will be created per application being debugged (in
+/// multi-session mode, one DebugAdapter corresponds to one incoming TCP
+/// connection, though a client may make multiple of these connections if it
+/// wants to debug multiple scripts concurrently, such as with a compound launch
+/// configuration in VS Code).
+///
+/// The lifecycle is described in the DAP spec here:
+/// https://microsoft.github.io/debug-adapter-protocol/overview#initialization
+///
+/// In summary:
+///
+/// The client will create a connection to the server (which will create an
+///   instance of the debug adapter) and send an `initializeRequest` message,
+///   wait for the server to return a response and then an initializedEvent
+/// The client will then send breakpoints and exception config
+///   (`setBreakpointsRequest`, `setExceptionBreakpoints`) and then a
+///   `configurationDoneRequest`.
+/// Finally, the client will send a `launchRequest` or `attachRequest` to start
+///   running/attaching to the script.
+///
+/// The client will continue to send requests during the debug session that may
+/// be in response to user actions (for example changing breakpoints or typing
+/// an expression into an evaluation console) or to events sent by the server
+/// (for example when the server sends a `StoppedEvent` it may cause the client
+/// to then send a `stackTraceRequest` or `scopesRequest` to get variables).
+abstract class DartDebugAdapter<T extends DartLaunchRequestArguments>
+    extends BaseDebugAdapter<T> {
+  late T args;
+  final _debuggerInitializedCompleter = Completer<void>();
+  final _configurationDoneCompleter = Completer<void>();
+
+  DartDebugAdapter(ByteStreamServerChannel channel, Logger? logger)
+      : super(channel, logger);
+
+  /// Completes when the debugger initialization has completed. Used to delay
+  /// processing isolate events while initialization is still running to avoid
+  /// race conditions (for example if an isolate unpauses before we have
+  /// processed its initial paused state).
+  Future<void> get debuggerInitialized => _debuggerInitializedCompleter.future;
+
+  /// configurationDone is called by the client when it has finished sending
+  /// any initial configuration (such as breakpoints and exception pause
+  /// settings).
+  ///
+  /// We delay processing `launchRequest`/`attachRequest` until this request has
+  /// been sent to ensure we're not still getting breakpoints (which are sent
+  /// per-file) while we're launching and initializing over the VM Service.
+  @override
+  FutureOr<void> configurationDoneRequest(
+    Request request,
+    ConfigurationDoneArguments? args,
+    void Function(void) sendResponse,
+  ) async {
+    _configurationDoneCompleter.complete();
+    sendResponse(null);
+  }
+
+  /// Overridden by sub-classes to handle when the client sends a
+  /// `disconnectRequest` (a forceful request to shut down).
+  FutureOr<void> disconnectImpl();
+
+  /// disconnectRequest is called by the client when it wants to forcefully shut
+  /// us down quickly. This comes after the `terminateRequest` which is intended
+  /// to allow a graceful shutdown.
+  ///
+  /// It's not very obvious from the names, but `terminateRequest` is sent first
+  /// (a request for a graceful shutdown) and `disconnectRequest` second (a
+  /// request for a forced shutdown).
+  ///
+  /// https://microsoft.github.io/debug-adapter-protocol/overview#debug-session-end
+  @override
+  FutureOr<void> disconnectRequest(
+    Request request,
+    DisconnectArguments? args,
+    void Function(void) sendResponse,
+  ) async {
+    await disconnectImpl();
+    sendResponse(null);
+  }
+
+  /// initializeRequest is the first request send by the client during
+  /// initialization and allows exchanging capabilities and configuration
+  /// between client and server.
+  ///
+  /// The lifecycle is described in the DAP spec here:
+  /// https://microsoft.github.io/debug-adapter-protocol/overview#initialization
+  /// with a summary in this classes description.
+  @override
+  FutureOr<void> initializeRequest(
+    Request request,
+    InitializeRequestArguments? args,
+    void Function(Capabilities) sendResponse,
+  ) async {
+    // TODO(dantup): Capture/honor editor-specific settings like linesStartAt1
+    sendResponse(Capabilities(
+      exceptionBreakpointFilters: [
+        ExceptionBreakpointsFilter(
+          filter: 'All',
+          label: 'All Exceptions',
+          defaultValue: false,
+        ),
+        ExceptionBreakpointsFilter(
+          filter: 'Unhandled',
+          label: 'Uncaught Exceptions',
+          defaultValue: true,
+        ),
+      ],
+      supportsClipboardContext: true,
+      // TODO(dantup): All of these...
+      // supportsConditionalBreakpoints: true,
+      supportsConfigurationDoneRequest: true,
+      supportsDelayedStackTraceLoading: true,
+      supportsEvaluateForHovers: true,
+      // supportsLogPoints: true,
+      // supportsRestartFrame: true,
+      supportsTerminateRequest: true,
+    ));
+
+    // This must only be sent AFTER the response!
+    sendEvent(InitializedEventBody());
+  }
+
+  /// Overridden by sub-classes to handle when the client sends a
+  /// `launchRequest` (a request to start running/debugging an app).
+  ///
+  /// Sub-classes can use the [args] field to access the arguments provided
+  /// to this request.
+  FutureOr<void> launchImpl();
+
+  /// launchRequest is called by the client when it wants us to to start the app
+  /// to be run/debug. This will only be called once (and only one of this or
+  /// attachRequest will be called).
+  @override
+  FutureOr<void> launchRequest(
+    Request request,
+    T args,
+    void Function(void) sendResponse,
+  ) async {
+    this.args = args;
+
+    // Don't start launching until configurationDone.
+    if (!_configurationDoneCompleter.isCompleted) {
+      logger?.call('Waiting for configurationDone request...');
+      await _configurationDoneCompleter.future;
+    }
+
+    // Delegate to the sub-class to launch the process.
+    await launchImpl();
+
+    sendResponse(null);
+  }
+
+  /// Sends an OutputEvent (without a newline, since calls to this method
+  /// may be used by buffered data).
+  void sendOutput(String category, String message) {
+    sendEvent(OutputEventBody(category: category, output: message));
+  }
+
+  /// Overridden by sub-classes to handle when the client sends a
+  /// `terminateRequest` (a request for a graceful shut down).
+  FutureOr<void> terminateImpl();
+
+  /// terminateRequest is called by the client when it wants us to gracefully
+  /// shut down.
+  ///
+  /// It's not very obvious from the names, but `terminateRequest` is sent first
+  /// (a request for a graceful shutdown) and `disconnectRequest` second (a
+  /// request for a forced shutdown).
+  ///
+  /// https://microsoft.github.io/debug-adapter-protocol/overview#debug-session-end
+  @override
+  FutureOr<void> terminateRequest(
+    Request request,
+    TerminateArguments? args,
+    void Function(void) sendResponse,
+  ) async {
+    terminateImpl();
+    sendResponse(null);
+  }
+}
+
+/// An implementation of [LaunchRequestArguments] that includes all fields used
+/// by the base Dart debug adapter.
+///
+/// This class represents the data passed from the client editor to the debug
+/// adapter in launchRequest, which is a request to start debugging an
+/// application.
+///
+/// Specialised adapters (such as Flutter) will likely extend this class with
+/// their own additional fields.
+class DartLaunchRequestArguments extends LaunchRequestArguments {
+  final String program;
+  final List<String>? args;
+  final String? cwd;
+  final String? vmServiceInfoFile;
+  final int? vmServicePort;
+  final List<String>? vmAdditionalArgs;
+  final bool? enableAsserts;
+  final bool? debugSdkLibraries;
+  final bool? evaluateGettersInDebugViews;
+  final bool? evaluateToStringInDebugViews;
+
+  /// Whether to send debug logging to clients in a custom `dart.log` event. This
+  /// is used both by the out-of-process tests to ensure the logs contain enough
+  /// information to track down issues, but also by Dart-Code to capture VM
+  /// service traffic in a unified log file.
+  final bool? sendLogsToClient;
+
+  DartLaunchRequestArguments({
+    Object? restart,
+    bool? noDebug,
+    required this.program,
+    this.args,
+    this.cwd,
+    this.vmServiceInfoFile,
+    this.vmServicePort,
+    this.vmAdditionalArgs,
+    this.enableAsserts,
+    this.debugSdkLibraries,
+    this.evaluateGettersInDebugViews,
+    this.evaluateToStringInDebugViews,
+    this.sendLogsToClient,
+  }) : super(restart: restart, noDebug: noDebug);
+
+  DartLaunchRequestArguments.fromMap(Map<String, Object?> obj)
+      : program = obj['program'] as String,
+        args = (obj['args'] as List?)?.cast<String>(),
+        cwd = obj['cwd'] as String?,
+        vmServiceInfoFile = obj['vmServiceInfoFile'] as String?,
+        vmServicePort = obj['vmServicePort'] as int?,
+        vmAdditionalArgs = (obj['vmAdditionalArgs'] as List?)?.cast<String>(),
+        enableAsserts = obj['enableAsserts'] as bool?,
+        debugSdkLibraries = obj['debugSdkLibraries'] as bool?,
+        evaluateGettersInDebugViews =
+            obj['evaluateGettersInDebugViews'] as bool?,
+        evaluateToStringInDebugViews =
+            obj['evaluateToStringInDebugViews'] as bool?,
+        sendLogsToClient = obj['sendLogsToClient'] as bool?,
+        super.fromMap(obj);
+
+  @override
+  Map<String, Object?> toJson() => {
+        ...super.toJson(),
+        'program': program,
+        if (args != null) 'args': args,
+        if (cwd != null) 'cwd': cwd,
+        if (vmServiceInfoFile != null) 'vmServiceInfoFile': vmServiceInfoFile,
+        if (vmServicePort != null) 'vmServicePort': vmServicePort,
+        if (vmAdditionalArgs != null) 'vmAdditionalArgs': vmAdditionalArgs,
+        if (enableAsserts != null) 'enableAsserts': enableAsserts,
+        if (debugSdkLibraries != null) 'debugSdkLibraries': debugSdkLibraries,
+        if (evaluateGettersInDebugViews != null)
+          'evaluateGettersInDebugViews': evaluateGettersInDebugViews,
+        if (evaluateToStringInDebugViews != null)
+          'evaluateToStringInDebugViews': evaluateToStringInDebugViews,
+        if (sendLogsToClient != null) 'sendLogsToClient': sendLogsToClient,
+      };
+
+  static DartLaunchRequestArguments fromJson(Map<String, Object?> obj) =>
+      DartLaunchRequestArguments.fromMap(obj);
+}

pkg/dds/lib/src/dap/adapters/dart_cli.dart

@@ -0,0 +1,85 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:pedantic/pedantic.dart';
+
+import '../logging.dart';
+import '../protocol_generated.dart';
+import '../protocol_stream.dart';
+import 'dart.dart';
+
+/// A DAP Debug Adapter for running and debugging Dart CLI scripts.
+class DartCliDebugAdapter extends DartDebugAdapter<DartLaunchRequestArguments> {
+  Process? _process;
+
+  @override
+  final parseLaunchArgs = DartLaunchRequestArguments.fromJson;
+
+  DartCliDebugAdapter(ByteStreamServerChannel channel, [Logger? logger])
+      : super(channel, logger);
+
+  /// Called by [disconnectRequest] to request that we forcefully shut down the
+  /// app being run (or in the case of an attach, disconnect).
+  FutureOr<void> disconnectImpl() {
+    // TODO(dantup): In Dart-Code DAP, we first try again with SIGINT and wait
+    // for a few seconds before sending sigkill.
+    _process?.kill(ProcessSignal.sigkill);
+  }
+
+  /// Called by [launchRequest] to request that we actually start the app to be
+  /// run/debugged.
+  ///
+  /// For debugging, this should start paused, connect to the VM Service, set
+  /// breakpoints, and resume.
+  Future<void> launchImpl() async {
+    final vmPath = Platform.resolvedExecutable;
+    final vmArgs = <String>[]; // TODO(dantup): enable-asserts, debug, etc.
+
+    final processArgs = [
+      ...vmArgs,
+      args.program,
+      ...?args.args,
+    ];
+
+    logger?.call('Spawning $vmPath with $processArgs in ${args.cwd}');
+    final process = await Process.start(
+      vmPath,
+      processArgs,
+      workingDirectory: args.cwd,
+    );
+    _process = process;
+
+    process.stdout.listen(_handleStdout);
+    process.stderr.listen(_handleStderr);
+    unawaited(process.exitCode.then(_handleExitCode));
+  }
+
+  /// Called by [terminateRequest] to request that we gracefully shut down the
+  /// app being run (or in the case of an attach, disconnect).
+  FutureOr<void> terminateImpl() async {
+    _process?.kill(ProcessSignal.sigint);
+    await _process?.exitCode;
+  }
+
+  void _handleExitCode(int code) {
+    final codeSuffix = code == 0 ? '' : ' ($code)';
+    logger?.call('Process exited ($code)');
+    // Always add a leading newline since the last written text might not have
+    // had one.
+    sendOutput('console', '\nExited$codeSuffix.');
+    sendEvent(TerminatedEventBody());
+  }
+
+  void _handleStderr(List<int> data) {
+    sendOutput('stderr', utf8.decode(data));
+  }
+
+  void _handleStdout(List<int> data) {
+    sendOutput('stdout', utf8.decode(data));
+  }
+}

pkg/dds/lib/src/dap/base_debug_adapter.dart

@@ -0,0 +1,198 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+
+import 'logging.dart';
+import 'protocol_common.dart';
+import 'protocol_generated.dart';
+import 'protocol_stream.dart';
+
+/// A base class for debug adapters.
+///
+/// Communicates over a [ByteStreamServerChannel] and turns messages into
+/// appropriate method calls/events.
+///
+/// This class does not implement any DA functionality, only message handling.
+abstract class BaseDebugAdapter<TLaunchArgs extends LaunchRequestArguments> {
+  int _sequence = 1;
+  final ByteStreamServerChannel _channel;
+  final Logger? logger;
+
+  BaseDebugAdapter(this._channel, this.logger) {
+    _channel.listen(_handleIncomingMessage);
+  }
+
+  /// Parses arguments for [launchRequest] into a type of [TLaunchArgs].
+  ///
+  /// This method must be implemented by the implementing class using a class
+  /// that corresponds to the arguments it expects (these may differ between
+  /// Dart CLI, Dart tests, Flutter, Flutter tests).
+  TLaunchArgs Function(Map<String, Object?>) get parseLaunchArgs;
+
+  FutureOr<void> configurationDoneRequest(
+    Request request,
+    ConfigurationDoneArguments? args,
+    void Function(void) sendResponse,
+  );
+
+  FutureOr<void> disconnectRequest(
+    Request request,
+    DisconnectArguments? args,
+    void Function(void) sendResponse,
+  );
+
+  /// Calls [handler] for an incoming request, using [fromJson] to parse its
+  /// arguments from the request.
+  ///
+  /// [handler] will be provided a function [sendResponse] that it can use to
+  /// sends its response without needing to build a [Response] from fields on
+  /// the request.
+  ///
+  /// [handler] must _always_ call [sendResponse], even if the response does not
+  /// require a body.
+  ///
+  /// If [handler] throws, its exception will be sent as an error response.
+  FutureOr<void> handle<TArg, TResp>(
+    Request request,
+    FutureOr<void> Function(Request, TArg, void Function(TResp)) handler,
+    TArg Function(Map<String, Object?>) fromJson,
+  ) async {
+    final args = request.arguments != null
+        ? fromJson(request.arguments as Map<String, Object?>)
+        // arguments are only valid to be null then TArg is nullable.
+        : null as TArg;
+
+    // Because handlers may need to send responses before they have finished
+    // executing (for example, initializeRequest needs to send its response
+    // before sending InitializedEvent()), we pass in a function `sendResponse`
+    // rather than using a return value.
+    var sendResponseCalled = false;
+    void sendResponse(TResp responseBody) {
+      assert(!sendResponseCalled,
+          'sendResponse was called multiple times by ${request.command}');
+      sendResponseCalled = true;
+      final response = Response(
+        success: true,
+        requestSeq: request.seq,
+        seq: _sequence++,
+        command: request.command,
+        body: responseBody,
+      );
+      _channel.sendResponse(response);
+    }
+
+    try {
+      await handler(request, args, sendResponse);
+      assert(sendResponseCalled,
+          'sendResponse was not called in ${request.command}');
+    } catch (e, s) {
+      final response = Response(
+        success: false,
+        requestSeq: request.seq,
+        seq: _sequence++,
+        command: request.command,
+        message: '$e',
+        body: '$s',
+      );
+      _channel.sendResponse(response);
+    }
+  }
+
+  FutureOr<void> initializeRequest(
+    Request request,
+    InitializeRequestArguments args,
+    void Function(Capabilities) sendResponse,
+  );
+
+  FutureOr<void> launchRequest(
+      Request request, TLaunchArgs args, void Function(void) sendResponse);
+
+  /// Sends an event, lookup up the event type based on the runtimeType of
+  /// [body].
+  void sendEvent(EventBody body) {
+    final event = Event(
+      seq: _sequence++,
+      event: eventTypes[body.runtimeType]!,
+      body: body,
+    );
+    _channel.sendEvent(event);
+  }
+
+  /// Sends a request to the client, looking up the request type based on the
+  /// runtimeType of [arguments].
+  void sendRequest(RequestArguments arguments) {
+    final request = Request(
+      seq: _sequence++,
+      command: commandTypes[arguments.runtimeType]!,
+      arguments: arguments,
+    );
+    _channel.sendRequest(request);
+  }
+
+  FutureOr<void> terminateRequest(
+    Request request,
+    TerminateArguments? args,
+    void Function(void) sendResponse,
+  );
+
+  /// Wraps a fromJson handler for requests that allow null arguments.
+  T? Function(Map<String, Object?>?) _allowNullArg<T extends RequestArguments>(
+    T Function(Map<String, Object?>) fromJson,
+  ) {
+    return (data) => data == null ? null : fromJson(data);
+  }
+
+  /// Handles incoming messages from the client editor.
+  void _handleIncomingMessage(ProtocolMessage message) {
+    if (message is Request) {
+      _handleIncomingRequest(message);
+    } else if (message is Response) {
+      _handleIncomingResponse(message);
+    } else {
+      throw Exception('Unknown Protocol message ${message.type}');
+    }
+  }
+
+  /// Handles an incoming request, calling the appropriate method to handle it.
+  void _handleIncomingRequest(Request request) {
+    if (request.command == 'initialize') {
+      handle(request, initializeRequest, InitializeRequestArguments.fromJson);
+    } else if (request.command == 'launch') {
+      handle(request, launchRequest, parseLaunchArgs);
+    } else if (request.command == 'terminate') {
+      handle(
+        request,
+        terminateRequest,
+        _allowNullArg(TerminateArguments.fromJson),
+      );
+    } else if (request.command == 'disconnect') {
+      handle(
+        request,
+        disconnectRequest,
+        _allowNullArg(DisconnectArguments.fromJson),
+      );
+    } else if (request.command == 'configurationDone') {
+      handle(
+        request,
+        configurationDoneRequest,
+        _allowNullArg(ConfigurationDoneArguments.fromJson),
+      );
+    } else {
+      final response = Response(
+        success: false,
+        requestSeq: request.seq,
+        seq: _sequence++,
+        command: request.command,
+        message: 'Unknown command: ${request.command}',
+      );
+      _channel.sendResponse(response);
+    }
+  }
+
+  void _handleIncomingResponse(Response response) {
+    // TODO(dantup): Implement this when the server sends requests to the client
+    // (for example runInTerminalRequest).
+  }
+}

pkg/dds/lib/src/dap/logging.dart

@@ -0,0 +1,5 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+typedef Logger = void Function(String);

pkg/dds/lib/src/dap/protocol_stream.dart

@@ -0,0 +1,126 @@
+// Copyright (c) 2021, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:convert';
+
+import 'logging.dart';
+import 'protocol_generated.dart';
+import 'protocol_stream_transformers.dart';
+
+// TODO(dantup): This class should mostly be shareable with the LSP version,
+// but the ProtocolMessage, Request, Response, Event classes are different so
+// will need specializations.
+
+/// A wrapper over a Stream/StreamSink that encodes/decores DAP/LSP
+/// request/response/event messages.
+class ByteStreamServerChannel {
+  final Stream<List<int>> _input;
+
+  final StreamSink<List<int>> _output;
+
+  final Logger? _logger;
+
+  /// Completer that will be signalled when the input stream is closed.
+  final Completer _closed = Completer();
+
+  /// True if [close] has been called.
+  bool _closeRequested = false;
+
+  ByteStreamServerChannel(this._input, this._output, this._logger);
+
+  /// Future that will be completed when the input stream is closed.
+  Future get closed {
+    return _closed.future;
+  }
+
+  void close() {
+    if (!_closeRequested) {
+      _closeRequested = true;
+      assert(!_closed.isCompleted);
+      _output.close();
+      _closed.complete();
+    }
+  }
+
+  StreamSubscription<String> listen(
+      void Function(ProtocolMessage message) onMessage,
+      {Function? onError,
+      void Function()? onDone}) {
+    return _input.transform(PacketTransformer()).listen(
+      (String data) => _readMessage(data, onMessage),
+      onError: onError,
+      onDone: () {
+        close();
+        if (onDone != null) {
+          onDone();
+        }
+      },
+    );
+  }
+
+  void sendEvent(Event event) => _sendLsp(event.toJson());
+
+  void sendRequest(Request request) => _sendLsp(request.toJson());
+
+  void sendResponse(Response response) => _sendLsp(response.toJson());
+
+  /// Read a request from the given [data] and use the given function to handle
+  /// the message.
+  void _readMessage(String data, void Function(ProtocolMessage) onMessage) {
+    // Ignore any further requests after the communication channel is closed.
+    if (_closed.isCompleted) {
+      return;
+    }
+    _logger?.call('<== [DAP] $data');
+    try {
+      final Map<String, Object?> json = jsonDecode(data);
+      final type = json['type'] as String;
+      if (type == 'request') {
+        onMessage(Request.fromJson(json));
+      } else if (type == 'event') {
+        onMessage(Event.fromJson(json));
+      } else if (type == 'response') {
+        onMessage(Response.fromJson(json));
+      } else {
+        _sendParseError(data);
+      }
+    } catch (e) {
+      _sendParseError(data);
+    }
+  }
+
+  /// Sends a message prefixed with the required LSP headers.
+  void _sendLsp(Map<String, Object?> json) {
+    // Don't send any further responses after the communication channel is
+    // closed.
+    if (_closeRequested) {
+      return;
+    }
+    final jsonEncodedBody = jsonEncode(json);
+    final utf8EncodedBody = utf8.encode(jsonEncodedBody);
+    final header = 'Content-Length: ${utf8EncodedBody.length}\r\n'
+        'Content-Type: application/vscode-jsonrpc; charset=utf-8\r\n\r\n';
+    final asciiEncodedHeader = ascii.encode(header);
+
+    // Header is always ascii, body is always utf8!
+    _write(asciiEncodedHeader);
+    _write(utf8EncodedBody);
+
+    _logger?.call('==> [DAP] $jsonEncodedBody');
+  }
+
+  void _sendParseError(String data) {
+    // TODO(dantup): Review LSP implementation of this when consolidating classes.
+    throw 'Message does not confirm to DAP spec: $data';
+  }
+
+  /// Send [bytes] to [_output].
+  void _write(List<int> bytes) {
+    runZonedGuarded(
+      () => _output.add(bytes),
+      (e, s) => close(),
+    );
+  }
+}

pkg/dds/lib/src/dap/protocol_stream_transformers.dart

@@ -0,0 +1,134 @@
+// Copyright (c) 2021, the Dart project authors. Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:convert';
+import 'dart:typed_data';
+
+class InvalidEncodingError {
+  final String headers;
+  InvalidEncodingError(this.headers);
+
+  @override
+  String toString() =>
+      'Encoding in supplied headers is not supported.\n\nHeaders:\n$headers';
+}
+
+/// Transforms a stream of LSP/DAP data in the form:
+///
+///     Content-Length: xxx\r\n
+///     Content-Type: application/vscode-jsonrpc; charset=utf-8\r\n
+///     \r\n
+///     { JSON payload }
+///
+/// into just the JSON payload, decoded with the specified encoding. Line endings
+/// for headers must be \r\n on all platforms as defined in the LSP spec.
+class PacketTransformer extends StreamTransformerBase<List<int>, String> {
+  @override
+  Stream<String> bind(Stream<List<int>> stream) {
+    late StreamSubscription<int> input;
+    late StreamController<String> _output;
+    final buffer = <int>[];
+    var isParsingHeaders = true;
+    ProtocolHeaders? headers;
+    _output = StreamController<String>(
+      onListen: () {
+        input = stream.expand((b) => b).listen(
+          (codeUnit) {
+            buffer.add(codeUnit);
+            if (isParsingHeaders && _endsWithCrLfCrLf(buffer)) {
+              headers = _parseHeaders(buffer);
+              buffer.clear();
+              isParsingHeaders = false;
+            } else if (!isParsingHeaders &&
+                buffer.length >= headers!.contentLength) {
+              // UTF-8 is the default - and only supported - encoding for LSP.
+              // The string 'utf8' is valid since it was published in the original spec.
+              // Any other encodings should be rejected with an error.
+              if ([null, 'utf-8', 'utf8']
+                  .contains(headers?.encoding?.toLowerCase())) {
+                _output.add(utf8.decode(buffer));
+              } else {
+                _output.addError(InvalidEncodingError(headers!.rawHeaders));
+              }
+              buffer.clear();
+              isParsingHeaders = true;
+            }
+          },
+          onError: _output.addError,
+          onDone: _output.close,
+        );
+      },
+      onPause: () => input.pause(),
+      onResume: () => input.resume(),
+      onCancel: () => input.cancel(),
+    );
+    return _output.stream;
+  }
+
+  /// Whether [buffer] ends in '\r\n\r\n'.
+  static bool _endsWithCrLfCrLf(List<int> buffer) {
+    var l = buffer.length;
+    return l > 4 &&
+        buffer[l - 1] == 10 &&
+        buffer[l - 2] == 13 &&
+        buffer[l - 3] == 10 &&
+        buffer[l - 4] == 13;
+  }
+
+  static String? _extractEncoding(String header) {
+    final charset = header
+        .split(';')
+        .map((s) => s.trim().toLowerCase())
+        .firstWhere((s) => s.startsWith('charset='), orElse: () => '');
+
+    return charset == '' ? null : charset.split('=')[1];
+  }
+
+  /// Decodes [buffer] into a String and returns the 'Content-Length' header value.
+  static ProtocolHeaders _parseHeaders(List<int> buffer) {
+    // Headers are specified as always ASCII in LSP.
+    final asString = ascii.decode(buffer);
+    final headers = asString.split('\r\n');
+    final lengthHeader =
+        headers.firstWhere((h) => h.startsWith('Content-Length'));
+    final length = lengthHeader.split(':').last.trim();
+    final contentTypeHeader = headers
+        .firstWhere((h) => h.startsWith('Content-Type'), orElse: () => '');
+    final encoding = _extractEncoding(contentTypeHeader);
+    return ProtocolHeaders(asString, int.parse(length), encoding);
+  }
+}
+
+class ProtocolHeaders {
+  final String rawHeaders;
+  final int contentLength;
+  final String? encoding;
+  ProtocolHeaders(this.rawHeaders, this.contentLength, this.encoding);
+}
+
+/// Transforms a stream of [Uint8List]s to [List<int>]s. Used because
+/// [ServerSocket] and [Socket] use [Uint8List] but stdin and stdout use
+/// [List<int>] and the LSP server needs to operate against both.
+class Uint8ListTransformer extends StreamTransformerBase<Uint8List, List<int>> {
+  // TODO(dantup): Is there a built-in (or better) way to do this?
+  @override
+  Stream<List<int>> bind(Stream<Uint8List> stream) {
+    late StreamSubscription<Uint8List> input;
+    late StreamController<List<int>> _output;
+    _output = StreamController<List<int>>(
+      onListen: () {
+        input = stream.listen(
+          (uints) => _output.add(uints),
+          onError: _output.addError,
+          onDone: _output.close,
+        );
+      },
+      onPause: () => input.pause(),
+      onResume: () => input.resume(),
+      onCancel: () => input.cancel(),
+    );
+    return _output.stream;
+  }
+}

pkg/dds/lib/src/dap/server.dart

@@ -0,0 +1,67 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:io';
+
+import 'package:dds/src/dap/adapters/dart.dart';
+import 'package:pedantic/pedantic.dart';
+
+import 'adapters/dart_cli.dart';
+import 'logging.dart';
+import 'protocol_stream.dart';
+import 'protocol_stream_transformers.dart';
+
+/// A DAP server that binds to a port and runs in multi-session mode.
+class DapServer {
+  final ServerSocket _socket;
+  final Logger? _logger;
+  final _channels = <ByteStreamServerChannel>{};
+  final _adapters = <DartDebugAdapter>{};
+
+  DapServer._(this._socket, this._logger) {
+    _socket.listen(_acceptConnection);
+  }
+
+  String get host => _socket.address.host;
+  int get port => _socket.port;
+
+  FutureOr<void> stop() async {
+    _channels.forEach((client) => client.close());
+    await _socket.close();
+  }
+
+  void _acceptConnection(Socket client) {
+    _logger?.call('Accepted connection from ${client.remoteAddress}');
+    client.done.then((_) {
+      _logger?.call('Connection from ${client.remoteAddress} closed');
+    });
+    _createAdapter(client.transform(Uint8ListTransformer()), client, _logger);
+  }
+
+  void _createAdapter(
+      Stream<List<int>> _input, StreamSink<List<int>> _output, Logger? logger) {
+    // TODO(dantup): This is hard-coded to DartCliDebugAdapter but will
+    //   ultimately need to support having a factory passed in to support
+    //   tests and/or being used in flutter_tools.
+    final channel = ByteStreamServerChannel(_input, _output, logger);
+    final adapter = DartCliDebugAdapter(channel);
+    _channels.add(channel);
+    _adapters.add(adapter);
+    unawaited(channel.closed.then((_) {
+      _channels.remove(channel);
+      _adapters.remove(adapter);
+    }));
+  }
+
+  /// Starts a DAP Server listening on [host]:[port].
+  static Future<DapServer> create({
+    String host = 'localhost',
+    int port = 0,
+    Logger? logger,
+  }) async {
+    final _socket = await ServerSocket.bind(host, port);
+    return DapServer._(_socket, logger);
+  }
+}

pkg/dds/test/dap/integration/no_debug_test.dart

@@ -0,0 +1,41 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:test/test.dart';
+
+import 'test_support.dart';
+
+main() {
+  setUpAll(startServerAndClient);
+  tearDownAll(stopServerAndClient);
+
+  group('noDebug', () {
+    test('runs a simple script', () async {
+      final testFile = createTestFile(r'''
+void main(List<String> args) async {
+  print('Hello!');
+  print('World!');
+  print('args: $args');
+}
+    ''');
+
+      final outputEvents = await dapClient.collectOutput(
+        launch: () => dapClient.launch(
+          testFile.path,
+          noDebug: true,
+          args: ['one', 'two'],
+        ),
+      );
+
+      final output = outputEvents.map((e) => e.output).join();
+      expectLines(output, [
+        'Hello!',
+        'World!',
+        'args: [one, two]',
+        '',
+        'Exited.',
+      ]);
+    });
+  });
+}

pkg/dds/test/dap/integration/test_client.dart

@@ -0,0 +1,218 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:io';
+
+import 'package:dds/src/dap/adapters/dart.dart';
+import 'package:dds/src/dap/logging.dart';
+import 'package:dds/src/dap/protocol_generated.dart';
+import 'package:dds/src/dap/protocol_stream.dart';
+import 'package:dds/src/dap/protocol_stream_transformers.dart';
+
+import 'test_server.dart';
+
+/// A helper class to simplify acting as a client for interacting with the
+/// [DapTestServer] in tests.
+class DapTestClient {
+  final Socket _socket;
+  final ByteStreamServerChannel _channel;
+  late final StreamSubscription<String> _subscription;
+  final Logger? _logger;
+  final bool captureVmServiceTraffic;
+  final _requestWarningDuration = const Duration(seconds: 2);
+
+  final Map<int, _OutgoingRequest> _pendingRequests = {};
+  final _eventController = StreamController<Event>.broadcast();
+  int _seq = 1;
+
+  DapTestClient._(
+    this._socket,
+    this._channel,
+    this._logger, {
+    this.captureVmServiceTraffic = false,
+  }) {
+    _subscription = _channel.listen(
+      _handleMessage,
+      onDone: () {
+        if (_pendingRequests.isNotEmpty) {
+          _logger?.call(
+              'Application terminated without a response to ${_pendingRequests.length} requests');
+        }
+        _pendingRequests.forEach((id, request) => request.completer.completeError(
+            'Application terminated without a response to request $id (${request.name})'));
+        _pendingRequests.clear();
+      },
+    );
+  }
+
+  /// Returns a stream of [OutputEventBody] events.
+  Stream<OutputEventBody> get outputEvents => events('output')
+      .map((e) => OutputEventBody.fromJson(e.body as Map<String, Object?>));
+
+  /// Collects all output events until the program terminates.
+  Future<List<OutputEventBody>> collectOutput(
+      {File? file, Future<Response> Function()? launch}) async {
+    final outputEventsFuture = outputEvents.toList();
+
+    // Launch script and wait for termination.
+    await Future.wait([
+      event('terminated'),
+      initialize(),
+      launch?.call() ?? this.launch(file!.path),
+    ], eagerError: true);
+
+    return outputEventsFuture;
+  }
+
+  Future<Response> disconnect() => sendRequest(DisconnectArguments());
+
+  /// Returns a Future that completes with the next [event] event.
+  Future<Event> event(String event) => _logIfSlow(
+      'Event "$event"',
+      _eventController.stream.firstWhere((e) => e.event == event,
+          orElse: () =>
+              throw 'Did not recieve $event event before stream closed'));
+
+  /// Returns a stream for [event] events.
+  Stream<Event> events(String event) {
+    return _eventController.stream.where((e) => e.event == event);
+  }
+
+  /// Send an initialize request to the server.
+  ///
+  /// This occurs before the request to start running/debugging a script and is
+  /// used to exchange capabilities and send breakpoints and other settings.
+  Future<Response> initialize({String exceptionPauseMode = 'None'}) async {
+    final responses = await Future.wait([
+      event('initialized'),
+      sendRequest(InitializeRequestArguments(adapterID: 'test')),
+      // TODO(dantup): Support setting exception pause modes.
+      // sendRequest(
+      //     SetExceptionBreakpointsArguments(filters: [exceptionPauseMode])),
+    ]);
+    await sendRequest(ConfigurationDoneArguments());
+    return responses[1] as Response; // Return the initialize response.
+  }
+
+  /// Send a launchRequest to the server, asking it to start a Dart program.
+  Future<Response> launch(
+    String program, {
+    List<String>? args,
+    String? cwd,
+    bool? noDebug,
+    bool? debugSdkLibraries,
+    bool? evaluateGettersInDebugViews,
+    bool? evaluateToStringInDebugViews,
+  }) {
+    return sendRequest(
+      DartLaunchRequestArguments(
+        noDebug: noDebug,
+        program: program,
+        cwd: cwd,
+        args: args,
+        debugSdkLibraries: debugSdkLibraries,
+        evaluateGettersInDebugViews: evaluateGettersInDebugViews,
+        evaluateToStringInDebugViews: evaluateToStringInDebugViews,
+        // When running out of process, VM Service traffic won't be available
+        // to the client-side logger, so force logging on which sends VM Service
+        // traffic in a custom event.
+        sendLogsToClient: captureVmServiceTraffic,
+      ),
+      // We can't automatically pick the command when using a custom type
+      // (DartLaunchRequestArguments).
+      overrideCommand: 'launch',
+    );
+  }
+
+  /// Sends an arbitrary request to the server.
+  ///
+  /// Returns a Future that completes when the server returns a corresponding
+  /// response.
+  Future<Response> sendRequest(Object? arguments,
+      {bool allowFailure = false, String? overrideCommand}) {
+    final command = overrideCommand ?? commandTypes[arguments.runtimeType]!;
+    final request =
+        Request(seq: _seq++, command: command, arguments: arguments);
+    final completer = Completer<Response>();
+    _pendingRequests[request.seq] =
+        _OutgoingRequest(completer, command, allowFailure);
+    _channel.sendRequest(request);
+    return _logIfSlow('Request "$command"', completer.future);
+  }
+
+  FutureOr<void> stop() async {
+    _channel.close();
+    await _socket.close();
+    await _subscription.cancel();
+  }
+
+  Future<Response> terminate() => sendRequest(TerminateArguments());
+
+  /// Handles an incoming message from the server, completing the relevant request
+  /// of raising the appropriate event.
+  void _handleMessage(message) {
+    if (message is Response) {
+      final pendingRequest = _pendingRequests.remove(message.requestSeq);
+      if (pendingRequest == null) {
+        return;
+      }
+      final completer = pendingRequest.completer;
+      if (message.success || pendingRequest.allowFailure) {
+        completer.complete(message);
+      } else {
+        completer.completeError(message);
+      }
+    } else if (message is Event) {
+      _eventController.add(message);
+
+      // When we see a terminated event, close the event stream so if any
+      // tests are waiting on something that will never come, they fail at
+      // a useful location.
+      if (message.event == 'terminated') {
+        _eventController.close();
+      }
+    }
+  }
+
+  /// Prints a warning if [future] takes longer than [_requestWarningDuration]
+  /// to complete.
+  ///
+  /// Returns [future].
+  Future<T> _logIfSlow<T>(String name, Future<T> future) {
+    var didComplete = false;
+    future.then((_) => didComplete = true);
+    Future.delayed(_requestWarningDuration).then((_) {
+      if (!didComplete) {
+        print(
+            '$name has taken longer than ${_requestWarningDuration.inSeconds}s');
+      }
+    });
+    return future;
+  }
+
+  /// Creates a [DapTestClient] that connects the server listening on
+  /// [host]:[port].
+  static FutureOr<DapTestClient> connect(
+    int port, {
+    String host = 'localhost',
+    bool captureVmServiceTraffic = false,
+    Logger? logger,
+  }) async {
+    final socket = await Socket.connect(host, port);
+    final channel = ByteStreamServerChannel(
+        socket.transform(Uint8ListTransformer()), socket, logger);
+
+    return DapTestClient._(socket, channel, logger,
+        captureVmServiceTraffic: captureVmServiceTraffic);
+  }
+}
+
+class _OutgoingRequest {
+  final Completer<Response> completer;
+  final String name;
+  final bool allowFailure;
+
+  _OutgoingRequest(this.completer, this.name, this.allowFailure);
+}

pkg/dds/test/dap/integration/test_server.dart

@@ -0,0 +1,37 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+
+import 'package:dds/src/dap/server.dart';
+
+abstract class DapTestServer {
+  String get host => _server.host;
+
+  int get port => _server.port;
+  DapServer get _server;
+
+  FutureOr<void> stop();
+}
+
+/// An instance of a DAP server running in-process (to aid debugging).
+///
+/// All communication still goes over the socket to ensure all messages are
+/// serialized and deserialized but it's not quite the same running out of
+/// process.
+class InProcessDapTestServer extends DapTestServer {
+  final DapServer _server;
+
+  InProcessDapTestServer._(this._server);
+
+  @override
+  FutureOr<void> stop() async {
+    await _server.stop();
+  }
+
+  static Future<InProcessDapTestServer> create() async {
+    final DapServer server = await DapServer.create();
+    return InProcessDapTestServer._(server);
+  }
+}

pkg/dds/test/dap/integration/test_support.dart

@@ -0,0 +1,48 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+import 'dart:io';
+
+import 'package:path/path.dart' as path;
+import 'package:test/test.dart';
+
+import 'test_client.dart';
+import 'test_server.dart';
+
+late DapTestClient dapClient;
+late DapTestServer dapServer;
+
+final _testFolders = <Directory>[];
+
+/// Creates a file in a temporary folder to be used as an application for testing.
+File createTestFile(String content) {
+  final testAppDir = Directory.systemTemp.createTempSync('dart-sdk-dap-test');
+  _testFolders.add(testAppDir);
+  final testFile = File(path.join(testAppDir.path, 'test_file.dart'));
+  testFile.writeAsStringSync(content);
+  return testFile;
+}
+
+/// Expects [actual] to equal the lines [expected], ignoring differences in line
+/// endings.
+void expectLines(String actual, List<String> expected) {
+  expect(actual.replaceAll('\r\n', '\n'), equals(expected.join('\n')));
+}
+
+/// Starts a DAP server and a DAP client that connects to it for use in tests.
+FutureOr<void> startServerAndClient() async {
+  // TODO(dantup): An Out-of-process option.
+  dapServer = await InProcessDapTestServer.create();
+  dapClient = await DapTestClient.connect(dapServer.port);
+}
+
+/// Shuts down the DAP server and client created by [startServerAndClient].
+FutureOr<void> stopServerAndClient() async {
+  await dapClient.stop();
+  await dapServer.stop();
+
+  // Clean up any temp folders created during the test run.
+  _testFolders.forEach((dir) => dir.deleteSync(recursive: true));
+}