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

test/null_stream_sink_test.dart

+// Copyright (c) 2016, 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";
+import "package:test/test.dart";
+
+import "utils.dart";
+
+void main() {
+  group("constructors", () {
+    test("done defaults to a completed future", () {
+      var sink = new NullStreamSink();
+      expect(sink.done, completes);
+    });
+
+    test("a custom future may be passed to done", () async {
+      var completer = new Completer();
+      var sink = new NullStreamSink(done: completer.future);
+
+      var doneFired = false;
+      sink.done.then((_) {
+        doneFired = true;
+      });
+      await flushMicrotasks();
+      expect(doneFired, isFalse);
+
+      completer.complete();
+      await flushMicrotasks();
+      expect(doneFired, isTrue);
+    });
+
+    test("NullStreamSink.error passes an error to done", () {
+      var sink = new NullStreamSink.error("oh no");
+      expect(sink.done, throwsA("oh no"));
+    });
+  });
+
+  group("events", () {
+    test("are silently dropped before close", () {
+      var sink = new NullStreamSink();
+      sink.add(1);
+      sink.addError("oh no");
+    });
+
+    test("throw StateErrors after close", () {
+      var sink = new NullStreamSink();
+      expect(sink.close(), completes);
+
+      expect(() => sink.add(1), throwsStateError);
+      expect(() => sink.addError("oh no"), throwsStateError);
+      expect(() => sink.addStream(new Stream.empty()), throwsStateError);
+    });
+
+    group("addStream", () {
+      test("listens to the stream then cancels immediately", () async {
+        var sink = new NullStreamSink();
+        var canceled = false;
+        var controller = new StreamController(onCancel: () {
+          canceled = true;
+        });
+
+        expect(sink.addStream(controller.stream), completes);
+        await flushMicrotasks();
+        expect(canceled, isTrue);
+      });
+
+      test("returns the cancel future", () async {
+        var completer = new Completer();
+        var sink = new NullStreamSink();
+        var controller = new StreamController(onCancel: () => completer.future);
+
+        var addStreamFired = false;
+        sink.addStream(controller.stream).then((_) {
+          addStreamFired = true;
+        });
+        await flushMicrotasks();
+        expect(addStreamFired, isFalse);
+
+        completer.complete();
+        await flushMicrotasks();
+        expect(addStreamFired, isTrue);
+      });
+
+      test("pipes errors from the cancel future through addStream", () async {
+        var sink = new NullStreamSink();
+        var controller = new StreamController(onCancel: () => throw "oh no");
+        expect(sink.addStream(controller.stream), throwsA("oh no"));
+      });
+
+      test("causes events to throw StateErrors until the future completes",
+          () async {
+        var sink = new NullStreamSink();
+        var future = sink.addStream(new Stream.empty());
+        expect(() => sink.add(1), throwsStateError);
+        expect(() => sink.addError("oh no"), throwsStateError);
+        expect(() => sink.addStream(new Stream.empty()), throwsStateError);
+
+        await future;
+        sink.add(1);
+        sink.addError("oh no");
+        expect(sink.addStream(new Stream.empty()), completes);
+      });
+    });
+  });
+
+  test("close returns the done future", () {
+    var sink = new NullStreamSink.error("oh no");
+    expect(sink.close(), throwsA("oh no"));
+  });
+}