Skip to content

F
Front-End
Unverified Commit caed03df authored by Zachary Anderson's avatar Zachary Anderson Committed by GitHub
Browse files
Options

Revert "Add an option for flutter daemon to listen on a TCP port (#95418)" (#95686) 

This reverts commit 2b46ea44.
parent 2b46ea44
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 261 additions and 671 deletions
packages/flutter_tools/lib/src/commands/attach.dart
View file @ caed03df
......@@ -18,7 +18,6 @@ import '../base/io.dart';
import '../build_info.dart';
import '../commands/daemon.dart';
import '../compile.dart';
import '../daemon.dart';
import '../device.dart';
import '../device_port_forwarder.dart';
import '../fuchsia/fuchsia_device.dart';
......@@ -236,10 +235,8 @@ known, it can be explicitly provided to attach via the command-line, e.g.
final Daemon daemon = boolArg('machine')
? Daemon(
DaemonConnection(
daemonStreams: StdioDaemonStreams(globals.stdio),
logger: globals.logger,
),
stdinCommandStream,
stdoutCommandResponse,
notifyingLogger: (globals.logger is NotifyingLogger)
? globals.logger as NotifyingLogger
: NotifyingLogger(verbose: globals.logger.isVerbose, parent: globals.logger),
......
packages/flutter_tools/lib/src/commands/run.dart
View file @ caed03df
......@@ -14,7 +14,6 @@ import '../base/common.dart';
import '../base/file_system.dart';
import '../base/utils.dart';
import '../build_info.dart';
import '../daemon.dart';
import '../device.dart';
import '../features.dart';
import '../globals.dart' as globals;
......@@ -557,10 +556,8 @@ class RunCommand extends RunCommandBase {
throwToolExit('"--machine" does not support "-d all".');
}
final Daemon daemon = Daemon(
DaemonConnection(
daemonStreams: StdioDaemonStreams(globals.stdio),
logger: globals.logger,
),
stdinCommandStream,
stdoutCommandResponse,
notifyingLogger: (globals.logger is NotifyingLogger)
? globals.logger as NotifyingLogger
: NotifyingLogger(verbose: globals.logger.isVerbose, parent: globals.logger),
......
packages/flutter_tools/lib/src/daemon.dart deleted 100644 → 0
View file @ 2b46ea44
// 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 'base/common.dart';
import 'base/io.dart';
import 'base/logger.dart';
import 'base/utils.dart';
import 'convert.dart';
/// Parse binary streams in the JSON RPC format understood by the daemon, and
/// convert it into a stream of JSON RPC messages.
Stream<Map<String, Object?>> _convertInputStream(Stream<List<int>> inputStream) {
return utf8.decoder.bind(inputStream)
.transform<String>(const LineSplitter())
.where((String line) => line.startsWith('[{') && line.endsWith('}]'))
.map<Map<String, Object?>?>((String line) {
line = line.substring(1, line.length - 1);
return castStringKeyedMap(json.decode(line));
})
.where((Map<String, Object?>? entry) => entry != null)
.cast<Map<String, Object?>>();
}
/// A stream that a [DaemonConnection] uses to communicate with each other.
abstract class DaemonStreams {
/// Stream that contains input to the [DaemonConnection].
Stream<Map<String, Object?>> get inputStream;
/// Outputs a message through the connection.
void send(Map<String, Object?> message);
/// Cleans up any resources used.
Future<void> dispose() async { }
}
/// A [DaemonStream] that uses stdin and stdout as the underlying streams.
class StdioDaemonStreams extends DaemonStreams {
StdioDaemonStreams(Stdio stdio) :
_stdio = stdio,
inputStream = _convertInputStream(stdio.stdin);
final Stdio _stdio;
@override
final Stream<Map<String, Object?>> inputStream;
@override
void send(Map<String, Object?> message) {
_stdio.stdoutWrite(
'[${json.encode(message)}]\n',
fallback: (String message, Object? error, StackTrace stack) {
throwToolExit('Failed to write daemon command response to stdout: $error');
},
);
}
}
/// A [DaemonStream] that uses [Socket] as the underlying stream.
class TcpDaemonStreams extends DaemonStreams {
/// Creates a [DaemonStreams] with an existing [Socket].
TcpDaemonStreams(
Socket socket, {
required Logger logger,
}): _logger = logger {
_socket = Future<Socket>.value(_initializeSocket(socket));
}
/// Connects to a remote host and creates a [DaemonStreams] from the socket.
TcpDaemonStreams.connect(
String host,
int port, {
required Logger logger,
}) : _logger = logger {
_socket = Socket.connect(host, port).then(_initializeSocket);
}
late final Future<Socket> _socket;
final StreamController<Map<String, Object?>> _commands = StreamController<Map<String, Object?>>();
final Logger _logger;
@override
Stream<Map<String, Object?>> get inputStream => _commands.stream;
@override
void send(Map<String, Object?> message) {
_socket.then((Socket socket) {
try {
socket.write('[${json.encode(message)}]\n');
} on SocketException catch (error) {
_logger.printError('Failed to write daemon command response to socket: $error');
// Failed to send, close the connection
socket.close();
}
});
}
Socket _initializeSocket(Socket socket) {
_commands.addStream(_convertInputStream(socket));
return socket;
}
@override
Future<void> dispose() async {
await (await _socket).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(
_handleData,
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<Map<String, Object?>> _commandSubscription;
int _outgoingRequestId = 0;
final Map<String, Completer<Object?>> _outgoingRequestCompleters = <String, Completer<Object?>>{};
final StreamController<Map<String, Object?>> _events = StreamController<Map<String, Object?>>.broadcast();
final StreamController<Map<String, Object?>> _incomingCommands = StreamController<Map<String, Object?>>();
/// A stream that contains all the incoming requests.
Stream<Map<String, Object?>> get incomingCommands => _incomingCommands.stream;
/// Listens to the event with the event name [eventToListen].
Stream<Object?> listenToEvent(String eventToListen) {
return _events.stream
.where((Map<String, Object?> event) => event['event'] == eventToListen)
.map<Object?>((Map<String, Object?> event) => event['params']);
}
/// 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]) 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);
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 ]) {
_daemonStreams.send(<String, Object?>{
'event': name,
if (params != null) 'params': params,
});
}
/// 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 _handleData(Map<String, Object?> 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(data);
}
} else if (data['event'] != null) {
// This is an event
_logger.printTrace('<- Event received: ${data['event']}');
_events.add(data);
} 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());
}
}
packages/flutter_tools/test/general.shard/daemon_test.dart deleted 100644 → 0
View file @ 2b46ea44
// 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:flutter_tools/src/base/common.dart';
import 'package:flutter_tools/src/base/io.dart';
import 'package:flutter_tools/src/base/logger.dart';
import 'package:flutter_tools/src/convert.dart';
import 'package:flutter_tools/src/daemon.dart';
import 'package:test/fake.dart';
import '../src/common.dart';
class FakeDaemonStreams extends DaemonStreams {
final StreamController<Map<String, dynamic>> inputs = StreamController<Map<String, dynamic>>();
final StreamController<Map<String, dynamic>> outputs = StreamController<Map<String, dynamic>>();
@override
Stream<Map<String, dynamic>> get inputStream {
return inputs.stream;
}
@override
void send(Map<String, dynamic> message) {
outputs.add(message);
}
@override
Future<void> dispose() async {
await inputs.close();
// In some tests, outputs have no listeners. We don't wait for outputs to close.
unawaited(outputs.close());
}
}
void main() {
late BufferLogger bufferLogger;
late FakeDaemonStreams daemonStreams;
late DaemonConnection daemonConnection;
setUp(() {
bufferLogger = BufferLogger.test();
daemonStreams = FakeDaemonStreams();
daemonConnection = DaemonConnection(
daemonStreams: daemonStreams,
logger: bufferLogger,
);
});
tearDown(() async {
await daemonConnection.dispose();
});
group('DaemonConnection receiving end', () {
testWithoutContext('redirects input to incoming commands', () async {
final Map<String, dynamic> commandToSend = <String, dynamic>{'id': 0, 'method': 'some_method'};
daemonStreams.inputs.add(commandToSend);
final Map<String, dynamic> commandReceived = await daemonConnection.incomingCommands.first;
await daemonStreams.dispose();
expect(commandReceived, commandToSend);
});
testWithoutContext('listenToEvent can receive the right events', () async {
final Future<List<dynamic>> events = daemonConnection.listenToEvent('event1').toList();
daemonStreams.inputs.add(<String, dynamic>{'event': 'event1', 'params': '1'});
daemonStreams.inputs.add(<String, dynamic>{'event': 'event2', 'params': '2'});
daemonStreams.inputs.add(<String, dynamic>{'event': 'event1', 'params': null});
daemonStreams.inputs.add(<String, dynamic>{'event': 'event1', 'params': 3});
await pumpEventQueue();
await daemonConnection.dispose();
expect(await events, <dynamic>['1', null, 3]);
});
});
group('DaemonConnection sending end', () {
testWithoutContext('sending requests', () async {
unawaited(daemonConnection.sendRequest('some_method', 'param'));
final Map<String, dynamic> data = await daemonStreams.outputs.stream.first;
expect(data['id'], isNotNull);
expect(data['method'], 'some_method');
expect(data['params'], 'param');
});
testWithoutContext('sending requests without param', () async {
unawaited(daemonConnection.sendRequest('some_method'));
final Map<String, dynamic> data = await daemonStreams.outputs.stream.first;
expect(data['id'], isNotNull);
expect(data['method'], 'some_method');
expect(data['params'], isNull);
});
testWithoutContext('sending response', () async {
daemonConnection.sendResponse('1', 'some_data');
final Map<String, dynamic> data = await daemonStreams.outputs.stream.first;
expect(data['id'], '1');
expect(data['method'], isNull);
expect(data['error'], isNull);
expect(data['result'], 'some_data');
});
testWithoutContext('sending response without data', () async {
daemonConnection.sendResponse('1');
final Map<String, dynamic> data = await daemonStreams.outputs.stream.first;
expect(data['id'], '1');
expect(data['method'], isNull);
expect(data['error'], isNull);
expect(data['result'], isNull);
});
testWithoutContext('sending error response', () async {
daemonConnection.sendErrorResponse('1', 'error', StackTrace.fromString('stack trace'));
final Map<String, dynamic> data = await daemonStreams.outputs.stream.first;
expect(data['id'], '1');
expect(data['method'], isNull);
expect(data['error'], 'error');
expect(data['trace'], 'stack trace');
});
testWithoutContext('sending events', () async {
daemonConnection.sendEvent('some_event', '123');
final Map<String, dynamic> data = await daemonStreams.outputs.stream.first;
expect(data['id'], isNull);
expect(data['event'], 'some_event');
expect(data['params'], '123');
});
testWithoutContext('sending events without params', () async {
daemonConnection.sendEvent('some_event');
final Map<String, dynamic> data = await daemonStreams.outputs.stream.first;
expect(data['id'], isNull);
expect(data['event'], 'some_event');
expect(data['params'], isNull);
});
});
group('DaemonConnection request and response', () {
testWithoutContext('receiving response from requests', () async {
final Future<dynamic> requestFuture = daemonConnection.sendRequest('some_method', 'param');
final Map<String, dynamic> data = await daemonStreams.outputs.stream.first;
expect(data['id'], isNotNull);
expect(data['method'], 'some_method');
expect(data['params'], 'param');
final String id = data['id'] as String;
daemonStreams.inputs.add(<String, dynamic>{'id': id, 'result': '123'});
expect(await requestFuture, '123');
});
testWithoutContext('receiving response from requests without result', () async {
final Future<dynamic> requestFuture = daemonConnection.sendRequest('some_method', 'param');
final Map<String, dynamic> data = await daemonStreams.outputs.stream.first;
expect(data['id'], isNotNull);
expect(data['method'], 'some_method');
expect(data['params'], 'param');
final String id = data['id'] as String;
daemonStreams.inputs.add(<String, dynamic>{'id': id});
expect(await requestFuture, null);
});
testWithoutContext('receiving error response from requests without result', () async {
final Future<dynamic> requestFuture = daemonConnection.sendRequest('some_method', 'param');
final Map<String, dynamic> data = await daemonStreams.outputs.stream.first;
expect(data['id'], isNotNull);
expect(data['method'], 'some_method');
expect(data['params'], 'param');
final String id = data['id'] as String;
daemonStreams.inputs.add(<String, dynamic>{'id': id, 'error': 'some_error', 'trace': 'stack trace'});
expect(requestFuture, throwsA('some_error'));
});
});
group('TcpDaemonStreams', () {
final Map<String, Object?> testCommand = <String, Object?>{
'id': 100,
'method': 'test',
};
late FakeSocket socket;
late TcpDaemonStreams daemonStreams;
setUp(() {
socket = FakeSocket();
daemonStreams = TcpDaemonStreams(socket, logger: bufferLogger);
});
test('parses the message received on the socket', () async {
socket.controller.add(Uint8List.fromList(utf8.encode('[${jsonEncode(testCommand)}]\n')));
final Map<String, Object?> command = await daemonStreams.inputStream.first;
expect(command, testCommand);
});
test('sends the encoded message through the socket', () async {
daemonStreams.send(testCommand);
await pumpEventQueue();
expect(socket.writtenObjects.length, 1);
expect(socket.writtenObjects[0].toString(), '[${jsonEncode(testCommand)}]\n');
});
test('dispose calls socket.close', () async {
await daemonStreams.dispose();
expect(socket.closeCalled, isTrue);
});
});
}
class FakeSocket extends Fake implements Socket {
bool closeCalled = false;
final StreamController<Uint8List> controller = StreamController<Uint8List>();
final List<Object?> writtenObjects = <Object?>[];
@override
StreamSubscription<Uint8List> listen(
void Function(Uint8List event)? onData, {
Function? onError,
void Function()? onDone,
bool? cancelOnError,
}) {
return controller.stream.listen(onData, onError: onError, onDone: onDone, cancelOnError: cancelOnError);
}
@override
void write(Object? object) {
writtenObjects.add(object);
}
@override
Future<void> close() async {
closeCalled = true;
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

HIAST - GitLab