Make StreamSubscription be the active part of a stream.

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 f233a982789d4ad4d18f2c88aba3147a9118a9ea..d86a17c1d8de72491c92c29d068310286db63a87 100644
--- a/runtime/bin/socket_patch.dart
+++ b/runtime/bin/socket_patch.dart
@@ -837,11 +837,6 @@ class _SocketStreamConsumer extends StreamConsumer<List<int>> {
       if (offset < buffer.length) {
         if (!paused) {
           paused = true;
-          // TODO(ajohnsen): It would be nice to avoid this check.
-          // Some info: socket._write can emit an event, if it fails to write.
-          // If the user closes the socket in that event, stop() will be called
-          // before we get a change to pause.
-          if (subscription == null) return;
           subscription.pause();
         }
         socket._enableWriteEvent();
@@ -874,6 +869,7 @@ class _SocketStreamConsumer extends StreamConsumer<List<int>> {
     if (subscription == null) return;
     subscription.cancel();
     subscription = null;
+    paused = false;
     socket._disableWriteEvent();
   }
 }