Make Stream transformation respect the single/multi subscriber nature of the source.

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 403 matching lines @@
  * * [_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> {
   _StreamListener _subscriber = null;
 
   Stream<T> asMultiSubscriberStream() {
-    return new _ForwardingMultiStream<T, T>().._source = this;
+    return new _SingleStreamMultiplexer<T>(this);
   }
 
+  bool get isSingleSubscription => true;
+
   /** 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.
 
   /**
@@ skipping 97 matching lines @@
 
   _MultiStreamImpl() {
     _nextLink = _previousLink = this;
   }
 
   Stream<T> asMultiSubscriberStream() => this;
 
   // ------------------------------------------------------------------
   // Helper functions that can be overridden in subclasses.
 
+  bool get isSingleSubscription => false;

[floitsch, 2013/01/14 16:09:09]

I don't think anybody wants to override this method :)
Move it above the comments.

[Lasse Reichstein Nielsen, 2013/01/15 07:19:51]

Done.

+
   /** Whether there are currently any subscribers on this [Stream]. */
   bool get _hasSubscribers => !_InternalLinkList.isEmpty(this);
 
   /**
    * Create the new subscription object.
    */
   _StreamListener<T> _createSubscription(
       void onData(T data),
       void onError(AsyncError error),
       void onDone(),
@@ skipping 532 matching lines @@
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
     }
     if (_timer != null) {
       _timer.cancel();
       _timer = null;
     }
     _pauseCount = 0;
   }
 }
+
+class _SingleStreamMultiplexer<T> extends _MultiStreamImpl<T> {
+  final _SingleStreamImpl<T> _source;
+  StreamSubscription<T> _subscription;
+
+  _SingleStreamMultiplexer(this._source);
+
+  void _onPauseStateChange() {
+    if (_isPaused) {
+      if (_subscription != null) {
+        _subscription.pause();
+      }
+    } else {
+      if (_subscription != null) {
+        _subscription.resume();
+      }
+    }
+  }
+
+  /**
+    * Subscribe or unsubscribe on [_source] depending on whether
+    * [_stream] has subscribers.
+    */
+  void _onSubscriptionStateChange() {
+    if (_hasSubscribers) {
+      assert(_subscription == null);
+      _subscription = _source.listen(this._add,
+                                     onError: this._signalError,
+                                     onDone: this._close);
+    } else {
+      // TODO(lrn): Check why this can happen.
+      if (_subscription == null) return;
+      _subscription.cancel();
+      _subscription = null;
+    }
+  }
+}