Added toMultiSubscriber on Stream.

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;
 
 // States shared by single/multi stream implementations.
 
 /// Initial and default state where the stream can receive and send events.
 const int _STREAM_OPEN = 0;
@@ skipping 385 matching lines @@
  * * [_onSubscriberStateChange]: Called when receiving the first subscriber or
  *                               when losing the last subscriber.
  * * [_onPauseStateChange]: Called when entering or leaving paused mode.
  * * [_hasSubscribers]: Test whether there are currently any subscribers.
  * * [_isPaused]: Test whether the stream is currently paused.
  * The user should not add new events while the stream is paused, but if it
  * happens anyway, the stream will enqueue the events just as when new events
  * 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;
 
   /** Whether there is currently a subscriber on this [Stream]. */
   bool get _hasSubscribers => _subscriber != null;
 
   // -------------------------------------------------------------------
   // Internal implementation.
 
   /**
    * Create the new subscription object.
    */
   _StreamSubscriptionImpl<T> _createSubscription(
       void onData(T data),
       void onError(AsyncError error),
       void onDone(),
       bool unsubscribeOnError) {
     return new _StreamSubscriptionImpl<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);
     _onSubscriptionStateChange();
     // TODO(floitsch): Should this be delayed?
     _handlePendingEvents();
   }
 
   /**
    * Handle a cancel requested from a [_StreamSubscriptionImpl].
    *
    * This method is called from [_StreamSubscriptionImpl.cancel].
    *
    * 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
     // that event has already received the event.
     if (!identical(_subscriber, subscriber)) {
       // You may unsubscribe more than once, only the first one counts.
       return;
     }
     _subscriber = null;
     int timesPaused = subscriber._setUnsubscribed();
     _updatePauseCount(-timesPaused);
     if (timesPaused > 0) {
       _onPauseStateChange();
     }
     _onSubscriptionStateChange();
   }
 
   void _forEachSubscriber(
-      void action(_StreamSubscriptionImpl<T> subscription)) {
-    _StreamSubscriptionImpl subscription = _subscriber;
+      void action(_StreamListener<T> subscription)) {
+    _StreamListener subscription = _subscriber;
     assert(subscription != null);
     _startFiring();
     action(subscription);
     _endFiring();
   }
 }
 
 // -------------------------------------------------------------------
 // Default implementation of a stream with subscribers.
 // -------------------------------------------------------------------
@@ skipping 28 matching lines @@
 class _MultiStreamImpl<T> extends _StreamImpl<T>
                           implements _InternalLinkList {
   // Link list implementation (mixin when possible).
   _InternalLink _nextLink;
   _InternalLink _previousLink;
 
   _MultiStreamImpl() {
     _nextLink = _previousLink = this;
   }
 
+  Stream<T> toMultiSubscriber() => this;
+
   // ------------------------------------------------------------------
   // Helper functions that can be overridden in subclasses.
 
   /** Whether there are currently any subscribers on this [Stream]. */
   bool get _hasSubscribers => !_InternalLinkList.isEmpty(this);
 
   /**
    * Create the new subscription object.
    */
   _StreamListener<T> _createSubscription(
@@ skipping 481 matching lines @@
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
     }
     if (_timer != null) {
       _timer.cancel();
       _timer = null;
     }
     _pauseCount = 0;
   }
 }