StreamConsumer has an addStream and a close functions.

tests/lib/async/slow_consumer3_test.dart

Index: tests/lib/async/slow_consumer3_test.dart
diff --git a/tests/lib/async/slow_consumer3_test.dart b/tests/lib/async/slow_consumer3_test.dart
index 6def12fd8b72993f45afec138064ac107afd2dff..8dde39919d18dacc7f38f142704ae91694f24612 100644
--- a/tests/lib/async/slow_consumer3_test.dart
+++ b/tests/lib/async/slow_consumer3_test.dart
@@ -19,10 +19,15 @@ class SlowConsumer extends StreamConsumer {
   final int bufferSize;
   final List bufferedData = [];
   int usedBufferSize = 0;
+  int finalCount;
 
   SlowConsumer(int this.bytesPerSecond, int this.bufferSize);
 
   Future consume(Stream stream) {
+    return addStream(stream).then((_) => close());
+  }
+
+  Future addStream(Stream stream) {
     Completer result = new Completer();
     var subscription;
     subscription = stream.listen(
@@ -44,9 +49,16 @@ class SlowConsumer extends StreamConsumer {
           });
         }
       },
-      onDone: () { result.complete(receivedCount); });
+      onDone: () {
+        finalCount = receivedCount;
+        result.complete(receivedCount);
+      });
     return result.future;
   }
+
+  Future close() {
+    return new Future.immediate(finalCount);
+  }
 }
 
 Stream<List> dataGenerator(int bytesTotal, int chunkSize) {