StreamConsumer has an addStream and a close functions.

pkg/http/test/safe_http_server.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.
 
 library safe_http_server;
 
 import 'dart:async';
 import 'dart:io';
 import 'dart:uri';
 
@@ skipping 13 matching lines @@
   static Future<SafeHttpServer> bind([String host = "127.0.0.1",
       int port = 0, int backlog = 0]) {
     return HttpServer.bind(host, port, backlog)
         .then((server) => new SafeHttpServer(server));
   }
 
   SafeHttpServer(HttpServer server)
       : super(server),
         _inner = server;
 
-  void close() => _inner.close();
+  Future close() => _inner.close();
 
   int get port => _inner.port;
 
   set sessionTimeout(int timeout) {
     _inner.sessionTimeout = timeout;
   }
 
   HttpConnectionsInfo connectionsInfo() => _inner.connectionsInfo();
 
   StreamSubscription<HttpRequest> listen(void onData(HttpRequest value),
@@ skipping 84 matching lines @@
 
   HttpHeaders get headers => _inner.headers;
   List<Cookie> get cookies => _inner.cookies;
   Future<Socket> detachSocket() => _inner.detachSocket();
   HttpConnectionInfo get connectionInfo => _inner.connectionInfo;
   void writeBytes(List<int> data) => _inner.writeBytes(data);
   Future<HttpResponse> consume(Stream<List<int>> stream) =>
     _inner.consume(stream);
   Future<HttpResponse> writeStream(Stream<List<int>> stream) =>
     _inner.writeStream(stream);
-  void close() => _inner.close();
+  Future close() => _inner.close();
   void write(Object obj) => _inner.write(obj);
   void writeAll(Iterable objects, [String separator = ""]) =>
       _inner.writeAll(objects, separator);
   void writeCharCode(int charCode) => _inner.writeCharCode(charCode);
   void writeln([Object obj = ""]) => _inner.writeln(obj);
 }