Fix stream_controller_test failure.

sdk/lib/async/stream.dart

 // Copyright (c) 2013, 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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 1117 matching lines @@
   }
 }
 
 class _EventTransformStreamSubscription<S, T>
     extends _BaseStreamSubscription<T>
     implements _EventOutputSink<T> {
   /** The transformer used to transform events. */
   final StreamEventTransformer<S, T> _transformer;
   /** Whether to unsubscribe when emitting an error. */
   final bool _unsubscribeOnError;
+  /** Whether this stream has sent a done event. */
+  bool _isClosed = false;
   /** Source of incoming events. */
   StreamSubscription<S> _subscription;
   /** Cached EventSink wrapper for this class. */
   EventSink<T> _sink;
 
   _EventTransformStreamSubscription(Stream<S> source,
                                     this._transformer,
                                     void onData(T data),
                                     void onError(AsyncError error),
                                     void onDone(),
@@ skipping 11 matching lines @@
   void pause([Future pauseSignal]) {
     if (_isSubscribed) _subscription.pause(pauseSignal);
   }
 
   void resume() {
     if (_isSubscribed) _subscription.resume();
   }
 
   void cancel() {
     if (_isSubscribed) {
-      _subscription.cancel();
+      StreamSubscription subscription = _subscription;
       _subscription = null;
+      subscription.cancel();
     }
+    _isClosed = true;
   }
 
   void _handleData(S data) {
     try {
       _transformer.handleData(data, _sink);
     } catch (e, s) {
       _sendError(_asyncError(e, s));
     }
   }
 
   void _handleError(AsyncError error) {
     try {
       _transformer.handleError(error, _sink);
     } catch (e, s) {
       _sendError(_asyncError(e, s, error));
     }
   }
 
   void _handleDone() {
     try {
+      _subscription = null;
       _transformer.handleDone(_sink);
     } catch (e, s) {
       _sendError(_asyncError(e, s));
     }
   }
 
   // EventOutputSink interface.
   void _sendData(T data) {
-    if (!_isSubscribed) return;
+    if (_isClosed) return;
     _onData(data);
   }
 
   void _sendError(AsyncError error) {
-    if (!_isSubscribed) return;
+    if (_isClosed) return;
     _onError(error);
     if (_unsubscribeOnError) {
       cancel();
     }
   }
 
   void _sendDone() {
-    if (!_isSubscribed) return;
-    _subscription.cancel();
-    _subscription = null;
+    if (_isClosed) throw new StateError("Already closed.");
+    _isClosed = true;
+    if (_isSubscribed) {
+      _subscription.cancel();
+      _subscription = null;
+    }
     _onDone();
   }
 }
 
 /* TODO(8997): Implement EventSink instead, */
 class _EventOutputSinkWrapper<T> extends StreamSink<T> {
   _EventOutputSink _sink;
   _EventOutputSinkWrapper(this._sink);
 
   void add(T data) { _sink._sendData(data); }
   void addError(AsyncError error) { _sink._sendError(error); }
   void close() { _sink._sendDone(); }
 }