Update dart:async to match the Dart repo.

tool/input_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;
 
 /** Abstract and private interface for a place to put events. */
 abstract class _EventSink<T> {
   void _add(T data);
   void _addError(Object error, StackTrace stackTrace);
@@ skipping 88 matching lines @@
 
   // TODO(floitsch): reuse another field
   /** The future [_onCancel] may return. */
   Future _cancelFuture;
 
   /**
    * Queue of pending events.
    *
    * Is created when necessary, or set in constructor for preconfigured events.
    */
-  _PendingEvents _pending;
+  _PendingEvents<T> _pending;
 
   _BufferingStreamSubscription(void onData(T data),
                                Function onError,
                                void onDone(),
                                bool cancelOnError)
       : _state = (cancelOnError ? _STATE_CANCEL_ON_ERROR : 0) {
     this.onData(onData);
     this.onError(onError);
     this.onDone(onDone);
   }
 
   /**
    * Sets the subscription's pending events object.
    *
    * This can only be done once. The pending events object is used for the
    * rest of the subscription's life cycle.
    */
-  void _setPendingEvents(_PendingEvents pendingEvents) {
+  void _setPendingEvents(_PendingEvents<T> pendingEvents) {
     assert(_pending == null);
     if (pendingEvents == null) return;
     _pending = pendingEvents;
     if (!pendingEvents.isEmpty) {
       _state |= _STATE_HAS_PENDING;
       _pending.schedule(this);
     }
   }
 
-  /**
-   * Extracts the pending events from a canceled stream.
-   *
-   * This can only be done during the [_onCancel] method call. After that,
-   * any remaining pending events will be cleared.
-   */
-  _PendingEvents _extractPending() {
-    assert(_isCanceled);
-    _PendingEvents events = _pending;
-    _pending = null;
-    return events;
-  }
-
   // StreamSubscription interface.
 
   void onData(void handleData(T event)) {
     if (handleData == null) handleData = _nullDataHandler;
-    _onData = _zone.registerUnaryCallback(handleData);
+    // TODO(floitsch): the return type should be 'void', and the type
+    // should be inferred.
+    _onData = _zone.registerUnaryCallback/*<dynamic, T>*/(handleData);
   }
 
   void onError(Function handleError) {
     if (handleError == null) handleError = _nullErrorHandler;
-    _onError = _registerErrorHandler(handleError, _zone);
+    _onError = _registerErrorHandler/*<T>*/(handleError, _zone);
   }
 
   void onDone(void handleDone()) {
     if (handleDone == null) handleDone = _nullDoneHandler;
     _onDone = _zone.registerCallback(handleDone);
   }
 
   void pause([Future resumeSignal]) {
     if (_isCanceled) return;
     bool wasPaused = _isPaused;
@@ skipping 25 matching lines @@
   Future cancel() {
     // The user doesn't want to receive any further events. If there is an
     // error or done event pending (waiting for the cancel to be done) discard
     // that event.
     _state &= ~_STATE_WAIT_FOR_CANCEL;
     if (_isCanceled) return _cancelFuture;
     _cancel();
     return _cancelFuture;
   }
 
-  Future asFuture([var futureValue]) {
-    _Future<T> result = new _Future<T>();
+  Future/*<E>*/ asFuture/*<E>*/([var/*=E*/ futureValue]) {
+    _Future/*<E>*/ result = new _Future/*<E>*/();
 
     // Overwrite the onDone and onError handlers.
     _onDone = () { result._complete(futureValue); };
     _onError = (error, stackTrace) {
       cancel();
       result._completeError(error, stackTrace);
     };
 
     return result;
   }
@@ skipping 45 matching lines @@
   }
 
   // _EventSink interface.
 
   void _add(T data) {
     assert(!_isClosed);
     if (_isCanceled) return;
     if (_canFire) {
       _sendData(data);
     } else {
-      _addPending(new _DelayedData(data));
+      _addPending(new _DelayedData<dynamic /*=T*/>(data));
     }
   }
 
   void _addError(Object error, StackTrace stackTrace) {
     if (_isCanceled) return;
     if (_canFire) {
       _sendError(error, stackTrace);  // Reports cancel after sending.
     } else {
       _addPending(new _DelayedError(error, stackTrace));
     }
@@ skipping 29 matching lines @@
 
   // Handle pending events.
 
   /**
    * Add a pending event.
    *
    * If the subscription is not paused, this also schedules a firing
    * of pending events later (if necessary).
    */
   void _addPending(_DelayedEvent event) {
-    _StreamImplEvents pending = _pending;
-    if (_pending == null) pending = _pending = new _StreamImplEvents();
+    _StreamImplEvents<T> pending = _pending;
+    if (_pending == null) {
+      pending = _pending = new _StreamImplEvents<dynamic /*=T*/>();
+    }
     pending.add(event);
     if (!_hasPending) {
       _state |= _STATE_HAS_PENDING;
       if (!_isPaused) {
         _pending.schedule(this);
       }
     }
   }
 
   /* _EventDispatch interface. */
 
   void _sendData(T data) {
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
     bool wasInputPaused = _isInputPaused;
     _state |= _STATE_IN_CALLBACK;
     _zone.runUnaryGuarded(_onData, data);
     _state &= ~_STATE_IN_CALLBACK;
     _checkState(wasInputPaused);
   }
 
-  void _sendError(Object error, StackTrace stackTrace) {
+  void _sendError(var error, StackTrace stackTrace) {
     assert(!_isCanceled);
     assert(!_isPaused);
     assert(!_inCallback);
     bool wasInputPaused = _isInputPaused;
 
     void sendError() {
       // If the subscription has been canceled while waiting for the cancel
       // future to finish we must not report the error.
       if (_isCanceled && !_waitsForCancel) return;
       _state |= _STATE_IN_CALLBACK;
-      if (_onError is ZoneBinaryCallback) {
-        _zone.runBinaryGuarded(_onError, error, stackTrace);
+      if (_onError is ZoneBinaryCallback<dynamic, Object, StackTrace>) {
+        ZoneBinaryCallback<dynamic, Object, StackTrace> errorCallback = _onError
+            as Object /*=ZoneBinaryCallback<dynamic, Object, StackTrace>*/;
+        _zone.runBinaryGuarded(errorCallback, error, stackTrace);
       } else {
-        _zone.runUnaryGuarded(_onError, error);
+        _zone.runUnaryGuarded/*<dynamic, dynamic>*/(
+            _onError as Object /*=ZoneUnaryCallback<dynamic, dynamic>*/, error);
       }
       _state &= ~_STATE_IN_CALLBACK;
     }
 
     if (_cancelOnError) {
       _state |= _STATE_WAIT_FOR_CANCEL;
       _cancel();
       if (_cancelFuture is Future) {
         _cancelFuture.whenComplete(sendError);
       } else {
@@ skipping 92 matching lines @@
 // -------------------------------------------------------------------
 abstract class _StreamImpl<T> extends Stream<T> {
   // ------------------------------------------------------------------
   // Stream interface.
 
   StreamSubscription<T> listen(void onData(T data),
                                { Function onError,
                                  void onDone(),
                                  bool cancelOnError }) {
     cancelOnError = identical(true, cancelOnError);
-    StreamSubscription subscription =
+    StreamSubscription<T> subscription =
         _createSubscription(onData, onError, onDone, cancelOnError);
     _onListen(subscription);
     return subscription;
   }
 
   // -------------------------------------------------------------------
   /** Create a subscription object. Called by [subcribe]. */
   StreamSubscription<T> _createSubscription(
       void onData(T data),
       Function onError,
       void onDone(),
       bool cancelOnError) {
     return new _BufferingStreamSubscription<T>(onData, onError, onDone,
                                                cancelOnError);
   }
 
   /** Hook called when the subscription has been created. */
   void _onListen(StreamSubscription subscription) {}
 }
 
-typedef _PendingEvents _EventGenerator();
+typedef _PendingEvents<T> _EventGenerator<T>();
 
 /** Stream that generates its own events. */
 class _GeneratedStreamImpl<T> extends _StreamImpl<T> {
-  final _EventGenerator _pending;
+  final _EventGenerator<T> _pending;
   bool _isUsed = false;
   /**
    * Initializes the stream to have only the events provided by a
    * [_PendingEvents].
    *
    * A new [_PendingEvents] must be generated for each listen.
    */
   _GeneratedStreamImpl(this._pending);
 
   StreamSubscription<T> _createSubscription(
       void onData(T data),
       Function onError,
       void onDone(),
       bool cancelOnError) {
     if (_isUsed) throw new StateError("Stream has already been listened to.");
     _isUsed = true;
-    return new _BufferingStreamSubscription(
+    return new _BufferingStreamSubscription<T>(
         onData, onError, onDone, cancelOnError).._setPendingEvents(_pending());
   }
 }
 
 
 /** Pending events object that gets its events from an [Iterable]. */
-class _IterablePendingEvents<T> extends _PendingEvents {
+class _IterablePendingEvents<T> extends _PendingEvents<T> {
   // The iterator providing data for data events.
   // Set to null when iteration has completed.
   Iterator<T> _iterator;
 
   _IterablePendingEvents(Iterable<T> data) : _iterator = data.iterator;
 
   bool get isEmpty => _iterator == null;
 
-  void handleNext(_EventDispatch dispatch) {
+  void handleNext(_EventDispatch<T> dispatch) {
     if (_iterator == null) {
       throw new StateError("No events pending.");
     }
     // Send one event per call to moveNext.
     // If moveNext returns true, send the current element as data.
     // If moveNext returns false, send a done event and clear the _iterator.
     // If moveNext throws an error, send an error and clear the _iterator.
     // After an error, no further events will be sent.
     bool isDone;
     try {
@@ skipping 73 matching lines @@
   }
 
   _DelayedEvent get next => null;
 
   void set next(_DelayedEvent _) {
     throw new StateError("No events after a done.");
   }
 }
 
 /** Superclass for provider of pending events. */
-abstract class _PendingEvents {
+abstract class _PendingEvents<T> {
   // No async event has been scheduled.
   static const int _STATE_UNSCHEDULED = 0;
   // An async event has been scheduled to run a function.
   static const int _STATE_SCHEDULED = 1;
   // An async event has been scheduled, but it will do nothing when it runs.
   // Async events can't be preempted.
   static const int _STATE_CANCELED = 3;
 
   /**
    * State of being scheduled.
@@ skipping 13 matching lines @@
 
   bool get isScheduled => _state == _STATE_SCHEDULED;
   bool get _eventScheduled => _state >= _STATE_SCHEDULED;
 
   /**
    * Schedule an event to run later.
    *
    * If called more than once, it should be called with the same dispatch as
    * argument each time. It may reuse an earlier argument in some cases.
    */
-  void schedule(_EventDispatch dispatch) {
+  void schedule(_EventDispatch<T> dispatch) {
     if (isScheduled) return;
     assert(!isEmpty);
     if (_eventScheduled) {
       assert(_state == _STATE_CANCELED);
       _state = _STATE_SCHEDULED;
       return;
     }
     scheduleMicrotask(() {
       int oldState = _state;
       _state = _STATE_UNSCHEDULED;
       if (oldState == _STATE_CANCELED) return;
       handleNext(dispatch);
     });
     _state = _STATE_SCHEDULED;
   }
 
   void cancelSchedule() {
     if (isScheduled) _state = _STATE_CANCELED;
   }
 
-  void handleNext(_EventDispatch dispatch);
+  void handleNext(_EventDispatch<T> dispatch);
 
   /** Throw away any pending events and cancel scheduled events. */
   void clear();
 }
 
 
 /** Class holding pending events for a [_StreamImpl]. */
-class _StreamImplEvents extends _PendingEvents {
+class _StreamImplEvents<T> extends _PendingEvents<T> {
   /// Single linked list of [_DelayedEvent] objects.
   _DelayedEvent firstPendingEvent = null;
   /// Last element in the list of pending events. New events are added after it.
   _DelayedEvent lastPendingEvent = null;
 
   bool get isEmpty => lastPendingEvent == null;
 
   void add(_DelayedEvent event) {
     if (lastPendingEvent == null) {
       firstPendingEvent = lastPendingEvent = event;
     } else {
       lastPendingEvent = lastPendingEvent.next = event;
     }
   }
 
-  void handleNext(_EventDispatch dispatch) {
+  void handleNext(_EventDispatch<T> dispatch) {
     assert(!isScheduled);
     _DelayedEvent event = firstPendingEvent;
     firstPendingEvent = event.next;
     if (firstPendingEvent == null) {
       lastPendingEvent = null;
     }
     event.perform(dispatch);
   }
 
   void clear() {
@@ skipping 14 matching lines @@
 
   void _insertBefore(_BroadcastLinkedList newNext) {
     _BroadcastLinkedList newPrevious = newNext._previous;
     newPrevious._next = this;
     newNext._previous = _previous;
     _previous._next = newNext;
     _previous = newPrevious;
   }
 }
 
-typedef void _broadcastCallback(StreamSubscription subscription);
+typedef void _BroadcastCallback<T>(StreamSubscription<T> subscription);
 
 /**
  * Done subscription that will send one done event as soon as possible.
  */
 class _DoneStreamSubscription<T> implements StreamSubscription<T> {
   static const int _DONE_SENT = 1;
   static const int _SCHEDULED = 2;
   static const int _PAUSED = 4;
 
   final Zone _zone;
@@ skipping 27 matching lines @@
     if (isPaused) {
       _state -= _PAUSED;
       if (!isPaused && !_isSent) {
         _schedule();
       }
     }
   }
 
   Future cancel() => null;
 
-  Future asFuture([futureValue]) {
-    _Future result = new _Future();
+  Future/*<E>*/ asFuture/*<E>*/([var/*=E*/ futureValue]) {
+    _Future/*<E>*/ result = new _Future/*<E>*/();
     _onDone = () { result._completeWithValue(null); };
     return result;
   }
 
   void _sendDone() {
     _state &= ~_SCHEDULED;
     if (isPaused) return;
     _state |= _DONE_SENT;
     if (_onDone != null) _zone.runGuarded(_onDone);
   }
 }
 
 class _AsBroadcastStream<T> extends Stream<T> {
   final Stream<T> _source;
-  final _broadcastCallback _onListenHandler;
-  final _broadcastCallback _onCancelHandler;
+  final _BroadcastCallback<T> _onListenHandler;
+  final _BroadcastCallback<T> _onCancelHandler;
   final Zone _zone;
 
   _AsBroadcastStreamController<T> _controller;
   StreamSubscription<T> _subscription;
 
   _AsBroadcastStream(this._source,
-                     void onListenHandler(StreamSubscription subscription),
-                     void onCancelHandler(StreamSubscription subscription))
-      : _onListenHandler = Zone.current.registerUnaryCallback(onListenHandler),
-        _onCancelHandler = Zone.current.registerUnaryCallback(onCancelHandler),
+                     void onListenHandler(StreamSubscription<T> subscription),
+                     void onCancelHandler(StreamSubscription<T> subscription))
+      // TODO(floitsch): the return type should be void and should be
+      // inferred.
+      : _onListenHandler = Zone.current.registerUnaryCallback
+            /*<dynamic, StreamSubscription<T>>*/(onListenHandler),
+        _onCancelHandler = Zone.current.registerUnaryCallback
+            /*<dynamic, StreamSubscription<T>>*/(onCancelHandler),
         _zone = Zone.current {
     _controller = new _AsBroadcastStreamController<T>(_onListen, _onCancel);
   }
 
   bool get isBroadcast => true;
 
   StreamSubscription<T> listen(void onData(T data),
                                { Function onError,
                                  void onDone(),
                                  bool cancelOnError}) {
     if (_controller == null || _controller.isClosed) {
       // Return a dummy subscription backed by nothing, since
       // it will only ever send one done event.
       return new _DoneStreamSubscription<T>(onDone);
     }
     if (_subscription == null) {
       _subscription = _source.listen(_controller.add,
                                      onError: _controller.addError,
                                      onDone: _controller.close);
     }
     cancelOnError = identical(true, cancelOnError);
     return _controller._subscribe(onData, onError, onDone, cancelOnError);
   }
 
   void _onCancel() {
     bool shutdown = (_controller == null) || _controller.isClosed;
     if (_onCancelHandler != null) {
-      _zone.runUnary(_onCancelHandler, new _BroadcastSubscriptionWrapper(this));
+      _zone.runUnary(
+          _onCancelHandler, new _BroadcastSubscriptionWrapper<T>(this));
     }
     if (shutdown) {
       if (_subscription != null) {
         _subscription.cancel();
         _subscription = null;
       }
     }
   }
 
   void _onListen() {
     if (_onListenHandler != null) {
-      _zone.runUnary(_onListenHandler, new _BroadcastSubscriptionWrapper(this));
+      _zone.runUnary(
+          _onListenHandler, new _BroadcastSubscriptionWrapper<T>(this));
     }
   }
 
   // Methods called from _BroadcastSubscriptionWrapper.
   void _cancelSubscription() {
     if (_subscription == null) return;
     // Called by [_controller] when it has no subscribers left.
     StreamSubscription subscription = _subscription;
     _subscription = null;
     _controller = null;  // Marks the stream as no longer listenable.
@@ skipping 49 matching lines @@
 
   Future cancel() {
     _stream._cancelSubscription();
     return null;
   }
 
   bool get isPaused {
     return _stream._isSubscriptionPaused;
   }
 
-  Future asFuture([var futureValue]) {
+  Future/*<E>*/ asFuture/*<E>*/([var/*=E*/ futureValue]) {
     throw new UnsupportedError(
         "Cannot change handlers of asBroadcastStream source subscription.");
   }
 }
 
 
 /**
  * Simple implementation of [StreamIterator].
  */
 class _StreamIteratorImpl<T> implements StreamIterator<T> {
@@ skipping 30 matching lines @@
   /// The current element represented by the most recent call to moveNext.
   ///
   /// Is null between the time moveNext is called and its future completes.
   T _current = null;
 
   /// The future returned by the most recent call to [moveNext].
   ///
   /// Also used to store the next value/error in case the stream provides an
   /// event before [moveNext] is called again. In that case, the stream will
   /// be paused to prevent further events.
-  var _futureOrPrefetch = null;
+  var/*Future<bool> or T*/ _futureOrPrefetch = null;
 
   /// The current state.
   int _state = _STATE_FOUND;
 
   _StreamIteratorImpl(final Stream<T> stream) {
     _subscription = stream.listen(_onData,
                                   onError: _onError,
                                   onDone: _onDone,
                                   cancelOnError: true);
   }
 
   T get current => _current;
 
   Future<bool> moveNext() {
     if (_state == _STATE_DONE) {
       return new _Future<bool>.immediate(false);
     }
     if (_state == _STATE_MOVING) {
       throw new StateError("Already waiting for next.");
     }
     if (_state == _STATE_FOUND) {
       _state = _STATE_MOVING;
       _current = null;
-      _futureOrPrefetch = new _Future<bool>();
-      return _futureOrPrefetch;
+      var result = new _Future<bool>();
+      _futureOrPrefetch = result;
+      return result;
     } else {
       assert(_state >= _STATE_EXTRA_DATA);
       switch (_state) {
         case _STATE_EXTRA_DATA:
           _state = _STATE_FOUND;
-          _current = _futureOrPrefetch;
+          _current = _futureOrPrefetch as Object /*=T*/;
           _futureOrPrefetch = null;
           _subscription.resume();
           return new _Future<bool>.immediate(true);
         case _STATE_EXTRA_ERROR:
           AsyncError prefetch = _futureOrPrefetch;
           _clear();
           return new _Future<bool>.immediateError(prefetch.error,
                                                   prefetch.stackTrace);
         case _STATE_EXTRA_DONE:
           _clear();
           return new _Future<bool>.immediate(false);
       }
     }
   }
 
   /** Clears up the internal state when the iterator ends. */
   void _clear() {
     _subscription = null;
     _futureOrPrefetch = null;
     _current = null;
     _state = _STATE_DONE;
   }
 
   Future cancel() {
     StreamSubscription subscription = _subscription;
-    // Cherry pick of: https://codereview.chromium.org//896793002
     if (subscription == null) return null;
     if (_state == _STATE_MOVING) {
-      _Future<bool> hasNext = _futureOrPrefetch;
+      _Future<bool> hasNext = _futureOrPrefetch as Object /*=_Future<bool>*/;
       _clear();
       hasNext._complete(false);
     } else {
       _clear();
     }
     return subscription.cancel();
   }
 
   void _onData(T data) {
     if (_state == _STATE_MOVING) {
       _current = data;
-      _Future<bool> hasNext = _futureOrPrefetch;
+      _Future<bool> hasNext = _futureOrPrefetch as Object /*=_Future<bool>*/;
       _futureOrPrefetch = null;
       _state = _STATE_FOUND;
       hasNext._complete(true);
       return;
     }
     _subscription.pause();
     assert(_futureOrPrefetch == null);
     _futureOrPrefetch = data;
     _state = _STATE_EXTRA_DATA;
   }
 
   void _onError(Object error, [StackTrace stackTrace]) {
     if (_state == _STATE_MOVING) {
-      _Future<bool> hasNext = _futureOrPrefetch;
+      _Future<bool> hasNext = _futureOrPrefetch as Object /*=_Future<bool>*/;
       // We have cancelOnError: true, so the subscription is canceled.
       _clear();
       hasNext._completeError(error, stackTrace);
       return;
     }
     _subscription.pause();
     assert(_futureOrPrefetch == null);
     _futureOrPrefetch = new AsyncError(error, stackTrace);
     _state = _STATE_EXTRA_ERROR;
   }
 
   void _onDone() {
      if (_state == _STATE_MOVING) {
-      _Future<bool> hasNext = _futureOrPrefetch;
+      _Future<bool> hasNext = _futureOrPrefetch as Object /*=_Future<bool>*/;
       _clear();
       hasNext._complete(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }
+
+/** An empty broadcast stream, sending a done event as soon as possible. */
+class _EmptyStream<T> extends Stream<T> {
+  const _EmptyStream() : super._internal();
+  bool get isBroadcast => true;
+  StreamSubscription<T> listen(void onData(T data),
+                               {Function onError,
+                                void onDone(),
+                                bool cancelOnError}) {
+    return new _DoneStreamSubscription<T>(onDone);
+  }
+}