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>
This commit is contained in:
Janice Collins
committed by
commit-bot@chromium.org
commit-bot@chromium.org
parent
3f36b1fed1
commit
40b4389891
@@ -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);
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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 = '']) {}
|
||||
}
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user