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 3de4c29597c8148429f16abbd40ccc6adb65e37c..dca4452c67773ff8a700285b54aa4628d03c475e 100644
--- a/runtime/bin/socket_patch.dart
+++ b/runtime/bin/socket_patch.dart
@@ -809,6 +809,7 @@ class _SocketStreamConsumer extends StreamConsumer<List<int>> {
     socket._ensureRawSocketSubscription();
     streamCompleter = new Completer<Socket>();
     if (socket._raw != null) {
+      assert(!!!paused);

[Anders Johnsen, 2013/05/22 13:07:08]

!!!?

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

!!!!!
Deliberately made stupid to mark it as debug code :)

       subscription = stream.listen(
           (data) {
             assert(!paused);
@@ -842,12 +843,12 @@ class _SocketStreamConsumer extends StreamConsumer<List<int>> {
       offset += socket._write(buffer, offset, buffer.length - offset);
       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;

[Anders Johnsen, 2013/05/22 13:07:08]

This line (and the comment) can be removed now, due to async future completion.

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

YEY

+          paused = true;

[Anders Johnsen, 2013/05/22 13:07:08]

Can you try without these pause?

[Lasse Reichstein Nielsen, 2013/05/24 06:02:49]

I'll try.

           subscription.pause();
         }
         socket._enableWriteEvent();
@@ -880,6 +881,7 @@ class _SocketStreamConsumer extends StreamConsumer<List<int>> {
     if (subscription == null) return;
     subscription.cancel();
     subscription = null;
+    paused = false;
     socket._disableWriteEvent();
   }
 }