Avoid the _onSubscriptionStateChange being called twice in some cases.

sdk/lib/async/stream_pipe.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;
 
 /** Utility function to create an [AsyncError] if [error] isn't one already. */
 AsyncError _asyncError(Object error, Object stackTrace, [AsyncError cause]) {
   if (error is AsyncError) return error;
   if (cause == null) return new AsyncError(error, stackTrace);
@@ skipping 22 matching lines @@
 _cancelAndError(StreamSubscription subscription, _FutureImpl future) =>
   (AsyncError error) {
     subscription.cancel();
     future._setError(error);
   };
 
 
 /**
  * A wrapper around a stream that allows independent subscribers.
  *
- * By default [this] subscribes to [_source] and forwards all events to its own
+ * By default [:this:] subscribes to [_source] and forwards all events to its own

[floitsch, 2013/01/14 16:31:24]

80chars.

If 'this' is not available yet, we should probably use something else.
By default the stream subs...

[Lasse Reichstein Nielsen, 2013/01/15 08:52:54]

It's not that "this" is not available *yet*, it is that it's not an identifier
at all. You can't write generic expressions inside [...], only identifiers (and
qualified identifiers should also be allowed eventually, but it's not in the
spec yet).

  * subscribers. It does not subscribe until there is a subscriber, and
  * unsubscribes again when there are no subscribers left.
  *
  * The events are passed through the [_handleData], [_handleError] and
  * [_handleDone] methods. Subclasses are supposed to add handling of some of
  * the events by overriding these methods.
  *
  * This class is intended for internal use only.
  */
 class _ForwardingMultiStream<S, T> extends _MultiStreamImpl<T> {
@@ skipping 446 matching lines @@
         _signalError(_asyncError(e, s));
         return null;
       }
       if (!isEqual) {
         _add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }