Rewrite Futures.

sdk/lib/async/broadcast_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;
 
 class _BroadcastStream<T> extends _ControllerStream<T> {
   _BroadcastStream(_StreamControllerLifecycle controller) : super(controller);
 
   bool get isBroadcast => true;
@@ skipping 89 matching lines @@
    * The future is completed whenever the done event has been sent to all
    * relevant listeners.
    * The relevant listeners are the ones that were listening when [close] was
    * called. When all of these have been canceled (sending the done event makes
    * them cancel, but they can also be canceled before sending the event),
    * this future completes.
    *
    * Any attempt to listen after calling [close] will throw, so there won't
    * be any further listeners.
    */
-  _FutureImpl _doneFuture;
+  _Future _doneFuture;
 
   _BroadcastStreamController(this._onListen, this._onCancel)
       : _state = _STATE_INITIAL {
     _next = _previous = this;
   }
 
   // StreamController interface.
 
   Stream<T> get stream => new _BroadcastStream<T>(this);
 
@@ skipping 12 matching lines @@
   /** Whether there are currently one or more subscribers. */
   bool get hasListener => !_isEmpty;
 
   /** Whether an event is being fired (sent to some, but not all, listeners). */
   bool get _isFiring => (_state & _STATE_FIRING) != 0;
 
   bool get _isAddingStream => (_state & _STATE_ADDSTREAM) != 0;
 
   bool get _mayAddEvent => (_state < _STATE_CLOSED);
 
-  _FutureImpl _ensureDoneFuture() {
+  _Future _ensureDoneFuture() {
     if (_doneFuture != null) return _doneFuture;
-    return _doneFuture = new _FutureImpl();
+    return _doneFuture = new _Future();
   }
 
   // Linked list helpers
 
   bool get _isEmpty => identical(_next, this);
 
   /** Adds subscription to linked list of active listeners. */
   void _addListener(_BroadcastSubscription<T> subscription) {
     assert(identical(subscription._next, subscription));
     // Insert in linked list just before `this`.
@@ skipping 151 matching lines @@
 
     if (_isEmpty) {
       _callOnCancel();
     }
   }
 
   void _callOnCancel() {
     assert(_isEmpty);
     if (isClosed && _doneFuture._mayComplete) {
       // When closed, _doneFuture is not null.
-      _doneFuture._asyncSetValue(null);
+      _doneFuture._asyncComplete(null);
     }
     _runGuarded(_onCancel);
   }
 }
 
 class _SyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
   _SyncBroadcastStreamController(void onListen(), void onCancel())
       : super(onListen, onCancel);
 
   // EventDispatch interface.
@@ skipping 13 matching lines @@
   }
 
   void _sendDone() {
     if (!_isEmpty) {
       _forEachListener((_BroadcastSubscription<T> subscription) {
         subscription._close();
       });
     } else {
       assert(_doneFuture != null);
       assert(_doneFuture._mayComplete);
-      _doneFuture._asyncSetValue(null);
+      _doneFuture._asyncComplete(null);
     }
   }
 }
 
 class _AsyncBroadcastStreamController<T> extends _BroadcastStreamController<T> {
   _AsyncBroadcastStreamController(void onListen(), void onCancel())
       : super(onListen, onCancel);
 
   // EventDispatch interface.
 
@@ skipping 19 matching lines @@
     if (!_isEmpty) {
       for (_BroadcastSubscriptionLink link = _next;
            !identical(link, this);
            link = link._next) {
         _BroadcastSubscription<T> subscription = link;
         subscription._addPending(const _DelayedDone());
       }
     } else {
       assert(_doneFuture != null);
       assert(_doneFuture._mayComplete);
-      _doneFuture._asyncSetValue(null);
+      _doneFuture._asyncComplete(null);
     }
   }
 }
 
 /**
  * Stream controller that is used by [Stream.asBroadcastStream].
  *
  * This stream controller allows incoming events while it is firing
  * other events. This is handled by delaying the events until the
  * current event is done firing, and then fire the pending events.
@@ skipping 71 matching lines @@
   void pause([Future resumeSignal]) {
     if (resumeSignal != null) resumeSignal.then(_resume);
     _pauseCount++;
   }
   void resume() { _resume(null); }
   void _resume(_) {
     if (_pauseCount > 0) _pauseCount--;
   }
   void cancel() {}
   bool get isPaused => _pauseCount > 0;
-  Future asFuture([Object value]) => new _FutureImpl();
+  Future asFuture([Object value]) => new _Future();
 }