Update dart:async to match the Dart repo.

tool/input_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;
 
 /** Runs user code and takes actions depending on success or failure. */
 _runUserCode(userCode(),
              onSuccess(value),
              onError(error, StackTrace stackTrace)) {
@@ skipping 29 matching lines @@
                                     _Future future,
                                     error, StackTrace stackTrace) {
   AsyncError replacement = Zone.current.errorCallback(error, stackTrace);
   if (replacement != null) {
     error = _nonNullError(replacement.error);
     stackTrace = replacement.stackTrace;
   }
   _cancelAndError(subscription, future, error, stackTrace);
 }
 
+typedef void _ErrorCallback(error, StackTrace stackTrace);
+
 /** Helper function to make an onError argument to [_runUserCode]. */
-_cancelAndErrorClosure(StreamSubscription subscription, _Future future) =>
-  ((error, StackTrace stackTrace) => _cancelAndError(
-      subscription, future, error, stackTrace));
+_ErrorCallback _cancelAndErrorClosure(
+    StreamSubscription subscription, _Future future) {
+  return (error, StackTrace stackTrace) {
+    _cancelAndError(subscription, future, error, stackTrace);
+  };
+}
 
 /** Helper function to cancel a subscription and wait for the potential future,
   before completing with a value. */
 void _cancelAndValue(StreamSubscription subscription, _Future future, value) {
   var cancelFuture = subscription.cancel();
   if (cancelFuture is Future) {
     cancelFuture.whenComplete(() => future._complete(value));
   } else {
     future._complete(value);
   }
@@ skipping 29 matching lines @@
       Function onError,
       void onDone(),
       bool cancelOnError) {
     return new _ForwardingStreamSubscription<S, T>(
         this, onData, onError, onDone, cancelOnError);
   }
 
   // Override the following methods in subclasses to change the behavior.
 
   void _handleData(S data, _EventSink<T> sink) {
-    dynamic outputData = data;
-    sink._add(outputData);
+    sink._add(data as Object /*=T*/);
   }
 
   void _handleError(error, StackTrace stackTrace, _EventSink<T> sink) {
     sink._addError(error, stackTrace);
   }
 
   void _handleDone(_EventSink<T> sink) {
     sink._close();
   }
 }
@@ skipping 39 matching lines @@
 
   void _onResume() {
     if (_subscription == null) return;
     _subscription.resume();
   }
 
   Future _onCancel() {
     if (_subscription != null) {
       StreamSubscription subscription = _subscription;
       _subscription = null;
-      subscription.cancel();
+      return subscription.cancel();
     }
     return null;
   }
 
   // Methods used as listener on source subscription.
 
   void _handleData(S data) {
     _stream._handleData(data, this);
   }
 
@@ skipping 42 matching lines @@
   }
 }
 
 
 typedef T _Transformation<S, T>(S value);
 
 /**
  * A stream pipe that converts data events before passing them on.
  */
 class _MapStream<S, T> extends _ForwardingStream<S, T> {
-  final _Transformation _transform;
+  final _Transformation<S, T> _transform;
 
   _MapStream(Stream<S> source, T transform(S event))
       : this._transform = transform, super(source);
 
   void _handleData(S inputEvent, _EventSink<T> sink) {
     T outputEvent;
     try {
       outputEvent = _transform(inputEvent);
     } catch (e, s) {
       _addErrorWithReplacement(sink, e, s);
@@ skipping 63 matching lines @@
         return;
       }
     } else {
       sink._addError(error, stackTrace);
     }
   }
 }
 
 
 class _TakeStream<T> extends _ForwardingStream<T, T> {
-  int _remaining;
+  final int _count;
 
   _TakeStream(Stream<T> source, int count)
-      : this._remaining = count, super(source) {
+      : this._count = count, super(source) {
     // This test is done early to avoid handling an async error
     // in the _handleData method.
     if (count is! int) throw new ArgumentError(count);
   }
 
+  StreamSubscription<T> _createSubscription(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
+    return new _StateStreamSubscription<T>(
+        this, onData, onError, onDone, cancelOnError, _count);
+  }
+
   void _handleData(T inputEvent, _EventSink<T> sink) {
-    if (_remaining > 0) {
+    _StateStreamSubscription<T> subscription = sink;
+    int count = subscription._count;
+    if (count > 0) {
       sink._add(inputEvent);
-      _remaining -= 1;
-      if (_remaining == 0) {
+      count -= 1;
+      subscription._count = count;
+      if (count == 0) {
         // Closing also unsubscribes all subscribers, which unsubscribes
         // this from source.
         sink._close();
       }
     }
   }
 }
 
+/**
+ * A [_ForwardingStreamSubscription] with one extra state field.
+ *
+ * Use by several different classes, some storing an integer, others a bool.
+ */
+class _StateStreamSubscription<T> extends _ForwardingStreamSubscription<T, T> {
+  // Raw state field. Typed access provided by getters and setters below.
+  var _sharedState;
+
+  _StateStreamSubscription(_ForwardingStream<T, T> stream, void onData(T data),
+                           Function onError, void onDone(),
+                           bool cancelOnError, this._sharedState)
+      : super(stream, onData, onError, onDone, cancelOnError);
+
+  bool get _flag => _sharedState;
+  void set _flag(bool flag) { _sharedState = flag; }
+  int get _count => _sharedState;
+  void set _count(int count) { _sharedState = count; }
+}
+
 
 class _TakeWhileStream<T> extends _ForwardingStream<T, T> {
   final _Predicate<T> _test;
 
   _TakeWhileStream(Stream<T> source, bool test(T value))
       : this._test = test, super(source);
 
   void _handleData(T inputEvent, _EventSink<T> sink) {
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
       _addErrorWithReplacement(sink, e, s);
       // The test didn't say true. Didn't say false either, but we stop anyway.
       sink._close();
       return;
     }
     if (satisfies) {
       sink._add(inputEvent);
     } else {
       sink._close();
     }
   }
 }
 
 class _SkipStream<T> extends _ForwardingStream<T, T> {
-  int _remaining;
+  final int _count;
 
   _SkipStream(Stream<T> source, int count)
-      : this._remaining = count, super(source) {
+      : this._count = count, super(source) {
     // This test is done early to avoid handling an async error
     // in the _handleData method.
     if (count is! int || count < 0) throw new ArgumentError(count);
   }
 
+  StreamSubscription<T> _createSubscription(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
+    return new _StateStreamSubscription<T>(
+        this, onData, onError, onDone, cancelOnError, _count);
+  }
+
   void _handleData(T inputEvent, _EventSink<T> sink) {
-    if (_remaining > 0) {
-      _remaining--;
+    _StateStreamSubscription<T> subscription = sink;
+    int count = subscription._count;
+    if (count > 0) {
+      subscription._count = count - 1;
       return;
     }
     sink._add(inputEvent);
   }
 }
 
 class _SkipWhileStream<T> extends _ForwardingStream<T, T> {
   final _Predicate<T> _test;
-  bool _hasFailed = false;
 
   _SkipWhileStream(Stream<T> source, bool test(T value))
       : this._test = test, super(source);
 
+  StreamSubscription<T> _createSubscription(
+      void onData(T data),
+      Function onError,
+      void onDone(),
+      bool cancelOnError) {
+    return new _StateStreamSubscription<T>(
+        this, onData, onError, onDone, cancelOnError, false);
+  }
+
   void _handleData(T inputEvent, _EventSink<T> sink) {
-    if (_hasFailed) {
+    _StateStreamSubscription<T> subscription = sink;
+    bool hasFailed = subscription._flag;
+    if (hasFailed) {
       sink._add(inputEvent);
       return;
     }
     bool satisfies;
     try {
       satisfies = _test(inputEvent);
     } catch (e, s) {
       _addErrorWithReplacement(sink, e, s);
       // A failure to return a boolean is considered "not matching".
-      _hasFailed = true;
+      subscription._flag = true;
       return;
     }
     if (!satisfies) {
-      _hasFailed = true;
+      subscription._flag = true;
       sink._add(inputEvent);
     }
   }
 }
 
 typedef bool _Equality<T>(T a, T b);
 
 class _DistinctStream<T> extends _ForwardingStream<T, T> {
   static var _SENTINEL = new Object();
 
   _Equality<T> _equals;
   var _previous = _SENTINEL;
 
   _DistinctStream(Stream<T> source, bool equals(T a, T b))
       : _equals = equals, super(source);
 
   void _handleData(T inputEvent, _EventSink<T> sink) {
     if (identical(_previous, _SENTINEL)) {
       _previous = inputEvent;
       return sink._add(inputEvent);
     } else {
       bool isEqual;
       try {
         if (_equals == null) {
           isEqual = (_previous == inputEvent);
         } else {
-          isEqual = _equals(_previous, inputEvent);
+          isEqual = _equals(_previous as Object /*=T*/, inputEvent);
         }
       } catch (e, s) {
         _addErrorWithReplacement(sink, e, s);
         return null;
       }
       if (!isEqual) {
         sink._add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }