blob: 898d2ee1b04c9a5af3750e1c51b90baf98873959 [file] [log] [blame] [edit]
// Copyright 2014 The Flutter Authors. 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:typed_data';
import 'package:meta/meta.dart';
import 'base/common.dart';
import 'base/io.dart';
import 'base/logger.dart';
import 'base/utils.dart';
import 'convert.dart';
/// A single message passed through the [DaemonConnection].
class DaemonMessage {
DaemonMessage(this.data, [this.binary]);
/// Content of the JSON message in the message.
final Map<String, Object?> data;
/// Stream of the binary content of the message.
///
/// Must be listened to if binary data is present.
final Stream<List<int>>? binary;
}
/// Data of an event passed through the [DaemonConnection].
class DaemonEventData {
DaemonEventData(this.eventName, this.data, [this.binary]);
/// The name of the event.
final String eventName;
/// The data of the event.
final Object? data;
/// Stream of the binary content of the event.
///
/// Must be listened to if binary data is present.
final Stream<List<int>>? binary;
}
const String _binaryLengthKey = '_binaryLength';
enum _InputStreamParseState {
json,
binary,
}
/// Converts a binary stream to a stream of [DaemonMessage].
///
/// The daemon JSON-RPC protocol is defined as follows: every single line of
/// text that starts with `[{` and ends with `}]` will be parsed as a JSON
/// message. The array should contain only one single object which contains the
/// message data.
///
/// If the JSON object contains the key [_binaryLengthKey] with an integer
/// value (will be referred to as N), the following N bytes after the newline
/// character will contain the binary part of the message.
@visibleForTesting
class DaemonInputStreamConverter {
DaemonInputStreamConverter(this.inputStream) {
// Lazily listen to the input stream.
_controller.onListen = () {
final StreamSubscription<List<int>> subscription = inputStream.listen((List<int> chunk) {
_processChunk(chunk);
}, onError: (Object error, StackTrace stackTrace) {
_controller.addError(error, stackTrace);
}, onDone: () {
unawaited(_controller.close());
});
_controller.onCancel = subscription.cancel;
// We should not handle onPause or onResume. When the stream is paused, we
// still need to read from the input stream.
};
}
final Stream<List<int>> inputStream;
final StreamController<DaemonMessage> _controller = StreamController<DaemonMessage>();
Stream<DaemonMessage> get convertedStream => _controller.stream;
// Internal states
/// The current parse state, whether we are expecting JSON or binary data.
_InputStreamParseState state = _InputStreamParseState.json;
/// The binary stream that is being transferred.
late StreamController<List<int>> currentBinaryStream;
/// Remaining length in bytes that have to be sent to the binary stream.
int remainingBinaryLength = 0;
/// Buffer to hold the current line of input data.
final BytesBuilder bytesBuilder = BytesBuilder(copy: false);
// Processes a single chunk received in the input stream.
void _processChunk(List<int> chunk) {
int start = 0;
while (start < chunk.length) {
if (state == _InputStreamParseState.json) {
start += _processChunkInJsonMode(chunk, start);
} else if (state == _InputStreamParseState.binary) {
final int bytesSent = _addBinaryChunk(chunk, start, remainingBinaryLength);
start += bytesSent;
remainingBinaryLength -= bytesSent;
if (remainingBinaryLength <= 0) {
assert(remainingBinaryLength == 0);
unawaited(currentBinaryStream.close());
state = _InputStreamParseState.json;
}
}
}
}
/// Processes a chunk in JSON mode, and returns the number of bytes processed.
int _processChunkInJsonMode(List<int> chunk, int start) {
const int LF = 10; // The '\n' character
// Search for newline character.
final int indexOfNewLine = chunk.indexOf(LF, start);
if (indexOfNewLine < 0) {
bytesBuilder.add(chunk.sublist(start));
return chunk.length - start;
}
bytesBuilder.add(chunk.sublist(start, indexOfNewLine + 1));
// Process chunk here
final Uint8List combinedChunk = bytesBuilder.takeBytes();
String jsonString = utf8.decode(combinedChunk).trim();
if (jsonString.startsWith('[{') && jsonString.endsWith('}]')) {
jsonString = jsonString.substring(1, jsonString.length - 1);
final Map<String, Object?>? value = castStringKeyedMap(json.decode(jsonString));
if (value != null) {
// Check if we need to consume another binary blob.
if (value[_binaryLengthKey] != null) {
remainingBinaryLength = value[_binaryLengthKey]! as int;
currentBinaryStream = StreamController<List<int>>();
state = _InputStreamParseState.binary;
_controller.add(DaemonMessage(value, currentBinaryStream.stream));
} else {
_controller.add(DaemonMessage(value));
}
}
}
return indexOfNewLine + 1 - start;
}
int _addBinaryChunk(List<int> chunk, int start, int maximumSizeToRead) {
if (start == 0 && chunk.length <= remainingBinaryLength) {
currentBinaryStream.add(chunk);
return chunk.length;
} else {
final int chunkRemainingLength = chunk.length - start;
final int sizeToRead = chunkRemainingLength < remainingBinaryLength ? chunkRemainingLength : remainingBinaryLength;
currentBinaryStream.add(chunk.sublist(start, start + sizeToRead));
return sizeToRead;
}
}
}
/// A stream that a [DaemonConnection] uses to communicate with each other.
class DaemonStreams {
DaemonStreams(
Stream<List<int>> rawInputStream,
StreamSink<List<int>> outputSink, {
required Logger logger,
}) :
_outputSink = outputSink,
inputStream = DaemonInputStreamConverter(rawInputStream).convertedStream,
_logger = logger;
/// Creates a [DaemonStreams] that uses stdin and stdout as the underlying streams.
DaemonStreams.fromStdio(Stdio stdio, { required Logger logger })
: this(stdio.stdin, stdio.stdout, logger: logger);
/// Creates a [DaemonStreams] that uses [Socket] as the underlying streams.
DaemonStreams.fromSocket(Socket socket, { required Logger logger })
: this(socket, socket, logger: logger);
/// Connects to a server and creates a [DaemonStreams] from the connection as the underlying streams.
factory DaemonStreams.connect(String host, int port, { required Logger logger }) {
final Future<Socket> socketFuture = Socket.connect(host, port);
final StreamController<List<int>> inputStreamController = StreamController<List<int>>();
final StreamController<List<int>> outputStreamController = StreamController<List<int>>();
socketFuture.then((Socket socket) {
inputStreamController.addStream(socket);
socket.addStream(outputStreamController.stream);
}).onError((Object error, StackTrace stackTrace) {
logger.printError('Socket error: $error');
logger.printTrace('$stackTrace');
// Propagate the error to the streams.
inputStreamController.addError(error, stackTrace);
unawaited(outputStreamController.close());
});
return DaemonStreams(inputStreamController.stream, outputStreamController.sink, logger: logger);
}
final StreamSink<List<int>> _outputSink;
final Logger _logger;
/// Stream that contains input to the [DaemonConnection].
final Stream<DaemonMessage> inputStream;
/// Outputs a message through the connection.
void send(Map<String, Object?> message, [ List<int>? binary ]) {
try {
if (binary != null) {
message[_binaryLengthKey] = binary.length;
}
_outputSink.add(utf8.encode('[${json.encode(message)}]\n'));
if (binary != null) {
_outputSink.add(binary);
}
} on StateError catch (error) {
_logger.printError('Failed to write daemon command response: $error');
// Failed to send, close the connection
_outputSink.close();
} on IOException catch (error) {
_logger.printError('Failed to write daemon command response: $error');
// Failed to send, close the connection
_outputSink.close();
}
}
/// Cleans up any resources used.
Future<void> dispose() async {
unawaited(_outputSink.close());
}
}
/// Connection between a flutter daemon and a client.
class DaemonConnection {
DaemonConnection({
required DaemonStreams daemonStreams,
required Logger logger,
}): _logger = logger,
_daemonStreams = daemonStreams {
_commandSubscription = daemonStreams.inputStream.listen(
_handleMessage,
onError: (Object error, StackTrace stackTrace) {
// We have to listen for on error otherwise the error on the socket
// will end up in the Zone error handler.
// Do nothing here and let the stream close handlers handle shutting
// down the daemon.
}
);
}
final DaemonStreams _daemonStreams;
final Logger _logger;
late final StreamSubscription<DaemonMessage> _commandSubscription;
int _outgoingRequestId = 0;
final Map<String, Completer<Object?>> _outgoingRequestCompleters = <String, Completer<Object?>>{};
final StreamController<DaemonEventData> _events = StreamController<DaemonEventData>.broadcast();
final StreamController<DaemonMessage> _incomingCommands = StreamController<DaemonMessage>();
/// A stream that contains all the incoming requests.
Stream<DaemonMessage> get incomingCommands => _incomingCommands.stream;
/// Listens to the event with the event name [eventToListen].
Stream<DaemonEventData> listenToEvent(String eventToListen) {
return _events.stream
.where((DaemonEventData event) => event.eventName == eventToListen);
}
/// Sends a request to the other end of the connection.
///
/// Returns a [Future] that resolves with the content.
Future<Object?> sendRequest(String method, [Object? params, List<int>? binary]) async {
final String id = '${++_outgoingRequestId}';
final Completer<Object?> completer = Completer<Object?>();
_outgoingRequestCompleters[id] = completer;
final Map<String, Object?> data = <String, Object?>{
'id': id,
'method': method,
if (params != null) 'params': params,
};
_logger.printTrace('-> Sending to daemon, id = $id, method = $method');
_daemonStreams.send(data, binary);
return completer.future;
}
/// Sends a response to the other end of the connection.
void sendResponse(Object id, [Object? result]) {
_daemonStreams.send(<String, Object?>{
'id': id,
if (result != null) 'result': result,
});
}
/// Sends an error response to the other end of the connection.
void sendErrorResponse(Object id, Object? error, StackTrace trace) {
_daemonStreams.send(<String, Object?>{
'id': id,
'error': error,
'trace': '$trace',
});
}
/// Sends an event to the client.
void sendEvent(String name, [ Object? params, List<int>? binary ]) {
_daemonStreams.send(<String, Object?>{
'event': name,
if (params != null) 'params': params,
}, binary);
}
/// Handles the input from the stream.
///
/// There are three kinds of data: Request, Response, Event.
///
/// Request:
/// {"id": <Object>. "method": <String>, "params": <optional, Object?>}
///
/// Response:
/// {"id": <Object>. "result": <optional, Object?>} for a successful response.
/// {"id": <Object>. "error": <Object>, "stackTrace": <String>} for an error response.
///
/// Event:
/// {"event": <String>. "params": <optional, Object?>}
void _handleMessage(DaemonMessage message) {
final Map<String, Object?> data = message.data;
if (data['id'] != null) {
if (data['method'] == null) {
// This is a response to previously sent request.
final String id = data['id']! as String;
if (data['error'] != null) {
// This is an error response.
_logger.printTrace('<- Error response received from daemon, id = $id');
final Object error = data['error']!;
final String stackTrace = data['stackTrace'] as String? ?? '';
_outgoingRequestCompleters.remove(id)?.completeError(error, StackTrace.fromString(stackTrace));
} else {
_logger.printTrace('<- Response received from daemon, id = $id');
final Object? result = data['result'];
_outgoingRequestCompleters.remove(id)?.complete(result);
}
} else {
_incomingCommands.add(message);
}
} else if (data['event'] != null) {
// This is an event
_logger.printTrace('<- Event received: ${data['event']}');
final Object? eventName = data['event'];
if (eventName is String) {
_events.add(DaemonEventData(
eventName,
data['params'],
message.binary,
));
} else {
throwToolExit('event name received is not string!');
}
} else {
_logger.printError('Unknown data received from daemon');
}
}
/// Cleans up any resources used in the connection.
Future<void> dispose() async {
await _commandSubscription.cancel();
await _daemonStreams.dispose();
unawaited(_events.close());
unawaited(_incomingCommands.close());
}
}