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

Fixes the test brekage and reverts commit caed03df.

packages/flutter_tools/lib/src/commands/attach.dart

@@ -18,6 +18,7 @@ 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';
@@ -235,8 +236,10 @@ known, it can be explicitly provided to attach via the command-line, e.g.
 
     final Daemon daemon = boolArg('machine')
       ? Daemon(
-          stdinCommandStream,
-          stdoutCommandResponse,
+          DaemonConnection(
+            daemonStreams: StdioDaemonStreams(globals.stdio),
+            logger: globals.logger,
+          ),
           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

@@ -14,6 +14,7 @@ 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;
@@ -556,8 +557,10 @@ class RunCommand extends RunCommandBase {
         throwToolExit('"--machine" does not support "-d all".');
       }
       final Daemon daemon = Daemon(
-        stdinCommandStream,
-        stdoutCommandResponse,
+        DaemonConnection(
+          daemonStreams: StdioDaemonStreams(globals.stdio),
+          logger: globals.logger,
+        ),
         notifyingLogger: (globals.logger is NotifyingLogger)
           ? globals.logger as NotifyingLogger
           : NotifyingLogger(verbose: globals.logger.isVerbose, parent: globals.logger),

packages/flutter_tools/lib/src/daemon.dart

+// 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

+// 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;
+  }
+}