StreamConsumer has an addStream and a close functions.

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 9bc05bfe0b38984df8485199c2c195bedb7a755a..b8ae7f92b85ad3ae4f1406856aec9fe04d3914c8 100644
--- a/tests/lib/async/slow_consumer_test.dart
+++ b/tests/lib/async/slow_consumer_test.dart
@@ -16,10 +16,15 @@ const int GB = KB * KB * KB;
 class SlowConsumer extends StreamConsumer {
   var current = new Future.immediate(0);
   final int bytesPerSecond;
+  int finalCount;
 
   SlowConsumer(int this.bytesPerSecond);
 
   Future consume(Stream stream) {
+    return addStream(stream).then((_) => close());
+  }
+
+  Future addStream(Stream stream) {
     bool done = false;
     Completer completer = new Completer();
     var subscription;
@@ -40,10 +45,17 @@ class SlowConsumer extends StreamConsumer {
         },
       onDone: () {
         done = true;
-        current.then((count) { completer.complete(count); });
+        current.then((count) {
+          finalCount = count;
+          completer.complete(count);
+        });
       });
     return completer.future;
   }
+
+  Future close() {
+    return new Future.immediate(finalCount);
+  }
 }
 
 class DataProvider {