Add Stream.drain() method.

tests/lib/async/stream_controller_async_test.dart

 // Copyright (c) 2011, 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.
 
 // Test the basic StreamController and StreamController.singleSubscription.
 library stream_controller_async_test;
 
 import "package:expect/expect.dart";
 import 'dart:async';
 import 'dart:isolate';
@@ skipping 277 matching lines @@
     f.then(expectAsync1((v) { Expect.equals(13, v);}));
     sentEvents.replay(c);
   });
 
   test("elementAt 2", () {
     StreamController c = new StreamController();
     Future f = c.stream.elementAt(20);
     f.catchError(expectAsync1((error) { Expect.isTrue(error is StateError); }));
     sentEvents.replay(c);
   });
+
+  test("drain", () {
+    StreamController c = new StreamController();
+    Future f = c.stream.drain();
+    f.then(expectAsync1((v) { Expect.equals(null, v);}));
+    sentEvents.replay(c);
+  });
+
+  test("drain error", () {
+    StreamController c = new StreamController();
+    Future f = c.stream.drain();
+    f.catchError(expectAsync1((error) { Expect.equals("error", error); }));
+    Events errorEvents = new Events()..error("error")..error("error2")..close();
+    errorEvents.replay(c);
+  });
+
 }
 
 testPause() {
   test("pause event-unpause", () {
     StreamController c = new StreamController();
     Events actualEvents = new Events.capture(c.stream);
     Events expectedEvents = new Events();
     expectedEvents.add(42);
     c.add(42);
     Expect.listEquals(expectedEvents.events, actualEvents.events);
@@ skipping 144 matching lines @@
   testStream("map", (s, act) => s.map(act));
   testStream("expand", (s, act) => s.expand(act));
   testStream("where", (s, act) => s.where(act));
   testStreamError("handleError", (s, act) => s.handleError(act));
   testStreamError("handleTest", (s, act) => s.handleError((v) {}, test: act));
   testFuture("forEach", (s, act) => s.forEach(act));
   testFuture("every", (s, act) => s.every(act));
   testFuture("any", (s, act) => s.any(act));
   testFuture("reduce", (s, act) => s.reduce((a,b) => act(b)));
   testFuture("fold", (s, act) => s.fold(0, (a,b) => act(b)));
+  testFuture("drain", (s, act) => s.drain().then(act));
 }
 
 main() {
   testController();
   testSingleController();
   testExtraMethods();
   testPause();
   testRethrow();
 }