Make EventTransformerStream's subscription not send events after cancel.

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 599 matching lines @@
    * If [this] is empty throws a [StateError]. Otherwise this method is
    * equivalent to [:this.elementAt(0):]
    */
   Future<T> get first {
     _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     subscription = this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ value) {
+        subscription.cancel();
         future._setValue(value);
-        subscription.cancel();
         return;
       },
       onError: future._setError,
       onDone: () {
         future._setError(new AsyncError(new StateError("No elements")));
       },
       unsubscribeOnError: true);
     return future;
   }
 
@@ skipping 34 matching lines @@
   Future<T> get single {
     _FutureImpl<T> future = new _FutureImpl<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
     subscription = this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ value) {
         if (foundResult) {
+          subscription.cancel();
           // This is the second element we get.
           Error error = new StateError("More than one element");
           future._setError(new AsyncError(error));
-          subscription.cancel();
           return;
         }
         foundResult = true;
         result = value;
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
@@ skipping 148 matching lines @@
    */
   Future<T> elementAt(int index) {
     if (index is! int || index < 0) throw new ArgumentError(index);
     _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     subscription = this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ value) {
         if (index == 0) {
+          subscription.cancel();
           future._setValue(value);
-          subscription.cancel();
           return;
         }
         index -= 1;
       },
       onError: future._setError,
       onDone: () {
         future._setError(new AsyncError(
             new StateError("Not enough elements for elementAt")));
       },
       unsubscribeOnError: true);
@@ skipping 264 matching lines @@
                                     void onError(AsyncError error),
                                     void onDone(),
                                     this._unsubscribeOnError)
       : super(onData, onError, onDone) {
     _sink = new _EventOutputSinkWrapper<T>(this);
     _subscription = source.listen(_handleData,
                                   onError: _handleError,
                                   onDone: _handleDone);
   }
 
+  /** Whether this subscription is still subscribed to its source. */
+  bool get _isSubscribed => _subscription != null;
+
   void pause([Future pauseSignal]) {
-    if (_subscription != null) _subscription.pause(pauseSignal);
+    if (_isSubscribed) _subscription.pause(pauseSignal);
   }
 
   void resume() {
-    if (_subscription != null) _subscription.resume();
+    if (_isSubscribed) _subscription.resume();
   }
 
   void cancel() {
-    if (_subscription != null) {
+    if (_isSubscribed) {
       _subscription.cancel();
       _subscription = null;
     }
   }
 
   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() {
-    _subscription = null;
     try {
       _transformer.handleDone(_sink);
     } catch (e, s) {
       _sendError(_asyncError(e, s));
     }
   }
 
   // EventOutputSink interface.
   void _sendData(T data) {
+    if (!_isSubscribed) return;
     _onData(data);
   }
 
   void _sendError(AsyncError error) {
+    if (!_isSubscribed) return;
     _onError(error);
     if (_unsubscribeOnError) {
       cancel();
     }
   }
 
   void _sendDone() {
-    // It's ok to cancel even if we have been unsubscribed already.
-    cancel();
+    if (!_isSubscribed) return;
+    _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(); }
 }