Do not allow streams to send events during state-change callbacks.

tests/lib/async/slow_consumer_test.dart

Index: tests/lib/async/slow_consumer_test.dart
diff --git a/tests/lib/async/slow_consumer_test.dart b/tests/lib/async/slow_consumer_test.dart
index 4b149244ab4a02fe44ce3172e0d656c188f969c5..f8b6902fc2cb8c9cc77481f60cc0dd0cf48a2d6f 100644
--- a/tests/lib/async/slow_consumer_test.dart
+++ b/tests/lib/async/slow_consumer_test.dart
@@ -20,6 +20,7 @@ class SlowConsumer extends StreamConsumer {
   SlowConsumer(int this.bytesPerSecond);
 
   Future consume(Stream stream) {
+    bool done = false;
     Completer completer = new Completer();
     var subscription;
     subscription = stream.listen(
@@ -29,15 +30,18 @@ class SlowConsumer extends StreamConsumer {
             // Simulated amount of time it takes to handle the data.
             int ms = data.length * 1000 ~/ bytesPerSecond;
             Duration duration = new Duration(milliseconds: ms);
-            subscription.pause();
+            if (!done) subscription.pause();
             return new Future.delayed(duration, () {
-              subscription.resume();
+              if (!done) subscription.resume();
               // Make sure we use data here to keep tracking it.
               return count + data.length;
             });
           });
         },
-      onDone: () { current.then((count) { completer.complete(count); }); });
+      onDone: () {
+        done = true;
+        current.then((count) { completer.complete(count); });
+      });
     return completer.future;
   }
 }
@@ -48,6 +52,7 @@ class DataProvider {
   int sentCount = 0;
   int targetCount;
   StreamController controller;
+  Timer pendingSend;
 
   DataProvider(int this.bytesPerSecond, int this.targetCount, this.chunkSize) {
     controller = new StreamController(onPauseStateChange: onPauseStateChange);
@@ -55,8 +60,11 @@ class DataProvider {
   }
 
   Stream get stream => controller.stream;
-

[Anders Johnsen, 2013/02/20 12:42:02]

Reintroduce this space.

   send() {
+    if (pendingSend != null) {
+      pendingSend.cancel();
+      pendingSend = null;
+    }
     if (controller.isPaused) return;
     if (sentCount == targetCount) {
       controller.close();
@@ -71,7 +79,9 @@ class DataProvider {
     controller.add(new List.fixedLength(listSize));
     int ms = listSize * 1000 ~/ bytesPerSecond;
     Duration duration = new Duration(milliseconds: ms);
-    if (!controller.isPaused) new Timer(duration, send);
+    if (!controller.isPaused) {
+      pendingSend = new Timer(duration, send);
+    }
   }
 
   onPauseStateChange() {