Added toMultiSubscriber on Stream.

sdk/lib/async/stream_impl.dart

Index: sdk/lib/async/stream_impl.dart
diff --git a/sdk/lib/async/stream_impl.dart b/sdk/lib/async/stream_impl.dart
index 617a616b25037a38a629cf357a5b3800f4e8ea0d..7c2e7dff09cafc92b21276bfa540ac833813535f 100644
--- a/sdk/lib/async/stream_impl.dart
+++ b/sdk/lib/async/stream_impl.dart
@@ -403,7 +403,15 @@ abstract class _StreamImpl<T> extends Stream<T> {
  * arrive while still firing an old event.
  */
 class _SingleStreamImpl<T> extends _StreamImpl<T> {
-  _StreamSubscriptionImpl _subscriber = null;
+  _StreamListener _subscriber = null;
+
+  Stream<T> toMultiSubscriber() {
+    _MultiStreamImpl<T> multiStream = new _MultiStreamImpl();
+    listen(multiStream._add,

[floitsch, 2013/01/08 14:06:00]

as discussed: don't listen right away.

[Lasse Reichstein Nielsen, 2013/01/09 15:19:17]

Done.

+           onError: multiStream._signalError,
+           onDone: multiStream._close);
+    return multiStream;
+  }
 
   /** Whether one or more active subscribers have requested a pause. */
   bool get _isPaused => !_hasSubscribers || super._isPaused;
@@ -426,9 +434,9 @@ class _SingleStreamImpl<T> extends _StreamImpl<T> {
         this, onData, onError, onDone, unsubscribeOnError);
   }
 
-  void _addListener(_StreamSubscriptionImpl subscription) {
+  void _addListener(_StreamListener subscription) {
     if (_hasSubscribers) {
-      throw new StateError("Stream has already subscriber.");
+      throw new StateError("Stream already has subscriber.");
     }
     _subscriber = subscription;
     subscription._setSubscribed(0);
@@ -445,7 +453,7 @@ class _SingleStreamImpl<T> extends _StreamImpl<T> {
    * If an event is currently firing, the cancel is delayed
    * until after the subscriber has received the event.
    */
-  void _cancel(_StreamSubscriptionImpl subscriber) {
+  void _cancel(_StreamListener subscriber) {
     assert(identical(subscriber._source, this));
     // We allow unsubscribing the currently firing subscription during
     // the event firing, because it is indistinguishable from delaying it since
@@ -464,8 +472,8 @@ class _SingleStreamImpl<T> extends _StreamImpl<T> {
   }
 
   void _forEachSubscriber(
-      void action(_StreamSubscriptionImpl<T> subscription)) {
-    _StreamSubscriptionImpl subscription = _subscriber;
+      void action(_StreamListener<T> subscription)) {
+    _StreamListener subscription = _subscriber;
     assert(subscription != null);
     _startFiring();
     action(subscription);
@@ -514,6 +522,8 @@ class _MultiStreamImpl<T> extends _StreamImpl<T>
     _nextLink = _previousLink = this;
   }
 
+  Stream<T> toMultiSubscriber() => this;
+
   // ------------------------------------------------------------------
   // Helper functions that can be overridden in subclasses.