Make StreamController be a StreamSink, not just an EventSink.

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 446 matching lines @@
     ..expectData(37, () {
       test.close();
     })
     ..expectDone()
     ..expectCancel(test.terminate);
     test.listen();
   });
 
   test("broadcast-controller-individual-pause", () {
     StreamProtocolTest test = new StreamProtocolTest.broadcast();
-    test.trace = true;
     var sub1;
     test..expectListen()
         ..expectData(42)
         ..expectData(42, () { sub1.pause(); })
         ..expectData(43, () {
       sub1.cancel();
       test.listen();
       test.add(44);
       test.expectData(44);
       test.expectData(44, test.terminate);
@@ skipping 13 matching lines @@
       test.add(87);
       sub.cancel();
     });
     test.expectCancel(() {
       test.add(37);
       test.terminate();
     });
   });
 }
 
+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();
   testExtraMethods();
   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);
 }