Make StreamController be a StreamSink, not just an EventSink.

sdk/lib/async/future_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;
 
 deprecatedFutureValue(_FutureImpl future) =>
   future._isComplete ? future._resultOrListeners : null;
 
 abstract class _Completer<T> implements Completer<T> {
@@ skipping 31 matching lines @@
       _Zone.current.handleUncaughtError(error);
     }
   }
 
   bool get isCompleted => _isComplete;
 }
 
 class _AsyncCompleter<T> extends _Completer<T> {
   void _setFutureValue(T value) {
     _FutureImpl future = this.future;
-    runAsync(() { future._setValue(value); });
+    future._asyncSetValue(value);
   }
 
   void _setFutureError(error) {
     _FutureImpl future = this.future;
-    runAsync(() { future._setError(error); });
+    future._asyncSetError(error);
   }
 }
 
 class _SyncCompleter<T> extends _Completer<T> {
   void _setFutureValue(T value) {
     _FutureImpl future = this.future;
     future._setValue(value);
   }
 
   void _setFutureError(error) {
@@ skipping 19 matching lines @@
   void _sendError(error);
 
   bool _inSameErrorZone(_Zone otherZone);
 }
 
 /** Adapter for a [_FutureImpl] to be a future result listener. */
 class _FutureListenerWrapper<T> implements _FutureListener<T> {
   _FutureImpl future;
   _FutureListener _nextListener;
   _FutureListenerWrapper(this.future);
-  _sendValue(T value) { future._setValue(value); }
-  _sendError(error) { future._setError(error); }
+  _sendValue(T value) { future._setValueUnchecked(value); }
+  _sendError(error) { future._setErrorUnchecked(error); }
   bool _inSameErrorZone(_Zone otherZone) => future._inSameErrorZone(otherZone);
 }
 
 /**
  * This listener is installed at error-zone boundaries. It signals an
  * uncaught error in the zone of origin when an error is sent from one error
  * zone to another.
  *
  * When a Future is listening to another Future and they have not been
  * instantiated in the same error-zone then Futures put an instance of this
@@ skipping 46 matching lines @@
 
 class _FutureImpl<T> implements Future<T> {
   // State of the future. The state determines the interpretation of the
   // [resultOrListeners] field.
   // TODO(lrn): rename field since it can also contain a chained future.
 
   /// Initial state, waiting for a result. In this state, the
   /// [resultOrListeners] field holds a single-linked list of
   /// [FutureListener] listeners.
   static const int _INCOMPLETE = 0;
+  /// Pending completion. Set when completed using [_asyncSetValue] or
+  /// [_asyncSetError]. It is an error to try to complete it again.
+  static const int _PENDING_COMPLETE = 1;
   /// The future has been chained to another future. The result of that
   /// other future becomes the result of this future as well.
   /// In this state, the [resultOrListeners] field holds the future that
   /// will give the result to this future. Both existing and new listeners are
   /// forwarded directly to the other future.
-  static const int _CHAINED = 1;
+  static const int _CHAINED = 2;
   /// The future has been chained to another future, but there hasn't been
   /// any listeners added to this future yet. If it is completed with an
   /// error, the error will be considered unhandled.
-  static const int _CHAINED_UNLISTENED = 3;
+  static const int _CHAINED_UNLISTENED = 6;
   /// The future has been completed with a value result.
-  static const int _VALUE = 4;
+  static const int _VALUE = 8;
   /// The future has been completed with an error result.
-  static const int _ERROR = 6;
+  static const int _ERROR = 12;
   /// Extra bit set when the future has been completed with an error result.
   /// but no listener has been scheduled to receive the error.
   /// If the bit is still set when a [runAsync] call triggers, the error will
   /// be reported to the top-level handler.
   /// Assigning a listener before that time will clear the bit.
-  static const int _UNHANDLED_ERROR = 8;
+  static const int _UNHANDLED_ERROR = 16;
 
   /** Whether the future is complete, and as what. */
   int _state = _INCOMPLETE;
 
   final _Zone _zone = _Zone.current.fork();
 
   bool get _isChained => (_state & _CHAINED) != 0;
   bool get _hasChainedListener => _state == _CHAINED;
   bool get _isComplete => _state >= _VALUE;
+  bool get _mayComplete => _state == _INCOMPLETE;
   bool get _hasValue => _state == _VALUE;
   bool get _hasError => _state >= _ERROR;
   bool get _hasUnhandledError => _state >= _UNHANDLED_ERROR;
 
   void _clearUnhandledError() {
     _state &= ~_UNHANDLED_ERROR;
   }
 
   /**
    * Either the result, a list of listeners or another future.
@@ skipping 79 matching lines @@
     return new _WhenFuture<T>(action).._subscribeTo(this);
   }
 
   Stream<T> asStream() => new Stream.fromFuture(this);
 
   bool _inSameErrorZone(_Zone otherZone) {
     return _zone.inSameErrorZone(otherZone);
   }
 
   void _setValue(T value) {
-    if (_isComplete) throw new StateError("Future already completed");
+    if (!_mayComplete) throw new StateError("Future already completed");
+    _setValueUnchecked(value);
+  }
+
+  void _setValueUnchecked(T value) {
     _FutureListener listeners = _isChained ? null : _removeListeners();
     _state = _VALUE;
     _resultOrListeners = value;
     while (listeners != null) {
       _FutureListener listener = listeners;
       listeners = listener._nextListener;
       listener._nextListener = null;
       listener._sendValue(value);
     }
   }
 
-  void _setError(error) {
-    if (_isComplete) throw new StateError("Future already completed");
+  void _setError(Object error) {
+    if (!_mayComplete) throw new StateError("Future already completed");
+    _setErrorUnchecked(error);
+  }
 
+  void _setErrorUnchecked(error) {

[floitsch, 2013/06/27 15:15:19]

If you add the "Object" above, also add it here.

[Lasse Reichstein Nielsen, 2013/06/28 12:57:38]

Done.

     _FutureListener listeners;
     bool hasListeners;
     if (_isChained) {
       listeners = null;
       hasListeners = (_state == _CHAINED);  // and not _CHAINED_UNLISTENED.
     } else {
       listeners = _removeListeners();
       hasListeners = (listeners != null);
     }
 
     _state = _ERROR;
     _resultOrListeners = error;
 
     if (!hasListeners) {
       _scheduleUnhandledError();
       return;
     }
     while (listeners != null) {
       _FutureListener listener = listeners;
       listeners = listener._nextListener;
       listener._nextListener = null;
       listener._sendError(error);
     }
   }
 
+  void _asyncSetValue(T value) {
+    if (!_mayComplete) throw new StateError("Future already completed");
+    _state = _PENDING_COMPLETE;
+    runAsync(() { _setValueUnchecked(value); });
+  }
+
+  void _asyncSetError(Object error) {
+    if (!_mayComplete) throw new StateError("Future already completed");
+    _state = _PENDING_COMPLETE;
+    runAsync(() { _setErrorUnchecked(error); });
+  }
+
   void _scheduleUnhandledError() {
     assert(_state == _ERROR);
     _state = _ERROR | _UNHANDLED_ERROR;
     // Wait for the rest of the current event's duration to see
     // if a subscriber is added to handle the error.
     runAsync(() {
       if (_hasUnhandledError) {
         // No error handler has been added since the error was set.
         _clearUnhandledError();
         // TODO(floitsch): Hook this into unhandled error handling.
@@ skipping 310 matching lines @@
           _setError(error);
         }, onError: _setError);
         return;
       }
     } catch (e, s) {
       error = _asyncError(e, s);
     }
     _setError(error);
   }
 }