StreamConsumer has an addStream and a close functions.

sdk/lib/io/http_impl.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.io;
 
 class _HttpIncoming extends Stream<List<int>> {
   final int _transferLength;
   final Completer _dataCompleter = new Completer();
   Stream<List<int>> _stream;
@@ skipping 389 matching lines @@
     _writeHeaders();
     if (data.length == 0) return;
     _ioSink.writeBytes(data);
   }
 
   Future<T> consume(Stream<List<int>> stream) {
     _writeHeaders();
     return _ioSink.consume(stream);
   }
 
-  Future<T> writeStream(Stream<List<int>> stream) {
+  Future<T> addStream(Stream<List<int>> stream) {
     _writeHeaders();
     return _ioSink.writeStream(stream).then((_) => this);
   }
 
-  void close() {
+  Future<T> writeStream(Stream<List<int>> stream) {
+    return addStream(stream);
+  }
+
+  Future close() {
     // TODO(ajohnsen): Currently, contentLength, chunkedTransferEncoding and
     // persistentConnection is not guaranteed to be in sync.
     if (!_headersWritten && !_ignoreBody && headers.contentLength == -1) {
       // If no body was written, _ignoreBody is false (it's not a HEAD
       // request) and the content-length is unspecified, set contentLength to 0.
       headers.chunkedTransferEncoding = false;
       headers.contentLength = 0;
     }
     _writeHeaders();
-    _ioSink.close();
+    return _ioSink.close();
   }
 
   Future<T> get done {
     _writeHeaders();
     return _ioSink.done;
   }
 
   void _writeHeaders() {
     if (_headersWritten) return;
     _headersWritten = true;
@@ skipping 44 matching lines @@
 
 class _HttpOutboundConsumer implements StreamConsumer {
   Function _consume;
   IOSink _ioSink;
   bool _asGZip;
   _HttpOutboundConsumer(IOSink this._ioSink,
                         Function this._consume,
                         bool this._asGZip);
 
   Future consume(var stream) => _consume(_ioSink, stream, _asGZip);
+
+  Future addStream(var stream) {
+    throw new UnimplementedError("_HttpOutboundConsumer.addStream");
+  }
+
+  Future close() {
+    throw new UnimplementedError("_HttpOutboundConsumer.close");
+  }
 }
 
 
 class _BufferTransformer extends StreamEventTransformer<List<int>, List<int>> {
   const int MIN_CHUNK_SIZE = 4 * 1024;
   const int MAX_BUFFER_SIZE = 16 * 1024;
 
   final _BufferList _buffer = new _BufferList();
 
   void handleData(List<int> data, EventSink<List<int>> sink) {
@@ skipping 387 matching lines @@
     return _consumeCompleter.future;
   }
 
   Future consume(Stream<List<int>> stream) {
     _onStream(stream)
         .then((_) => _consumeCompleter.complete(),
               onError: _consumeCompleter.completeError);
     // Use .then to ensure a Future branch.
     return _consumeCompleter.future.then((_) => this);
   }
+
+  Future addStream(Stream<List<int>> stream) {
+    throw new UnimplementedError("_HttpOutgoing.addStream");
+  }
+
+  Future close() {
+    throw new UnimplementedError("_HttpOutgoing.close");
+  }
 }
 
 
 class _HttpClientConnection {
   final String key;
   final Socket _socket;
   final _HttpParser _httpParser;
   StreamSubscription _subscription;
   final _HttpClient _httpClient;
 
@@ skipping 739 matching lines @@
   void writeAll(Iterable objects, [String separator = ""]) {
     _socket.writeAll(objects, separator);
   }
 
   void writeBytes(List<int> bytes) => _socket.writeBytes(bytes);
 
   Future<Socket> consume(Stream<List<int>> stream) {
     return _socket.consume(stream);
   }
 
+  Future<Socket> addStream(Stream<List<int>> stream) {
+    return _socket.addStream(stream);
+  }
+
   Future<Socket> writeStream(Stream<List<int>> stream) {
     return _socket.writeStream(stream);
   }
 
   void destroy() => _socket.destroy();
 
-  void close() => _socket.close();
+  Future close() => _socket.close();
 
   Future<Socket> get done => _socket.done;
 
   int get port => _socket.port;
 
   String get remoteHost => _socket.remoteHost;
 
   int get remotePort => _socket.remotePort;
 
   bool setOption(SocketOption option, bool enabled) {
@@ skipping 107 matching lines @@
 
 
 class _RedirectInfo implements RedirectInfo {
   const _RedirectInfo(int this.statusCode,
                       String this.method,
                       Uri this.location);
   final int statusCode;
   final String method;
   final Uri location;
 }