Change singleSubscription/multiSubscription to normal/broadcast.

sdk/lib/async/stream_controller.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;
 
 // -------------------------------------------------------------------
 // Default implementation of a stream with a controller for adding
 // events to the stream.
 // -------------------------------------------------------------------
 
 /**
  * A controller and the stream it controls.
  *
  * This controller allows sending data, error and done events on
  * its [stream].
  * This class can be used to create a simple stream that others
  * can listen on, and to push events to that stream.
  *
  * It's possible to check whether the stream is paused or not, and whether
  * it has subscribers or not, as well as getting a callback when either of
  * these change.
  */
-class StreamController<T> extends Stream<T> implements StreamSink<T> {

[floitsch, 2013/01/22 14:37:56]

This looks like an unrelated change.
At least update the description?

[Lasse Reichstein Nielsen, 2013/01/22 15:33:37]

Description updated.

+class StreamController<T> implements StreamSink<T> {
   _StreamImpl<T> _stream;

[floitsch, 2013/01/22 14:37:56]

Should we make this simply final and initialize it in the constructor?

[Lasse Reichstein Nielsen, 2013/01/22 15:33:37]

Yes. We can do that now that the state-change-callbacks are not instance
methods.

-  Stream<T> get stream => _stream;
 
   /**
    * A controller with a [stream] that supports multiple subscribers.
    *
    * The [onPauseStateChange] function is called when the stream becomes
    * paused or resumes after being paused. The current pause state can
    * be read from [isPaused]. Ignored if [:null:].
    *
    * The [onSubscriptionStateChange] function is called when the stream
    * receives its first listener or loses its last. The current subscription
@@ skipping 16 matching lines @@
    * The [onSubscriptionStateChange] function is called when the stream
    * receives its first listener or loses its last. The current subscription
    * state can be read from [hasSubscribers]. Ignored if [:null:].
    */
   StreamController({void onPauseStateChange(),
                     void onSubscriptionStateChange()}) {
     _stream = new _SingleControllerStream<T>(onSubscriptionStateChange,
                                              onPauseStateChange);
   }
 
-  bool get isSingleSubscription => _stream.isSingleSubscription;
-
-  Stream<T> asMultiSubscriptionStream() => _stream.asMultiSubscriptionStream();
-
-  StreamSubscription listen(void onData(T data),
-                            { void onError(AsyncError error),
-                              void onDone(),
-                              bool unsubscribeOnError}) {
-    return _stream.listen(onData,
-                          onError: onError,
-                          onDone: onDone,
-                          unsubscribeOnError: unsubscribeOnError);
-  }
+  /** The stream that this controller is controlling. */
+  Stream<T> get stream => _stream;
 
   /**
    * Returns a view of this object that only exposes the [StreamSink] interface.
    */
   StreamSink<T> get sink => new StreamSinkView<T>(this);
 
   /** Whether one or more active subscribers have requested a pause. */
   bool get isPaused => _stream._isPaused;
 
   /** Whether there are currently any subscribers on this [Stream]. */
@@ skipping 26 matching lines @@
     _stream._signalError(asyncError);
   }
 
   /**
    * Send or enqueue a "done" message.
    *
    * The "done" message should be sent at most once by a stream, and it
    * should be the last message sent.
    */
   void close() { _stream._close(); }
-
-  void forEachSubscriber(void action(_StreamSubscriptionImpl<T> subscription)) {
-    _stream._forEachSubscriber(() {
-      try {
-        action();
-      } on AsyncError catch (e) {
-        e.throwDelayed();
-      } catch (e, s) {
-        new AsyncError(e, s).throwDelayed();
-      }
-    });
-  }
 }
 
 typedef void _NotificationHandler();
 
 class _MultiControllerStream<T> extends _MultiStreamImpl<T> {
   _NotificationHandler _subscriptionHandler;
   _NotificationHandler _pauseHandler;
 
   _MultiControllerStream(this._subscriptionHandler, this._pauseHandler);
 
@@ skipping 13 matching lines @@
   _SingleControllerStream(this._subscriptionHandler, this._pauseHandler);
 
   void _onSubscriptionStateChange() {
     if (_subscriptionHandler != null) _subscriptionHandler();
   }
 
   void _onPauseStateChange() {
     if (_pauseHandler != null) _pauseHandler();
   }
 }