From 40b43898910bd0feef477103aa5f997635aaddbf Mon Sep 17 00:00:00 2001
From: Janice Collins <jcollins@google.com>
Date: Tue, 7 Apr 2020 20:02:16 +0000
Subject: [PATCH] Add AnalysisServer client implementation for isolates.

Change-Id: I1b4a9394a996cdc9907273689d2c9dc0c900de55
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/142556
Reviewed-by: Paul Berry <paulberry@google.com>
Commit-Queue: Janice Collins <jcollins@google.com>
---
 pkg/nnbd_migration/lib/isolate_server.dart    | 164 +++++++++++
 pkg/nnbd_migration/pubspec.yaml               |   1 +
 .../test/isolate_server_test.dart             | 264 ++++++++++++++++++
 pkg/nnbd_migration/test/test_all.dart         |   2 +
 4 files changed, 431 insertions(+)
 create mode 100644 pkg/nnbd_migration/lib/isolate_server.dart
 create mode 100644 pkg/nnbd_migration/test/isolate_server_test.dart

diff --git a/pkg/nnbd_migration/lib/isolate_server.dart b/pkg/nnbd_migration/lib/isolate_server.dart
new file mode 100644
index 00000000000..c4ecfa21875
--- /dev/null
+++ b/pkg/nnbd_migration/lib/isolate_server.dart
@@ -0,0 +1,164 @@
+// Copyright (c) 2020, 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.
+
+/// This library implements [ServerBase], except unlike analysis_server_client's
+/// `Server`, [Server] runs the analysis server in an isolate.
+
+import 'dart:async';
+import 'dart:convert';
+import 'dart:isolate';
+
+import 'package:analysis_server/starter.dart';
+import 'package:analysis_server_client/listener/server_listener.dart';
+import 'package:analysis_server_client/src/server_base.dart';
+import 'package:analysis_server_client/protocol.dart';
+import 'package:stream_channel/isolate_channel.dart';
+
+/// Wrap server arguments and communication port into a single parameter
+/// for [Isolate.spawn].
+class _IsolateParameters {
+  final List<String> arguments;
+  final SendPort sendPort;
+
+  _IsolateParameters(this.arguments, this.sendPort);
+}
+
+/// Manage an analysis_server launched in an isolate.
+class Server extends ServerBase {
+  /// Server isolate object, or `null` if server hasn't been started yet
+  /// or if the server has already been stopped.
+  Isolate _isolate;
+
+  /// The [ReceivePort] data subscription via an [IsolateChannel], or `null`
+  /// if either [listenToOutput] has not been called or [stop] has been called.
+  StreamSubscription<String> _receiveSubscription;
+
+  /// Construct a Server.
+  ///
+  /// [isolate] and [isolateChannel] are testing-only parameters, allowing
+  /// you to bypass start().
+  Server(
+      {ServerListener listener,
+      Isolate isolate,
+      IsolateChannel isolateChannel,
+      bool stdioPassthrough = false})
+      : _isolate = isolate,
+        _isolateChannel = isolateChannel,
+        super(listener: listener, stdioPassthrough: stdioPassthrough);
+
+  /// Completes when the [_isolate] has exited.
+  Completer isolateExited = Completer();
+
+  /// The [IsolateChannel] by which this class communicates with the [_isolate].
+  IsolateChannel _isolateChannel;
+
+  /// Force kill the server.  The returned future completes when the isolate
+  /// is dead.
+  Future<void> kill({String reason = 'none'}) {
+    listener?.killingServerProcess(reason);
+    final isolate = _isolate;
+    final isolateExitedOriginal = isolateExited;
+    _isolate = null;
+    isolateExited = null;
+    isolate.kill(priority: Isolate.immediate);
+    return isolateExitedOriginal.future;
+  }
+
+  /// Start listening to output from the server,
+  /// and deliver notifications to [notificationProcessor].
+  void listenToOutput({NotificationProcessor notificationProcessor}) {
+    _receiveSubscription = _isolateChannel.stream
+        .transform(utf8.decoder)
+        .transform(LineSplitter())
+        .listen((line) => outputProcessor(line, notificationProcessor));
+  }
+
+  /// Send a command to the server. An 'id' will be automatically assigned.
+  /// The returned [Future] will be completed when the server acknowledges
+  /// the command with a response.
+  /// If the server acknowledges the command with a normal (non-error) response,
+  /// the future will be completed with the 'result' field from the response.
+  /// If the server acknowledges the command with an error response,
+  /// the future will be completed with an error.
+  Future<Map<String, dynamic>> send(
+          String method, Map<String, dynamic> params) =>
+      sendCommandWith(method, params, _isolateChannel.sink.add);
+
+  /// Start the server in a new [Isolate].
+  Future start({
+    String clientId,
+    String clientVersion,
+    int diagnosticPort,
+    String instrumentationLogFile,
+    String sdkPath,
+    bool suppressAnalytics = true,
+    bool useAnalysisHighlight2 = false,
+  }) async {
+    if (_isolate != null) {
+      throw Exception('Isolate already started');
+    }
+
+    // Even though this is an isolate, most of the analysis server code
+    // can't tell the difference.  So we construct "command line" arguments
+    // just the same as analysis_server_client.
+    List<String> arguments = [];
+    arguments.addAll(getServerArguments(
+        clientId: clientId,
+        clientVersion: clientVersion,
+        suppressAnalytics: suppressAnalytics,
+        diagnosticPort: diagnosticPort,
+        instrumentationLogFile: instrumentationLogFile,
+        sdkPath: sdkPath,
+        useAnalysisHighlight2: useAnalysisHighlight2));
+
+    listener?.startingServer('((isolate))', arguments);
+    ReceivePort receivePort = ReceivePort();
+    ReceivePort onExitReceivePort = ReceivePort();
+    ReceivePort onErrorReceivePort = ReceivePort();
+    onExitReceivePort.listen((_) {
+      isolateExited.complete(0);
+    });
+    onErrorReceivePort.listen((_) {
+      listener?.unexpectedStop(null);
+    });
+    _isolateChannel = IsolateChannel<List<int>>.connectReceive(receivePort);
+    _isolate = await Isolate.spawn(
+        _runIsolate, _IsolateParameters(arguments, receivePort.sendPort),
+        onExit: onExitReceivePort.sendPort);
+  }
+
+  /// This is the function passed to [Isolate.spawn] to actually begin
+  /// the server.
+  static void _runIsolate(_IsolateParameters parameters) {
+    ServerStarter starter = ServerStarter();
+    // TODO(jcollins-g): consider a refactor that does not require passing
+    // text arguments to start the server.
+    starter.start(parameters.arguments, parameters.sendPort);
+  }
+
+  /// Attempt to gracefully shutdown the server.
+  /// If that fails, then kill the isolate.
+  Future<void> stop({Duration timeLimit}) async {
+    timeLimit ??= const Duration(seconds: 5);
+    if (_isolate == null) {
+      // isolate already exited
+      return;
+    }
+    final future = send(SERVER_REQUEST_SHUTDOWN, null);
+    final isolate = _isolate;
+    _isolate = null;
+    await future
+        // fall through to wait for exit
+        .timeout(timeLimit, onTimeout: () {
+      return null;
+    }).whenComplete(() async {
+      await _receiveSubscription?.cancel();
+      _receiveSubscription = null;
+    });
+    return isolateExited.future.timeout(timeLimit, onTimeout: () {
+      listener?.killingServerProcess('server failed to exit');
+      isolate.kill(priority: Isolate.immediate);
+    });
+  }
+}
diff --git a/pkg/nnbd_migration/pubspec.yaml b/pkg/nnbd_migration/pubspec.yaml
index 2d65cbcab31..55968c0c14d 100644
--- a/pkg/nnbd_migration/pubspec.yaml
+++ b/pkg/nnbd_migration/pubspec.yaml
@@ -7,6 +7,7 @@ dependencies:
   analyzer: ^0.37.0
   analyzer_plugin: ^0.2.2
   path: ^1.6.2
+  stream_channel: any
   yaml: any
 dev_dependencies:
   args: ^1.5.2
diff --git a/pkg/nnbd_migration/test/isolate_server_test.dart b/pkg/nnbd_migration/test/isolate_server_test.dart
new file mode 100644
index 00000000000..c352204610e
--- /dev/null
+++ b/pkg/nnbd_migration/test/isolate_server_test.dart
@@ -0,0 +1,264 @@
+// Copyright (c) 2017, 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 'dart:isolate';
+
+import 'package:analysis_server_client/protocol.dart';
+import 'package:async/src/stream_sink_transformer.dart';
+import 'package:nnbd_migration/isolate_server.dart';
+import 'package:stream_channel/isolate_channel.dart';
+import 'package:stream_channel/src/stream_channel_transformer.dart';
+import 'package:stream_channel/stream_channel.dart';
+import 'package:test/test.dart';
+
+void main() {
+  FakeIsolate isolate;
+  FakeIsolateChannel isolateChannel;
+  Server server;
+
+  setUp(() async {
+    isolate = FakeIsolate();
+    isolateChannel = FakeIsolateChannel();
+    server = Server(isolate: isolate, isolateChannel: isolateChannel);
+  });
+
+  group('listenToOutput', () {
+    test('good', () async {
+      isolateChannel.stream = _goodMessage();
+
+      final future = server.send('blahMethod', null);
+      server.listenToOutput();
+
+      final response = await future;
+      expect(response['foo'], 'bar');
+    });
+
+    test('error', () async {
+      isolateChannel.stream = _badMessage();
+
+      final future = server.send('blahMethod', null);
+      future.catchError((e) {
+        expect(e, const TypeMatcher<RequestError>());
+        final error = e as RequestError;
+        expect(error.code, RequestErrorCode.UNKNOWN_REQUEST);
+        expect(error.message, 'something went wrong');
+        expect(error.stackTrace, 'some long stack trace');
+      });
+      server.listenToOutput();
+    });
+
+    test('event', () async {
+      isolateChannel.stream = _eventMessage();
+
+      final completer = Completer();
+      void eventHandler(Notification notification) {
+        expect(notification.event, 'fooEvent');
+        expect(notification.params.length, 2);
+        expect(notification.params['foo'] as String, 'bar');
+        expect(notification.params['baz'] as String, 'bang');
+        completer.complete();
+      }
+
+      server.send('blahMethod', null);
+      server.listenToOutput(notificationProcessor: eventHandler);
+      await completer.future;
+    });
+  });
+
+  group('stop', () {
+    test('ok', () async {
+      final fakeOut = StreamController<List<int>>();
+      isolateChannel.stream = fakeOut.stream;
+      // ignore: unawaited_futures
+      isolateChannel.fakeIn.controller.stream.first.then((_) {
+        var encoded = json.encode({'id': '0'});
+        fakeOut.add(utf8.encoder.convert('$encoded\n'));
+      });
+      server.isolateExited.complete();
+      server.listenToOutput();
+      await server.stop(timeLimit: const Duration(milliseconds: 1));
+      expect(isolate.killed, isFalse);
+    });
+    test('stopped', () async {
+      final fakeOut = StreamController<List<int>>();
+      isolateChannel.stream = fakeOut.stream;
+
+      server.isolateExited.complete();
+      server.listenToOutput();
+      await server.stop(timeLimit: const Duration(milliseconds: 1));
+      expect(isolate.killed, isFalse);
+    });
+    test('kill', () async {
+      final fakeOut = StreamController<List<int>>();
+      isolateChannel.stream = fakeOut.stream;
+
+      server.listenToOutput();
+      await server.stop(timeLimit: const Duration(milliseconds: 10));
+      expect(isolate.killed, isTrue);
+    });
+  });
+}
+
+final _badErrorMessage = {
+  'code': 'UNKNOWN_REQUEST',
+  'message': 'something went wrong',
+  'stackTrace': 'some long stack trace'
+};
+
+Stream<List<int>> _badMessage() async* {
+  yield utf8.encoder.convert('Observatory listening on foo bar\n');
+  final sampleJson = {
+    'id': '0',
+    'error': _badErrorMessage,
+  };
+  yield utf8.encoder.convert(json.encode(sampleJson));
+}
+
+Stream<List<int>> _eventMessage() async* {
+  yield utf8.encoder.convert('Observatory listening on foo bar\n');
+  final sampleJson = {
+    'event': 'fooEvent',
+    'params': {'foo': 'bar', 'baz': 'bang'}
+  };
+  yield utf8.encoder.convert(json.encode(sampleJson));
+}
+
+Stream<List<int>> _goodMessage() async* {
+  yield utf8.encoder.convert('Observatory listening on foo bar\n');
+  final sampleJson = {
+    'id': '0',
+    'result': {'foo': 'bar'}
+  };
+  yield utf8.encoder.convert(json.encode(sampleJson));
+}
+
+class FakeIsolate implements Isolate {
+  bool killed = false;
+
+  @override
+  void addErrorListener(SendPort port) => throw UnimplementedError();
+
+  @override
+  void addOnExitListener(SendPort port, {Object response}) =>
+      throw UnimplementedError();
+
+  @override
+  SendPort get controlPort => throw UnimplementedError();
+
+  @override
+  String get debugName => throw UnimplementedError();
+
+  @override
+  Stream get errors => throw UnimplementedError();
+
+  @override
+  Capability get pauseCapability => throw UnimplementedError();
+
+  @override
+  void ping(SendPort port,
+          {Object response, int priority = Isolate.immediate}) =>
+      throw UnimplementedError();
+
+  @override
+  void removeErrorListener(SendPort port) => throw UnimplementedError();
+
+  @override
+  void removeOnExitListener(SendPort port) => throw UnimplementedError();
+
+  @override
+  void resume(Capability capability) => throw UnimplementedError();
+
+  @override
+  Capability get terminateCapability => throw UnimplementedError();
+
+  @override
+  void kill({int priority = Isolate.beforeNextEvent}) {
+    killed = true;
+  }
+
+  @override
+  Capability pause([Capability resumeCapability]) => throw UnimplementedError();
+
+  @override
+  void setErrorsFatal(bool errorsAreFatal) => throw UnimplementedError();
+}
+
+class FakeIsolateChannel<T> implements IsolateChannel<T> {
+  FakeIsolateInput fakeIn = FakeIsolateInput();
+
+  @override
+  StreamChannel<S> cast<S>() => throw UnimplementedError();
+
+  @override
+  StreamChannel<T> changeSink(
+          StreamSink<T> Function(StreamSink<T> sink) change) =>
+      throw UnimplementedError();
+
+  @override
+  StreamChannel<T> changeStream(Stream<T> Function(Stream<T> stream) change) =>
+      throw UnimplementedError();
+
+  @override
+  void pipe(StreamChannel<T> other) => throw UnimplementedError();
+
+  @override
+  StreamSink<T> get sink => fakeIn as StreamSink<T>;
+
+  @override
+  Stream<T> stream;
+
+  @override
+  StreamChannel<S> transform<S>(StreamChannelTransformer<S, T> transformer) =>
+      throw UnimplementedError();
+
+  @override
+  StreamChannel<T> transformSink(StreamSinkTransformer<T, T> transformer) =>
+      throw UnimplementedError();
+
+  @override
+  StreamChannel<T> transformStream(StreamTransformer<T, T> transformer) =>
+      throw UnimplementedError();
+}
+
+class FakeIsolateInput implements IOSink {
+  final controller = StreamController<String>();
+
+  @override
+  Encoding encoding;
+
+  @override
+  Future get done => null;
+
+  @override
+  void add(List<int> data) {
+    controller.add(utf8.decode(data));
+  }
+
+  @override
+  void addError(Object error, [StackTrace stackTrace]) {}
+
+  @override
+  Future addStream(Stream<List<int>> stream) => null;
+
+  @override
+  Future close() => null;
+
+  @override
+  Future flush() => null;
+
+  @override
+  void write(Object obj) {}
+
+  @override
+  void writeAll(Iterable objects, [String separator = '']) {}
+
+  @override
+  void writeCharCode(int charCode) {}
+
+  @override
+  void writeln([Object obj = '']) {}
+}
diff --git a/pkg/nnbd_migration/test/test_all.dart b/pkg/nnbd_migration/test/test_all.dart
index 76a98449770..28221f2945e 100644
--- a/pkg/nnbd_migration/test/test_all.dart
+++ b/pkg/nnbd_migration/test/test_all.dart
@@ -19,6 +19,7 @@ import 'fantasyland/test_all.dart' as fantasyland;
 import 'fix_aggregator_test.dart' as fix_aggregator_test;
 import 'fix_builder_test.dart' as fix_builder_test;
 import 'instrumentation_test.dart' as instrumentation_test;
+import 'isolate_server_test.dart' as isolate_server_test;
 import 'node_builder_test.dart' as node_builder_test;
 import 'nullability_node_test.dart' as nullability_node_test;
 import 'utilities/test_all.dart' as utilities;
@@ -38,6 +39,7 @@ main() {
     fix_aggregator_test.main();
     fix_builder_test.main();
     instrumentation_test.main();
+    isolate_server_test.main();
     node_builder_test.main();
     nullability_node_test.main();
     utilities.main();