Add Stream.fromIterable

tests/lib/async/slow_consumer3_test.dart

Index: tests/lib/async/slow_consumer3_test.dart
diff --git a/tests/lib/async/slow_consumer2_test.dart b/tests/lib/async/slow_consumer3_test.dart
similarity index 59%
copy from tests/lib/async/slow_consumer2_test.dart
copy to tests/lib/async/slow_consumer3_test.dart
index 8b59aaace0e90a3c2520bbd49071f0ea77bf3879..09e9a0320f7a4b54807e408a34716c3851e6aa80 100644
--- a/tests/lib/async/slow_consumer2_test.dart
+++ b/tests/lib/async/slow_consumer3_test.dart
@@ -4,7 +4,7 @@
 
 // VMOptions=--old_gen_heap_size=32
 
-library slow_consumer2_test;
+library slow_consumer3_test;
 
 import 'dart:async';
 import 'dart:isolate';
@@ -48,52 +48,25 @@ class SlowConsumer extends StreamConsumer {
   }
 }
 
-class DataProvider extends StreamController {
-  final int chunkSize;
-  final int bytesPerSecond;
-  int sentCount = 0;
-  int targetCount;
-
-  DataProvider(int this.bytesPerSecond, int this.targetCount, this.chunkSize) {
-    new Timer(0, (_) => send());
-  }
-
-  send() {
-    if (isPaused) return;
-    if (sentCount == targetCount) {
-      close();
-      return;
-    }
-    int listSize = chunkSize;
-    sentCount += listSize;
-    if (sentCount > targetCount) {
-      listSize -= sentCount - targetCount;
-      sentCount = targetCount;
-    }
-    add(new List.fixedLength(listSize));
-    int ms = listSize * 1000 ~/ bytesPerSecond;
-    if (!isPaused) new Timer(ms, (_) => send());
-  }
-
-  onPauseStateChange() {
-    // We don't care if we just unpaused or paused. In either case we just
-    // call send which will test it for us.
-    send();
-  }
+Stream<List> dataGenerator(int bytesTotal, int chunkSize) {
+  int chunks = bytesTotal ~/ chunkSize;
+  return new Stream.fromIterable(new Iterable.generate(chunks, (_) {
+    // This assumes one byte per entry. In practice it will be more.
+    return new List<int>.fixedLength(chunkSize);
+  }));
 }
 
 main() {
   var port = new ReceivePort();
-  // The data provider can deliver 800MB/s of data. It sends 100MB of data to
-  // the slower consumer who can only read 200MB/s. The data is sent in 1MB
-  // chunks. The consumer has a buffer of 5MB. That is, it can accept a few
-  // packages without pausing its input.
+  // The data provider can deliver 800MBs of data as fast as it is
+  // requested. The data is sent in 1MB chunks. The consumer has a buffer of
+  // 5MB. That is, it can accept a few packages without pausing its input.
   //
   // This test is limited to 32MB of heap-space (see VMOptions on top of the
   // file). If the consumer doesn't pause the data-provider it will run out of
   // heap-space.
 
-  new DataProvider(800 * MB, 100 * MB, 1 * MB)
+  dataGenerator(100 * MB, 1 * MB)
     .pipe(new SlowConsumer(200 * MB, 5 * MB))
     .then((count) {
       port.close();