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.broadcast(sync: sync)
+                      : new StreamController(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.asBroadcastStream()
+                                                : c.stream);
+    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.broadcast(sync: sync)
+                      : new StreamController(sync: sync);
+    var expected = new Events()
+        ..add(42)..error("error")
+        ..add(1)..add(2)..add(3);
+    var stream = asBroadcast ? c.stream.asBroadcastStream() : c.stream;
+    var actual = new Events();
+    var sub;
+    // Cancel subscription after receiving "3" event.
+    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((_) {
+          Expect.listEquals(expected.events, actual.events);
+          // Close controller as well. It has no listener. If it is a broadcast
+          // stream, it will still be open, and we read the done" future before

[floitsch, 2013/06/27 15:15:19]

spurious '"'.

[Lasse Reichstein Nielsen, 2013/06/28 12:57:38]

Done.

+          // closing. A normal stream is already done when its listener cancels.
+          Future doneFuture = sink.done;
+          sink.close();
+          return doneFuture;
+        })
+        .then((_) {
+          // No change in events.
+          Expect.listEquals(expected.events, actual.events);
+          done();
+        });
+  });
+
+  test("$type-controller-sink-paused", () {
+    var done = expectAsync0((){});
+    var c = broadcast ? new StreamController.broadcast(sync: sync)
+                      : new StreamController(sync: sync);
+    var expected = new Events()
+        ..add(42)..error("error")
+        ..add(1)..add(2)..add(3);
+    if (!asBroadcast) {
+      // If asBroadcast, the done future fires when the done event leaves
+      // the controller's subscription. It has no way to know when the
+      // broadcast-stream's listeners are done.
+      expected..add(4)..add(5)..add(43)..close();

[floitsch, 2013/06/27 15:15:19]

If I understand correctly the broadcast subscription will buffer on its own for
15ms and then only add the remaining events.
So if we waited before doing the Expect.listEquals we would get the same result
on both sides?
If yes we should test it:

var delayedFuture;

in sub.pause: delayedFuture = ...;

in the final .then (where we execute the listEquals): check twice: once before,
and once after the delayedFuture is done.

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

[floitsch, 2013/06/27 15:15:19]

bad indentation

[Lasse Reichstein Nielsen, 2013/06/28 12:57:38]

Done.

+          }
+          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);
+          return sink.close();
+        })
+        .then((_) {

[Lasse Reichstein Nielsen, 2013/06/28 12:57:38]

I'll try adding a pause here (50 ms, just to be sure we're past the 15ms pause).

+          Expect.listEquals(expected.events, actual.events);
+          done();
+        });
+  });
+}
+
 main() {
   testController();
   testSingleController();
   testExtraMethods();
   testPause();
   testRethrow();
   testBroadcastController();
+  testSink(sync: true, broadcast: false, asBroadcast: false);
+  testSink(sync: true, broadcast: false, asBroadcast: true);
+  testSink(sync: true, broadcast: true, asBroadcast: false);
+  testSink(sync: false, broadcast: false, asBroadcast: false);
+  testSink(sync: false, broadcast: false, asBroadcast: true);
+  testSink(sync: false, broadcast: true, asBroadcast: false);

[floitsch, 2013/06/27 15:15:19]

add test, when the addStream produces errors.

 }