Revert "Rewrite Futures."

sdk/lib/async/stream_controller.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;
 
 // -------------------------------------------------------------------
 // Controller for creating and adding events to a stream.
 // -------------------------------------------------------------------
 
@@ skipping 232 matching lines @@
   int _state = _STATE_INITIAL;
 
   /**
    * Future completed when the stream sends its last event.
    *
    * This is also the future returned by [close].
    */
   // TODO(lrn): Could this be stored in the varData field too, if it's not
   // accessed until the call to "close"? Then we need to special case if it's
   // accessed earlier, or if close is called before subscribing.
-  _Future _doneFuture;
+  _FutureImpl _doneFuture;
 
   _StreamController();
 
   _NotificationHandler get _onListen;
   _NotificationHandler get _onPause;
   _NotificationHandler get _onResume;
   _NotificationHandler get _onCancel;
 
   // Return a new stream every time. The streams are equal, but not identical.
   Stream<T> get stream => new _ControllerStream(this);
@@ skipping 76 matching lines @@
     if (isClosed) {
       return new StateError("Cannot add event after closing");
     }
     assert(_isAddingStream);
     return new StateError("Cannot add event while adding a stream");
   }
 
   // StreamSink interface.
   Future addStream(Stream<T> source) {
     if (!_mayAddEvent) throw _badEventState();
-    if (_isCanceled) return new _Future.immediate(null);
+    if (_isCanceled) return new _FutureImpl.immediate(null);
     _StreamControllerAddStreamState addState =
         new _StreamControllerAddStreamState(this, _varData, source);
     _varData = addState;
     _state |= _STATE_ADDSTREAM;
     return addState.addStreamFuture;
   }
 
   Future get done => _ensureDoneFuture();
 
   Future _ensureDoneFuture() {
     if (_doneFuture == null) {
-      _doneFuture = new _Future();
-      if (_isCanceled) _doneFuture._complete(null);
+      _doneFuture = new _FutureImpl();
+      if (_isCanceled) _doneFuture._setValue(null);
     }
     return _doneFuture;
   }
 
   /**
    * Send or enqueue a data event.
    */
   void add(T value) {
     if (!_mayAddEvent) throw _badEventState();
     _add(value);
@@ skipping 98 matching lines @@
   void _recordCancel(StreamSubscription<T> subscription) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.cancel();
     }
     _varData = null;
     _state =
         (_state & ~(_STATE_SUBSCRIBED | _STATE_ADDSTREAM)) | _STATE_CANCELED;
     _runGuarded(_onCancel);
     if (_doneFuture != null && _doneFuture._mayComplete) {
-      _doneFuture._asyncComplete(null);
+      _doneFuture._asyncSetValue(null);
     }
   }
 
   void _recordPause(StreamSubscription<T> subscription) {
     if (_isAddingStream) {
       _StreamControllerAddStreamState addState = _varData;
       addState.pause();
     }
     _runGuarded(_onPause);
   }
@@ skipping 149 matching lines @@
   Future close() => _target.close();
   Future addStream(Stream<T> source) => _target.addStream(source);
   Future get done => _target.done;
 }
 
 /**
  * Object containing the state used to handle [StreamController.addStream].
  */
 class _AddStreamState<T> {
   // [_FutureImpl] returned by call to addStream.
-  _Future addStreamFuture;
+  _FutureImpl addStreamFuture;
 
   // Subscription on stream argument to addStream.
   StreamSubscription addSubscription;
 
   _AddStreamState(_EventSink<T> controller, Stream source)
-      : addStreamFuture = new _Future(),
+      : addStreamFuture = new _FutureImpl(),
         addSubscription = source.listen(controller._add,
                                         onError: controller._addError,
                                         onDone: controller._close,
                                         cancelOnError: true);
 
   void pause() {
     addSubscription.pause();
   }
 
   void resume() {
     addSubscription.resume();
   }
 
   void cancel() {
     addSubscription.cancel();
     complete();
   }
 
   void complete() {
-    addStreamFuture._asyncComplete(null);
+    addStreamFuture._asyncSetValue(null);
   }
 }
 
 class _StreamControllerAddStreamState<T> extends _AddStreamState<T> {
   // The subscription or pending data of a _StreamController.
   // Stored here because we reuse the `_varData` field  in the _StreamController
   // to store this state object.
   var varData;
 
   _StreamControllerAddStreamState(_StreamController controller,
                                   this.varData,
                                   Stream source) : super(controller, source) {
     if (controller.isPaused) {
       addSubscription.pause();
     }
   }
 }