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 99eb92367b941d962cd38376a74b7c38db6c3d18..90ec1b3e0b026344eb8248530fe38fde94bf2c00 100644
--- a/sdk/lib/async/stream_impl.dart
+++ b/sdk/lib/async/stream_impl.dart
@@ -412,7 +412,11 @@ 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> asMultiSubscriberStream() {
+    return new _ForwardingMultiStream<T, T>().._source = this;
+  }
 
   /** Whether one or more active subscribers have requested a pause. */
   bool get _isPaused => !_hasSubscribers || super._isPaused;
@@ -435,9 +439,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);
@@ -457,7 +461,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
@@ -476,8 +480,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);
@@ -526,6 +530,8 @@ class _MultiStreamImpl<T> extends _StreamImpl<T>
     _nextLink = _previousLink = this;
   }
 
+  Stream<T> asMultiSubscriberStream() => this;
+
   // ------------------------------------------------------------------
   // Helper functions that can be overridden in subclasses.