Add ClosedStreamSink, and *Completer.setError, and StreamSinkCompleter.fromFuture.

test/stream_completer_test.dart

 // Copyright (c) 2015, 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.
 
 import "dart:async";
 
 import "package:async/async.dart" show StreamCompleter;
 import "package:test/test.dart";
 
 import "utils.dart";
@@ skipping 319 matching lines @@
     expect(controller.hasListener, isFalse);
     completer.setSourceStream(controller.stream);
     await flushMicrotasks();
     expect(controller.hasListener, isTrue);
     controller.addError(42);
     await done.then(unreachable("data"), onError: (error) {
       expect(error, 42);
     });
     expect(controller.hasListener, isFalse);
   });
+
+  group("setError()", () {
+    test("produces a stream that emits a single error", () {
+      var completer = new StreamCompleter();
+      completer.stream.listen(
+          unreachable("data"),
+          onError: expectAsync((error, stackTrace) {
+            expect(error, equals("oh no"));
+          }),
+          onDone: expectAsync(() {}));
+
+      completer.setError("oh no");
+    });
+
+    test("produces a stream that emits a single error on a later listen",
+        () async {
+      var completer = new StreamCompleter();
+      completer.setError("oh no");
+      await flushMicrotasks();
+
+      completer.stream.listen(
+          unreachable("data"),
+          onError: expectAsync((error, stackTrace) {
+            expect(error, equals("oh no"));
+          }),
+          onDone: expectAsync(() {}));
+    });
+  });
 }
 
 Stream<int> createStream() async* {
   yield 1;
   await flushMicrotasks();
   yield 2;
   await flushMicrotasks();
   yield 3;
   await flushMicrotasks();
   yield 4;
 }