Make StreamController be a StreamSink, not just an EventSink.

tests/lib/async/stream_controller_async_test.dart

Index: tests/lib/async/stream_controller_async_test.dart
diff --git a/tests/lib/async/stream_controller_async_test.dart b/tests/lib/async/stream_controller_async_test.dart
index 7c134055bc249134739f8f69a72e0a50b0bb5738..450be2a635d80e2478497f0acb79872a70e74b0e 100644
--- a/tests/lib/async/stream_controller_async_test.dart
+++ b/tests/lib/async/stream_controller_async_test.dart
@@ -464,7 +464,6 @@ void testBroadcastController() {
 
   test("broadcast-controller-individual-pause", () {
     StreamProtocolTest test = new StreamProtocolTest.broadcast();
-    test.trace = true;
     var sub1;
     test..expectListen()
         ..expectData(42)
@@ -498,6 +497,94 @@ void testBroadcastController() {
   });
 }
 
+void testSink(bool sync, bool broadcast, bool asBroadcast) {
+  String type = "${sync?"S":"A"}${broadcast?"B":"S"}${asBroadcast?"aB":""}";
+  test("$type-controller-sink", () {
+    var done = expectAsync0((){});
+    var c = broadcast ? new StreamController(sync: sync)
+                      : new StreamController.broadcast(sync: sync);
+    var expected = new Events()
+        ..add(42)..error("error")
+        ..add(1)..add(2)..add(3)..add(4)..add(5)
+        ..add(43)..close();
+    var actual = new Events.capture(asBroadcast ? c.stream
+                                                : c.stream.asBroadcastStream());
+    var sink = c.sink;
+    sink.add(42);
+    sink.addError("error");
+    sink.addStream(new Stream.fromIterable([1, 2, 3, 4, 5]))
+        .then((_) {
+          sink.add(43);
+          return sink.close();
+        })
+        .then((_) {
+          Expect.listEquals(expected.events, actual.events);
+          done();
+        });
+  });
+
+  test("$type-controller-sink-canceled", () {
+    var done = expectAsync0((){});
+    var c = broadcast ? new StreamController(sync: sync)
+                      : new StreamController.broadcast(sync: sync);
+    var expected = new Events()
+        ..add(42)..error("error")
+        ..add(1)..add(2)..add(3);
+    var stream = asBroadcast ? c.stream : c.stream.asBroadcastStream();
+    var actual = new Events();
+    var sub;
+    sub = stream.listen((v) {
+      if (v == 3) sub.cancel();
+      actual.add(v);
+    }, onError: actual.error);
+    var sink = c.sink;
+    sink.add(42);
+    sink.addError("error");
+    sink.addStream(new Stream.fromIterable([1, 2, 3, 4, 5]))
+        .then((_) {
+          return sink.done;
+        })
+        .then((_) {
+          Expect.listEquals(expected.events, actual.events);
+          done();
+        });
+  });
+
+  test("$type-controller-sink-paused", () {
+    var done = expectAsync0((){});
+    var c = broadcast ? new StreamController(sync: sync)
+                      : new StreamController.broadcast(sync: sync);
+    var expected = new Events()
+        ..add(42)..error("error")
+        ..add(1)..add(2)..add(3);
+        // This is wrong, it should get more events before being done.

[floitsch, 2013/06/06 15:08:29]

Don't understand.
The subscription is only paused. So the other elements (4, 5, 43) should be in
the expected list. no?

+    var stream = asBroadcast ? c.stream : c.stream.asBroadcastStream();
+    var actual = new Events();
+    var sub;
+    sub = stream.listen((v) {
+          if (v == 3) {
+            sub.pause(new Future.delayed(const Duration(milliseconds: 15), () {}));

[floitsch, 2013/06/06 15:08:29]

long line.

+          }
+          actual.add(v);
+        },
+        onError: actual.error,
+        onDone: actual.close);
+    var sink = c.sink;
+    sink.add(42);
+    sink.addError("error");
+    sink.addStream(new Stream.fromIterable([1, 2, 3, 4, 5]))
+        .then((_) {
+          sink.add(43);
+          sink.close();
+          return sink.done;
+        })
+        .then((_) {
+          Expect.listEquals(expected.events, actual.events);
+          done();
+        });
+  });
+}
+
 main() {
   testController();
   testSingleController();
@@ -505,4 +592,10 @@ main() {
   testPause();
   testRethrow();
   testBroadcastController();
+  testSink(true, false, false);
+  testSink(false, false, false);
+  testSink(true, false, true);
+  testSink(false, false, true);
+  testSink(true, true, false);
+  testSink(false, true, false);
 }