Close the IntegrationTestTestDevice stream when the VM service shuts down (#90022)

packages/flutter_tools/lib/src/test/integration_test_device.dart

@@ -97,7 +97,14 @@ class IntegrationTestTestDevice implements TestDevice {
       );
     });
 
-    unawaited(remoteMessages.pipe(controller.local.sink));
+    remoteMessages.listen(
+      (String s) => controller.local.sink.add(s),
+      onError: (Object error, StackTrace stack) => controller.local.sink.addError(error, stack),
+    );
+    unawaited(vmService.service.onDone.whenComplete(
+      () => controller.local.sink.close(),
+    ));
+
     return controller.foreign;
   }
 

packages/flutter_tools/test/general.shard/integration_test_device_test.dart

@@ -11,6 +11,7 @@ import 'package:flutter_tools/src/device.dart';
 import 'package:flutter_tools/src/test/integration_test_device.dart';
 import 'package:flutter_tools/src/test/test_device.dart';
 import 'package:flutter_tools/src/vmservice.dart';
+import 'package:stream_channel/stream_channel.dart';
 import 'package:vm_service/vm_service.dart' as vm_service;
 
 import '../src/context.dart';
@@ -211,4 +212,21 @@ void main() {
       Device device,
     }) async => fakeVmServiceHost.vmService,
   });
+
+  testUsingContext('Can handle closing of the VM service', () async {
+    final StreamChannel<String> channel = await testDevice.start('entrypointPath');
+    await fakeVmServiceHost.vmService.dispose();
+    expect(await channel.stream.isEmpty, true);
+  }, overrides: <Type, Generator>{
+    VMServiceConnector: () => (Uri httpUri, {
+      ReloadSources reloadSources,
+      Restart restart,
+      CompileExpression compileExpression,
+      GetSkSLMethod getSkSLMethod,
+      PrintStructuredErrorLogMethod printStructuredErrorLogMethod,
+      io.CompressionOptions compression,
+      Device device,
+      Logger logger,
+    }) async => fakeVmServiceHost.vmService,
+  });
 }