close sinks cleanup (#5838)

part of https://github.com/flutter/flutter/issues/5789

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

@@ -636,7 +636,11 @@ class AnalysisServer {
     _errorsController.add(new FileAnalysisErrors(file, errors));
   }
 
-  Future<bool> dispose() async => _process?.kill();
+  Future<bool> dispose() async {
+    await _analyzingController.close();
+    await _errorsController.close();
+    return _process?.kill();
+  }
 }
 
 class FileAnalysisErrors {

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

@@ -100,7 +100,10 @@ class Daemon {
     // Start listening.
     commandStream.listen(
       (Map<String, dynamic> request) => _handleRequest(request),
-      onDone: () => _onExitCompleter.complete(0)
+      onDone: () {
+        if (!_onExitCompleter.isCompleted)
+            _onExitCompleter.complete(0);
+      }
     );
   }
 
@@ -579,6 +582,10 @@ class NotifyingLogger extends Logger {
     printStatus(message);
     return new Status();
   }
+
+  void dispose() {
+    _messageController.close();
+  }
 }
 
 /// A running application, started by this daemon.

packages/flutter_tools/test/daemon_test.dart

@@ -42,6 +42,7 @@ void main() {
     tearDown(() {
       if (daemon != null)
         return daemon.shutdown();
+      notifyingLogger.dispose();
     });
 
     _testUsingContext('daemon.version', () async {
@@ -57,6 +58,8 @@ void main() {
       expect(response['id'], 0);
       expect(response['result'], isNotEmpty);
       expect(response['result'] is String, true);
+      responses.close();
+      commands.close();
     });
 
     _testUsingContext('daemon.logMessage', () {
@@ -77,6 +80,8 @@ void main() {
         Map<String, String> logMessage = response['params'];
         expect(logMessage['level'], 'error');
         expect(logMessage['message'], 'daemon.logMessage test');
+        responses.close();
+        commands.close();
       });
     });
 
@@ -90,6 +95,8 @@ void main() {
       );
       commands.add(<String, dynamic>{'id': 0, 'method': 'daemon.shutdown'});
       return daemon.onExit.then((int code) {
+        responses.close();
+        commands.close();
         expect(code, 0);
       });
     });
@@ -111,6 +118,8 @@ void main() {
       Map<String, dynamic> response = await responses.stream.where(_notEvent).first;
       expect(response['id'], 0);
       expect(response['error'], contains('deviceId is required'));
+      responses.close();
+      commands.close();
     });
 
     _testUsingContext('daemon.restart', () async {
@@ -130,6 +139,8 @@ void main() {
       Map<String, dynamic> response = await responses.stream.where(_notEvent).first;
       expect(response['id'], 0);
       expect(response['error'], contains('appId is required'));
+      responses.close();
+      commands.close();
     });
 
     _testUsingContext('daemon.stop', () async {
@@ -149,6 +160,8 @@ void main() {
       Map<String, dynamic> response = await responses.stream.where(_notEvent).first;
       expect(response['id'], 0);
       expect(response['error'], contains('appId is required'));
+      responses.close();
+      commands.close();
     });
 
     _testUsingContext('device.getDevices', () async {
@@ -163,6 +176,8 @@ void main() {
       Map<String, dynamic> response = await responses.stream.where(_notEvent).first;
       expect(response['id'], 0);
       expect(response['result'], isList);
+      responses.close();
+      commands.close();
     });
   });
 }

packages/flutter_tools/test/protocol_discovery_test.dart

@@ -48,6 +48,7 @@ void main() {
       expect(await nextPort, 52584);
 
       discoverer.cancel();
+      logReader.dispose();
     });
   });
 }

packages/flutter_tools/test/src/mocks.dart

@@ -62,6 +62,10 @@ class MockDeviceLogReader extends DeviceLogReader {
   Stream<String> get logLines => _linesController.stream;
 
   void addLine(String line) => _linesController.add(line);
+
+  void dispose() {
+    _linesController.close();
+  }
 }
 
 void applyMocksToCommand(FlutterCommand command) {