StreamConsumer has an addStream and a close functions.

sdk/lib/async/stream.dart

 // Copyright (c) 2013, 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.
 
 part of dart.async;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 219 matching lines @@
    * in order.
    */
   Stream expand(Iterable convert(T value)) {
     return new _ExpandStream<T, dynamic>(this, convert);
   }
 
   /**
    * 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);
   }
 
   /**
    * Chains this stream as the input of the provided [StreamTransformer].
    *
    * Returns the result of [:streamTransformer.bind:] itself.
    */
   Stream transform(StreamTransformer<T, dynamic> streamTransformer) {
     return streamTransformer.bind(this);
@@ skipping 726 matching lines @@
 
 
 /**
  * The target of a [Stream.pipe] call.
  *
  * The [Stream.pipe] call will pass itself to this object, and then return
  * the resulting [Future]. The pipe should complete the future when it's
  * 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);
 }
 
 
 /**
  * The target of a [Stream.transform] call.
  *
  * The [Stream.transform] call will pass itself to this object and then return
  * the resulting stream.
  */
@@ skipping 206 matching lines @@
 
 /* TODO(8997): Implement EventSink instead, */
 class _EventOutputSinkWrapper<T> extends StreamSink<T> {
   _EventOutputSink _sink;
   _EventOutputSinkWrapper(this._sink);
 
   void add(T data) { _sink._sendData(data); }
   void addError(AsyncError error) { _sink._sendError(error); }
   void close() { _sink._sendDone(); }
 }