blob: 360897b3d536a37bf64a8144ca0ec50f437f0f6e [file] [log] [blame]
// 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 '../async_message_grouper.dart';
import '../worker_protocol.pb.dart';
import '../constants.dart';
import '../utils.dart';
/// A connection from a `BazelWorkerDriver` to a worker.
/// Unlike `WorkerConnection` there is no synchronous version of this class.
/// This is because drivers talk to multiple workers, so they should never block
/// when waiting for the response of any individual worker.
abstract class DriverConnection {
Future<WorkResponse> readResponse();
void writeRequest(WorkRequest request);
Future cancel();
/// Default implementation of [DriverConnection] that works with [Stdin]
/// and [Stdout].
class StdDriverConnection implements DriverConnection {
final AsyncMessageGrouper _messageGrouper;
final StreamSink<List<int>> _outputStream;
Future<void> get done => _messageGrouper.done;
{Stream<List<int>> inputStream, StreamSink<List<int>> outputStream})
: _messageGrouper = new AsyncMessageGrouper(inputStream ?? stdin),
_outputStream = outputStream ?? stdout;
factory StdDriverConnection.forWorker(Process worker) =>
new StdDriverConnection(
inputStream: worker.stdout, outputStream: worker.stdin);
/// Note: This will attempts to recover from invalid proto messages by parsing
/// them as strings. This is a common error case for workers (they print a
/// message to stdout on accident). This isn't perfect however as it only
/// happens if the parsing throws, you can still hang indefinitely if the
/// [MessageGrouper] doesn't find what it thinks is the end of a proto
/// message.
Future<WorkResponse> readResponse() async {
var buffer = await;
if (buffer == null) return null;
WorkResponse response;
try {
response = new WorkResponse.fromBuffer(buffer);
} catch (_) {
try {
// Try parsing the message as a string and set that as the output.
var output = utf8.decode(buffer);
var response = new WorkResponse()
..exitCode = EXIT_CODE_ERROR
..output = 'Worker sent an invalid response:\n$output';
return response;
} catch (_) {
// Fall back to original exception and rethrow if we fail to parse as
// a string.
return response;
void writeRequest(WorkRequest request) {
Future cancel() async {
await _outputStream.close();
await _messageGrouper.cancel();
/// [DriverConnection] that works with an isolate via a [SendPort].
class IsolateDriverConnection implements DriverConnection {
final StreamIterator _receivePortIterator;
final SendPort _sendPort;
IsolateDriverConnection._(this._receivePortIterator, this._sendPort);
/// Creates a driver connection for a worker in an isolate. Provide the
/// [receivePort] attached to the [SendPort] that the isolate was created
/// with.
static Future<IsolateDriverConnection> create(ReceivePort receivePort) async {
var receivePortIterator = new StreamIterator(receivePort);
await receivePortIterator.moveNext();
var sendPort = receivePortIterator.current as SendPort;
return new IsolateDriverConnection._(receivePortIterator, sendPort);
Future<WorkResponse> readResponse() async {
await _receivePortIterator.moveNext();
return WorkResponse.fromBuffer(_receivePortIterator.current as List<int>);
void writeRequest(WorkRequest request) {
Future cancel() async {}