Revert "Rewrite Futures."

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);
@@ skipping 172 matching lines @@
     if (!_inCallback) {
       // otherwise checkState will be called after firing or callback completes.
       _state |= _STATE_IN_CALLBACK;
       _onCancel();
       _pending = null;
       _state &= ~_STATE_IN_CALLBACK;
     }
   }
 
   Future asFuture([var futureValue]) {
-    _Future<T> result = new _Future<T>();
+    _FutureImpl<T> result = new _FutureImpl<T>();
 
     // Overwrite the onDone and onError handlers.
-    _onDone = () { result._complete(futureValue); };
+    _onDone = () { result._setValue(futureValue); };
     _onError = (error) {
       cancel();
-      result._completeError(error);
+      result._setError(error);
     };
 
     return result;
   }
 
   // State management.
 
   bool get _isInputPaused => (_state & _STATE_INPUT_PAUSED) != 0;
   bool get _isClosed => (_state & _STATE_CLOSED) != 0;
   bool get _isCanceled => (_state & _STATE_CANCELED) != 0;
@@ skipping 499 matching lines @@
   void pause([Future resumeSignal]) {
     _pauseCounter++;
     if (resumeSignal != null) resumeSignal.then((_) { resume(); });
   }
   void resume() {
     if (_pauseCounter > 0) _pauseCounter--;
   }
   void cancel() {}
   bool get isPaused => _pauseCounter > 0;
 
-  Future asFuture([futureValue]) => new _Future();
+  Future asFuture([futureValue]) => new _FutureImpl();
 }
 
 class _AsBroadcastStream<T> extends Stream<T> {
   final Stream<T> _source;
   final _broadcastCallback _onListenHandler;
   final _broadcastCallback _onCancelHandler;
   final _Zone _zone;
 
   _AsBroadcastStreamController<T> _controller;
   StreamSubscription<T> _subscription;
@@ skipping 180 matching lines @@
     _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);
+      return new _FutureImpl<bool>.immediate(false);
     }
     if (_state == _STATE_MOVING) {
       throw new StateError("Already waiting for next.");
     }
     if (_state == _STATE_FOUND) {
       _state = _STATE_MOVING;
-      _futureOrPrefetch = new _Future<bool>();
+      _futureOrPrefetch = new _FutureImpl<bool>();
       return _futureOrPrefetch;
     } else {
       assert(_state >= _STATE_EXTRA_DATA);
       switch (_state) {
         case _STATE_EXTRA_DATA:
           _state = _STATE_FOUND;
           _current = _futureOrPrefetch;
           _futureOrPrefetch = null;
           _subscription.resume();
-          return new _Future<bool>.immediate(true);
+          return new _FutureImpl<bool>.immediate(true);
         case _STATE_EXTRA_ERROR:
           Object prefetch = _futureOrPrefetch;
           _clear();
-          return new _Future<bool>.immediateError(prefetch);
+          return new _FutureImpl<bool>.immediateError(prefetch);
         case _STATE_EXTRA_DONE:
           _clear();
-          return new _Future<bool>.immediate(false);
+          return new _FutureImpl<bool>.immediate(false);
       }
     }
   }
 
   /** Clears up the internal state when the iterator ends. */
   void _clear() {
     _subscription = null;
     _futureOrPrefetch = null;
     _current = null;
     _state = _STATE_DONE;
   }
 
   void cancel() {
     StreamSubscription subscription = _subscription;
     if (_state == _STATE_MOVING) {
-      _Future<bool> hasNext = _futureOrPrefetch;
+      _FutureImpl<bool> hasNext = _futureOrPrefetch;
       _clear();
-      hasNext._complete(false);
+      hasNext._setValue(false);
     } else {
       _clear();
     }
     subscription.cancel();
   }
 
   void _onData(T data) {
     if (_state == _STATE_MOVING) {
       _current = data;
-      _Future<bool> hasNext = _futureOrPrefetch;
+      _FutureImpl<bool> hasNext = _futureOrPrefetch;
       _futureOrPrefetch = null;
       _state = _STATE_FOUND;
-      hasNext._complete(true);
+      hasNext._setValue(true);
       return;
     }
     _subscription.pause();
     assert(_futureOrPrefetch == null);
     _futureOrPrefetch = data;
     _state = _STATE_EXTRA_DATA;
   }
 
   void _onError(Object error) {
     if (_state == _STATE_MOVING) {
-      _Future<bool> hasNext = _futureOrPrefetch;
+      _FutureImpl<bool> hasNext = _futureOrPrefetch;
       // We have cancelOnError: true, so the subscription is canceled.
       _clear();
-      hasNext._completeError(error);
+      hasNext._setError(error);
       return;
     }
     _subscription.pause();
     assert(_futureOrPrefetch == null);
     _futureOrPrefetch = error;
     _state = _STATE_EXTRA_ERROR;
   }
 
   void _onDone() {
      if (_state == _STATE_MOVING) {
-      _Future<bool> hasNext = _futureOrPrefetch;
+      _FutureImpl<bool> hasNext = _futureOrPrefetch;
       _clear();
-      hasNext._complete(false);
+      hasNext._setValue(false);
       return;
     }
     _subscription.pause();
     _futureOrPrefetch = null;
     _state = _STATE_EXTRA_DONE;
   }
 }