StreamConsumer has an addStream and a close functions.

runtime/bin/socket_patch.dart

Index: runtime/bin/socket_patch.dart
diff --git a/runtime/bin/socket_patch.dart b/runtime/bin/socket_patch.dart
index 0e9c25f9d499a5f95cd063c4ef5e39cbaa60103d..cb289e88459a478fc707cc4782c1817c4e2c35df 100644
--- a/runtime/bin/socket_patch.dart
+++ b/runtime/bin/socket_patch.dart
@@ -689,6 +689,34 @@ class _SocketStreamConsumer extends StreamConsumer<List<int>, Socket> {
     return socket._doneFuture;
   }
 
+  Future<Socket> addStream(Stream<List<int>> stream) {
+    Completer completer = new Completer<Socket>();
+    if (socket._raw != null) {
+      subscription = stream.listen(
+          (data) {
+            assert(!paused);
+            assert(buffer == null);
+            buffer = data;
+            offset = 0;
+            write();
+          },
+          onError: (error) {
+            socket._consumerDone(error);
+            completer.completeError(error.error, error.stackTrace);
+          },
+          onDone: () {
+            completer.complete(socket);
+          },
+          unsubscribeOnError: true);
+    }
+    return completer.future;
+  }
+
+  Future<Socket> close() {
+    socket._consumerDone();
+    return completer.future;
+  }
+
   void write() {
     try {
       if (subscription == null) return;