// Copyright (c) 2018, 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:analysis_server_client/listener/server_listener.dart'; import 'package:analysis_server_client/protocol.dart'; import 'package:path/path.dart'; /// Type of callbacks used to process notifications. typedef void NotificationProcessor(Notification notification); /// Instances of the class [Server] manage a server process, /// and facilitate communication to and from the server. /// /// Clients may not extend, implement or mix-in this class. class Server { /// If not `null`, [_listener] will be sent information /// about interactions with the server. ServerListener _listener; /// Server process object, or `null` if server hasn't been started yet /// or if the server has already been stopped. Process _process; /// Commands that have been sent to the server but not yet acknowledged, /// and the [Completer] objects which should be completed /// when acknowledgement is received. final _pendingCommands = >>{}; /// Number which should be used to compute the 'id' /// to send in the next command sent to the server. int _nextId = 0; /// The stderr subscription or `null` if either /// [listenToOutput] has not been called or [stop] has been called. StreamSubscription _stderrSubscription; /// The stdout subscription or `null` if either /// [listenToOutput] has not been called or [stop] has been called. StreamSubscription _stdoutSubscription; Server({ServerListener listener, Process process}) : this._listener = listener, this._process = process; /// Force kill the server. Returns exit code future. Future kill({String reason = 'none'}) { _listener?.killingServerProcess(reason); final process = _process; _process = null; process.kill(); return process.exitCode; } /// Start listening to output from the server, /// and deliver notifications to [notificationProcessor]. void listenToOutput({NotificationProcessor notificationProcessor}) { _stdoutSubscription = _process.stdout .transform(utf8.decoder) .transform(new LineSplitter()) .listen((String line) { String trimmedLine = line.trim(); // Guard against lines like: // {"event":"server.connected","params":{...}}Observatory listening on ... const observatoryMessage = 'Observatory listening on '; if (trimmedLine.contains(observatoryMessage)) { trimmedLine = trimmedLine .substring(0, trimmedLine.indexOf(observatoryMessage)) .trim(); } if (trimmedLine.isEmpty) { return; } _listener?.messageReceived(trimmedLine); Map message; try { message = json.decoder.convert(trimmedLine); } catch (exception) { _listener?.badMessage(trimmedLine, exception); return; } final id = message[Response.ID]; if (id != null) { // Handle response final completer = _pendingCommands.remove(id); if (completer == null) { _listener?.unexpectedResponse(message, id); } if (message.containsKey(Response.ERROR)) { completer.completeError(new RequestError.fromJson( new ResponseDecoder(null), '.error', message[Response.ERROR])); } else { completer.complete(message[Response.RESULT]); } } else { // Handle notification final String event = message[Notification.EVENT]; if (event != null) { if (notificationProcessor != null) { notificationProcessor( new Notification(event, message[Notification.PARAMS])); } } else { _listener?.unexpectedMessage(message); } } }); _stderrSubscription = _process.stderr .transform(utf8.decoder) .transform(new LineSplitter()) .listen((String line) { String trimmedLine = line.trim(); _listener?.errorMessage(trimmedLine); }); } /// 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> send( String method, Map params) { String id = '${_nextId++}'; Map command = { Request.ID: id, Request.METHOD: method }; if (params != null) { command[Request.PARAMS] = params; } final completer = new Completer>(); _pendingCommands[id] = completer; String line = json.encode(command); _listener?.requestSent(line); _process.stdin.add(utf8.encoder.convert("$line\n")); return completer.future; } /// Start the server. /// /// If [profileServer] is `true`, the server will be started /// with "--observe" and "--pause-isolates-on-exit", allowing the observatory /// to be used. /// /// If [serverPath] is specified, then that analysis server will be launched, /// otherwise the analysis server snapshot in the SDK will be launched. Future start({ String clientId, String clientVersion, int diagnosticPort, String instrumentationLogFile, bool profileServer: false, String sdkPath, String serverPath, int servicesPort, bool suppressAnalytics: true, bool useAnalysisHighlight2: false, }) async { if (_process != null) { throw new Exception('Process already started'); } String dartBinary = Platform.executable; // The integration tests run 3x faster when run from snapshots // (you need to run test.py with --use-sdk). if (serverPath == null) { // Look for snapshots/analysis_server.dart.snapshot. serverPath = normalize(join(dirname(Platform.resolvedExecutable), 'snapshots', 'analysis_server.dart.snapshot')); if (!FileSystemEntity.isFileSync(serverPath)) { // Look for dart-sdk/bin/snapshots/analysis_server.dart.snapshot. serverPath = normalize(join(dirname(Platform.resolvedExecutable), 'dart-sdk', 'bin', 'snapshots', 'analysis_server.dart.snapshot')); } } List arguments = []; // // Add VM arguments. // if (profileServer) { if (servicesPort == null) { arguments.add('--observe'); } else { arguments.add('--observe=$servicesPort'); } arguments.add('--pause-isolates-on-exit'); } else if (servicesPort != null) { arguments.add('--enable-vm-service=$servicesPort'); } if (Platform.packageConfig != null) { arguments.add('--packages=${Platform.packageConfig}'); } // // Add the server executable. // arguments.add(serverPath); // // Add server arguments. // // TODO(danrubel): Consider moving all cmdline argument consts // out of analysis_server and into analysis_server_client if (clientId != null) { arguments.add('--client-id'); arguments.add(clientId); } if (clientVersion != null) { arguments.add('--client-version'); arguments.add(clientVersion); } if (suppressAnalytics) { arguments.add('--suppress-analytics'); } if (diagnosticPort != null) { arguments.add('--port'); arguments.add(diagnosticPort.toString()); } if (instrumentationLogFile != null) { arguments.add('--instrumentation-log-file=$instrumentationLogFile'); } if (sdkPath != null) { arguments.add('--sdk=$sdkPath'); } if (useAnalysisHighlight2) { arguments.add('--useAnalysisHighlight2'); } _listener?.startingServer(dartBinary, arguments); _process = await Process.start(dartBinary, arguments); _process.exitCode.then((int code) { if (code != 0 && _process != null) { // Report an error if server abruptly terminated _listener?.unexpectedStop(code); } }); } /// Attempt to gracefully shutdown the server. /// If that fails, then kill the process. Future stop({Duration timeLimit}) async { timeLimit ??= const Duration(seconds: 5); if (_process == null) { // Process already exited return -1; } final future = send(SERVER_REQUEST_SHUTDOWN, null); final process = _process; _process = null; await future // fall through to wait for exit .timeout(timeLimit, onTimeout: () { return null; }).whenComplete(() async { await _stderrSubscription?.cancel(); _stderrSubscription = null; await _stdoutSubscription?.cancel(); _stdoutSubscription = null; }); return process.exitCode.timeout( timeLimit, onTimeout: () { _listener?.killingServerProcess('server failed to exit'); process.kill(); return process.exitCode; }, ); } }