Make StreamController be a StreamSink, not just an EventSink.

sdk/lib/async/stream_impl.dart

 // Copyright (c) 2012, 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.
 
 part of dart.async;
 
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
   void _addError(Object error);
@@ skipping 737 matching lines @@
     if (_controller == null) {
       // Return a dummy subscription backed by nothing, since
       // it won't ever receive any events.
       return new _DummyStreamSubscription<T>();
     }
     if (_subscription == null) {
       _subscription = _source.listen(_controller.add,
                                      onError: _controller.addError,
                                      onDone: _controller.close);
     }
-    return _controller.stream.listen(onData, onError: onError, onDone: onDone,
-                                     cancelOnError: cancelOnError);
+    if (onData == null) onData = _nullDataHandler;
+    if (onError == null) onError = _nullErrorHandler;
+    if (onDone == null) onDone = _nullDoneHandler;
+    cancelOnError = identical(true, cancelOnError);
+    return _controller._subscribe(onData, onError, onDone, cancelOnError);
   }
 
   void _onCancel() {
     bool shutdown = (_controller == null) || _controller.isClosed;
     if (_onCancelHandler != null) {
       _zone.executePeriodicCallbackGuarded(
           () => _onCancelHandler(new _BroadcastSubscriptionWrapper(this)));
     }
     if (shutdown) {
       if (_subscription != null) {
@@ skipping 229 matching lines @@
       _FutureImpl<bool> hasNext = _futureOrPrefetch;
       _clear();
       hasNext._setValue(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }