Add public-facing method and class that allows intercepting stream events.

sdk/lib/async/merge_stream.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;
 
 class _SupercedeEntry<T> {
   final _SupercedeStream stream;
   Stream<T> source;
   StreamSubscription subscription = null;
@@ skipping 141 matching lines @@
  * activated.
  *
  * If the source completes, the entry is removed from [stream].
  */
 class _CycleEntry<T> {
   final _CyclicScheduleStream stream;
   /** A single source stream for the [_CyclicScheduleStream]. */
   Stream source;
   /** The active subscription, if any. */
   StreamSubscription subscription = null;
+  /** Whether the subscription is currently paused. */
+  bool isPaused = false;
   /** Next entry in a linked list of entries. */
   _CycleEntry next;
 
   _CycleEntry(this.stream, this.source);
 
   void cancel() {
-    // This method may be called event if this entry has never been activated.
+    // This method may be called even if this entry has never been activated.
     if (subscription != null) {
       subscription.cancel();
       subscription = null;
+      isPaused = false;
     }
   }
 
   void pause() {
     ensureSubscribed();
-    if (!subscription.isPaused) {
+    if (!isPaused) {
       subscription.pause();
+      isPaused = true;
     }
   }
 
   void activate() {
     ensureSubscribed();
-    if (subscription.isPaused) {
+    if (isPaused) {
+      isPaused = false;
       subscription.resume();
     }
   }
 
   void ensureSubscribed() {
     if (subscription == null) {
       subscription =
           source.listen(stream._onData,
                         onError: stream._signalError,
                         onDone: stream._onDone);
@@ skipping 78 matching lines @@
     if (_currentEntry.next == null) {
       _close();
       _currentEntry = _lastEntry = null;
     } else {
       // Remove the current entry from the list now that it's complete.
       _currentEntry = _currentEntry.next;
       _currentEntry.activate();
     }
   }
 }