[io] Fix a bug where Socket.addStream never completes after the socket is destroyed.

Bug:#54735
Change-Id: I8895ff6271eb1019ddad13c171f36cbec5fd84cd
TEST=tests/standalone/io/client_socket_destroy_and_add_stream_test.dart
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/348764
Commit-Queue: Brian Quinlan <[email protected]>
Reviewed-by: Alexander Aprelev <[email protected]>

Author: brianquinlan

sdk/lib/_internal/vm/bin/socket_patch.dart

@@ -2184,6 +2184,8 @@ class _SocketStreamConsumer implements StreamConsumer<List<int>> {
         assert(buffer == null);
         done();
       }, cancelOnError: true);
+    } else {
+      done();
     }
     return completer.future;
   }

tests/standalone/io/client_socket_destroy_and_add_stream_test.dart

@@ -0,0 +1,35 @@
+// Copyright (c) 2024, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+// Tests that adding a stream to a `Socket` that has been `destroy`ed returns
+// a `Future` that completes.
+
+import "dart:async";
+import "dart:io";
+
+import "package:async_helper/async_helper.dart";
+import "package:expect/expect.dart";
+
+void main() async {
+  asyncStart();
+
+  final server = await ServerSocket.bind("127.0.0.1", 0);
+  late final Socket connectedSocket;
+  server.listen((socket) {
+    // Note: must keep socket alive for the duration of the test.
+    // Otherwise GC might collect it and and shutdown this side of socket
+    // which would cause writing to abort.
+    connectedSocket = socket;
+    // Passive block data by not subscribing to socket.
+  });
+
+  final client = await Socket.connect("127.0.0.1", server.port);
+  client.listen((data) {}, onDone: server.close);
+  client.add(new List.filled(1024 * 1024, 0));
+  client.destroy();
+  await client.addStream(Stream.fromIterable([
+    [1, 2, 3, 4]
+  ]));
+  asyncEnd();
+}