StreamConsumer has an addStream and a close functions.

sdk/lib/async/stream.dart

Index: sdk/lib/async/stream.dart
diff --git a/sdk/lib/async/stream.dart b/sdk/lib/async/stream.dart
index 38ace6dab4daa9a182908c72b7d38e446710cefc..e7a5ab685f075cdab3a2db36bd9dd23962dac3fb 100644
--- a/sdk/lib/async/stream.dart
+++ b/sdk/lib/async/stream.dart
@@ -237,6 +237,8 @@ abstract class Stream<T> {
    * Binds this stream as the input of the provided [StreamConsumer].
    */
   Future pipe(StreamConsumer<T, dynamic> streamConsumer) {
+    // TODO(floitsch): switch to:
+    // streamConsumer.addStream(this).then((_) => streamConsumer.close());
     return streamConsumer.consume(this);
   }
 
@@ -983,6 +985,16 @@ class EventSinkView<T> extends StreamSink<T> {
  * done.
  */
 abstract class StreamConsumer<S, T> {
+  // TODO(floitsch): generic types.
+  // Currently not possible to add generic types, since they clash with other
+  // types that have already been used.
+  Future addStream(Stream<S> stream);
+  Future close();
+
+
+  /**
+   * Consume is deprecated. Use [addStream] followed by [close] instead.
+   */
   Future<T> consume(Stream<S> stream);
 }