[flutter_tool] Migrate DAP off ProcessUtils.writelnToStdinUnsafe (#171081)

Instead, queue up each write to ensure we don't try to write to the
stream concurrently which leads to "Bad state: StreamSink is bound to a
stream"

Fixes https://github.com/Dart-Code/Dart-Code/issues/5554

## Pre-launch Checklist

- [x] I read the [Contributor Guide] and followed the process outlined
there for submitting PRs.
- [x] I read the [Tree Hygiene] wiki page, which explains my
responsibilities.
- [x] I read and followed the [Flutter Style Guide], including [Features
we expect every widget to implement].
- [x] I signed the [CLA].
- [x] I listed at least one issue that this PR fixes in the description
above.
- [x] I updated/added relevant documentation (doc comments with `///`).
- [x] I added new tests to check the change I am making, or this PR is
[test-exempt].
- [x] I followed the [breaking change policy] and added [Data Driven
Fixes] where supported.
- [ ] All existing and new tests are passing.

If you need help, consider asking for advice on the #hackers-new channel
on [Discord].

<!-- Links -->
[Contributor Guide]:
https://github.com/flutter/flutter/blob/main/docs/contributing/Tree-hygiene.md#overview
[Tree Hygiene]:
https://github.com/flutter/flutter/blob/main/docs/contributing/Tree-hygiene.md
[test-exempt]:
https://github.com/flutter/flutter/blob/main/docs/contributing/Tree-hygiene.md#tests
[Flutter Style Guide]:
https://github.com/flutter/flutter/blob/main/docs/contributing/Style-guide-for-Flutter-repo.md
[Features we expect every widget to implement]:
https://github.com/flutter/flutter/blob/main/docs/contributing/Style-guide-for-Flutter-repo.md#features-we-expect-every-widget-to-implement
[CLA]: https://cla.developers.google.com/
[flutter/tests]: https://github.com/flutter/tests
[breaking change policy]:
https://github.com/flutter/flutter/blob/main/docs/contributing/Tree-hygiene.md#handling-breaking-changes
[Discord]:
https://github.com/flutter/flutter/blob/main/docs/contributing/Chat.md
[Data Driven Fixes]:
https://github.com/flutter/flutter/blob/main/docs/contributing/Data-driven-Fixes.md

packages/flutter_tools/lib/src/debug_adapters/flutter_adapter.dart

@@ -323,6 +323,18 @@ class FlutterDebugAdapter extends FlutterBaseDebugAdapter with VmServiceInfoFile
     return completer.future;
   }
 
+  /// A future that completes when the last-queued write to the Flutter process
+  /// completes and is flushed. This prevents multiple attempts to write to the
+  /// processes stdin stream that can cause exceptions.
+  ///
+  /// See:
+  ///   - https://github.com/Dart-Code/Dart-Code/issues/5554
+  ///   - https://github.com/flutter/flutter/issues/137184
+  ///
+  /// [sendFlutterMessage] will replace this value each time it writes a
+  /// message.
+  Future<void> _currentFlutterProcessStdinWrite = Future<void>.value();
+
   /// Sends a message to the Flutter run daemon.
   ///
   /// Throws `DebugAdapterException` if a Flutter process is not yet running.
@@ -336,7 +348,21 @@ class FlutterDebugAdapter extends FlutterBaseDebugAdapter with VmServiceInfoFile
     // Flutter requests are always wrapped in brackets as an array.
     final String payload = '[$messageString]\n';
     _logTraffic('==> [Flutter] $payload');
-    await ProcessUtils.writelnToStdinUnsafe(stdin: process.stdin, line: payload);
+
+    _currentFlutterProcessStdinWrite = _currentFlutterProcessStdinWrite.then((_) {
+      return ProcessUtils.writelnToStdinGuarded(
+        stdin: process.stdin,
+        line: payload,
+        onError: (Object e, _) {
+          // Ignore failures to write to the stream, it means the process has
+          // terminated and will be handled by the exit handler.
+          logger?.call(
+            'Error writing to "flutter run" stdin. '
+            'It is likely the process has terminated: $e',
+          );
+        },
+      );
+    });
   }
 
   /// Called by [terminateRequest] to request that we gracefully shut down the app being run (or in the case of an attach, disconnect).

packages/flutter_tools/test/integration.shard/debug_adapter/flutter_adapter_test.dart

@@ -379,6 +379,44 @@ The relevant error-causing widget was:
       await dap.client.terminate();
     });
 
+    /// Sending many hot reload requests at once previously could result in
+    /// failures because they tried to write to the internal `flutter run`
+    /// process concurrently (via ) which caused "Bad state: StreamSink is bound
+    /// to a stream".
+    ///
+    /// See:
+    /// - https://github.com/Dart-Code/Dart-Code/issues/5554
+    /// - https://github.com/flutter/flutter/issues/137184
+    testWithoutContext('can handle many hot reload requests concurrently', () async {
+      final BasicProject project = BasicProject();
+      await project.setUpIn(tempDir);
+
+      // Launch the app and wait for it to print "topLevelFunction".
+      await Future.wait(<Future<void>>[
+        dap.client.stdoutOutput.firstWhere(
+          (String output) => output.startsWith('topLevelFunction'),
+        ),
+        dap.client.start(
+          launch:
+              () => dap.client.launch(
+                cwd: project.dir.path,
+                noDebug: true,
+                toolArgs: <String>['-d', 'flutter-tester'],
+              ),
+        ),
+      ], eagerError: true);
+
+      try {
+        await Future.wait(Iterable<Future<void>>.generate(50, (_) => dap.client.hotReload()));
+      } on Response catch (e) {
+        // If the request throws, extract the error message for better failure
+        // message.
+        fail(e.message ?? 'hotReload request failed with unknown error');
+      }
+
+      await dap.client.terminate();
+    });
+
     testWithoutContext('sends progress notifications during hot reload', () async {
       final BasicProject project = BasicProject();
       await project.setUpIn(tempDir);