StreamConsumer has an addStream and a close functions.

tests/lib/async/slow_consumer2_test.dart

 // Copyright (c) 2012, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 // VMOptions=--old_gen_heap_size=64
 
 library slow_consumer2_test;
 
 import 'dart:async';
 import 'dart:isolate';
 
 const int KB = 1024;
 const int MB = KB * KB;
 const int GB = KB * KB * KB;
 
 class SlowConsumer extends StreamConsumer {
   int receivedCount = 0;
   final int bytesPerSecond;
   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(
       (List<int> data) {
         receivedCount += data.length;
         usedBufferSize += data.length;
         bufferedData.add(data);
         int currentBufferedDataLength = bufferedData.length;
         if (usedBufferSize > bufferSize) {
           subscription.pause();
           usedBufferSize = 0;
           int ms = data.length * 1000 ~/ bytesPerSecond;
           Duration duration = new Duration(milliseconds: ms);
           new Timer(duration, () {
             for (int i = 0; i < currentBufferedDataLength; i++) {
               bufferedData[i] = null;
             }
             subscription.resume();
           });
         }
       },
-      onDone: () { result.complete(receivedCount); });
+      onDone: () {
+        finalCount = receivedCount;
+        result.complete(receivedCount);
+      });
     return result.future;
   }
+
+  Future close() {
+    return new Future.immediate(finalCount);
+  }
 }
 
 class DataProvider {
   final int chunkSize;
   final int bytesPerSecond;
   int sentCount = 0;
   int targetCount;
   StreamController controller;
 
   DataProvider(int this.bytesPerSecond, int this.targetCount, this.chunkSize) {
@@ skipping 39 matching lines @@
   // 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).stream
     .pipe(new SlowConsumer(200 * MB, 5 * MB))
     .then((count) {
       port.close();
       Expect.equals(100 * MB, count);
     });
 }