Rename unsubscribeOnError to cancelOnError.

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 156 matching lines @@
    * Adds a subscription to this stream.
    *
    * On each data event from this stream, the subscriber's [onData] handler
    * is called. If [onData] is null, nothing happens.
    *
    * On errors from this stream, the [onError] handler is given a
    * [AsyncError] object describing the error.
    *
    * If this stream closes, the [onDone] handler is called.
    *
-   * If [unsubscribeOnError] is true, the subscription is ended when
+   * If [cancelOnError] is true, the subscription is ended when
    * the first error is reported. The default is false.
    */
   StreamSubscription<T> listen(void onData(T event),
                                { void onError(AsyncError error),
                                  void onDone(),
-                                 bool unsubscribeOnError});
+                                 bool cancelOnError});
 
   /**
    * Creates a new stream from this stream that discards some data events.
    *
    * The new stream sends the same error and done events as this stream,
    * but it only sends the data events that satisfy the [test].
    */
   Stream<T> where(bool test(T event)) {
     return new _WhereStream<T>(this, test);
   }
@@ skipping 77 matching lines @@
         }
       },
       onError: result._setError,
       onDone: () {
         if (!seenFirst) {
           result._setError(new AsyncError(new StateError("No elements")));
         } else {
           result._setValue(value);
         }
       },
-      unsubscribeOnError: true
+      cancelOnError: true
     );
     return result;
   }
 
   /** Reduces a sequence of values by repeatedly applying [combine]. */
   Future fold(var initialValue, combine(var previous, T element)) {
     _FutureImpl result = new _FutureImpl();
     var value = initialValue;
     StreamSubscription subscription;
     subscription = this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ element) {
         _runUserCode(
           () => combine(value, element),
           (newValue) { value = newValue; },
           _cancelAndError(subscription, result)
         );
       },
       onError: (AsyncError e) {
         result._setError(e);
       },
       onDone: () {
         result._setValue(value);
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return result;
   }
 
   /**
    * Checks whether [match] occurs in the elements provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> contains(T match) {
@@ skipping 11 matching lines @@
                 future._setValue(true);
               }
             },
             _cancelAndError(subscription, future)
           );
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
-        unsubscribeOnError: true);
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Checks whether [test] accepts all elements provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> every(bool test(T element)) {
@@ skipping 11 matching lines @@
                 future._setValue(false);
               }
             },
             _cancelAndError(subscription, future)
           );
         },
         onError: future._setError,
         onDone: () {
           future._setValue(true);
         },
-        unsubscribeOnError: true);
+        cancelOnError: true);
     return future;
   }
 
   /**
    * Checks whether [test] accepts any element provided by this stream.
    *
    * Completes the [Future] when the answer is known.
    * If this stream reports an error, the [Future] will report that error.
    */
   Future<bool> any(bool test(T element)) {
@@ skipping 11 matching lines @@
                 future._setValue(true);
               }
             },
             _cancelAndError(subscription, future)
           );
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
-        unsubscribeOnError: true);
+        cancelOnError: true);
     return future;
   }
 
 
   /** Counts the elements in the stream. */
   Future<int> get length {
     _FutureImpl<int> future = new _FutureImpl<int>();
     int count = 0;
     this.listen(
       (_) { count++; },
       onError: future._setError,
       onDone: () {
         future._setValue(count);
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 
   /** Reports whether this stream contains any elements. */
   Future<bool> get isEmpty {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
       (_) {
         subscription.cancel();
         future._setValue(false);
       },
       onError: future._setError,
       onDone: () {
         future._setValue(true);
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 
   /** Collects the data of this stream in a [List]. */
   Future<List<T>> toList() {
     List<T> result = <T>[];
     _FutureImpl<List<T>> future = new _FutureImpl<List<T>>();
     this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ data) {
         result.add(data);
       },
       onError: future._setError,
       onDone: () {
         future._setValue(result);
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 
   /** Collects the data of this stream in a [Set]. */
   Future<Set<T>> toSet() {
     Set<T> result = new Set<T>();
     _FutureImpl<Set<T>> future = new _FutureImpl<Set<T>>();
     this.listen(
       // TODO(ahe): Restore type when feature is implemented in dart2js
       // checked mode. http://dartbug.com/7733
       (/*T*/ data) {
         result.add(data);
       },
       onError: future._setError,
       onDone: () {
         future._setValue(result);
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 
   /**
    * Provides at most the first [n] values of this stream.
    *
    * Forwards the first [n] data events of this stream, and all error
    * events, to the returned stream, and ends with a done event.
    *
    * If this stream produces fewer than [count] values before it's done,
@@ skipping 61 matching lines @@
       // checked mode. http://dartbug.com/7733
       (/*T*/ value) {
         subscription.cancel();
         future._setValue(value);
         return;
       },
       onError: future._setError,
       onDone: () {
         future._setError(new AsyncError(new StateError("No elements")));
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 
   /**
    * Returns the last element.
    *
    * If [this] is empty throws a [StateError].
    */
   Future<T> get last {
     _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) {
         foundResult = true;
         result = value;
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
         }
         future._setError(new AsyncError(new StateError("No elements")));
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 
   /**
    * Returns the single element.
    *
    * If [this] is empty or has more than one element throws a [StateError].
    */
   Future<T> get single {
     _FutureImpl<T> future = new _FutureImpl<T>();
@@ skipping 15 matching lines @@
         result = value;
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
         }
         future._setError(new AsyncError(new StateError("No elements")));
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the first element of this stream matching [test].
    *
    * Returns a future that is filled with the first element of this stream
    * that [test] returns true for.
    *
    * If no such element is found before this stream is done, and a
@@ skipping 24 matching lines @@
       },
       onError: future._setError,
       onDone: () {
         if (defaultValue != null) {
           _runUserCode(defaultValue, future._setValue, future._setError);
           return;
         }
         future._setError(
             new AsyncError(new StateError("firstMatch ended without match")));
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the last element in this stream matching [test].
    *
    * As [firstWhere], except that the last matching element is found.
    * That means that the result cannot be provided before this stream
    * is done.
    */
@@ skipping 23 matching lines @@
           future._setValue(result);
           return;
         }
         if (defaultValue != null) {
           _runUserCode(defaultValue, future._setValue, future._setError);
           return;
         }
         future._setError(
             new AsyncError(new StateError("lastMatch ended without match")));
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 
   /**
    * Finds the single element in this stream matching [test].
    *
    * Like [lastMatch], except that it is an error if more than one
    * matching element occurs in the stream.
    */
   Future<T> singleWhere(bool test(T value)) {
@@ skipping 24 matching lines @@
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
         }
         future._setError(
             new AsyncError(new StateError("single ended without match")));
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 
   /**
    * Returns the value of the [index]th data event of this stream.
    *
    * If an error event occurs, the future will end with this error.
    *
    * If this stream provides fewer than [index] elements before closing,
    * an error is reported.
@@ skipping 11 matching lines @@
           future._setValue(value);
           return;
         }
         index -= 1;
       },
       onError: future._setError,
       onDone: () {
         future._setError(new AsyncError(
             new StateError("Not enough elements for elementAt")));
       },
-      unsubscribeOnError: true);
+      cancelOnError: true);
     return future;
   }
 }
 
 /**
  * A control object for the subscription on a [Stream].
  *
  * When you subscribe on a [Stream] using [Stream.listen],
  * a [StreamSubscription] object is returned. This object
  * is used to later unsubscribe again, or to temporarily pause
@@ skipping 36 matching lines @@
    */
   void resume();
 
   /**
    * Returns a future that handles the [onDone] and [onError] callbacks.
    *
    * This method *overwrites* the existing [onDone] and [onError] callbacks
    * with new ones that complete the returned future.
    *
    * In case of an error the subscription will automatically cancel (even
-   * when it was listening with `unsubscribeOnError` set to `false`).
+   * when it was listening with `cancelOnError` set to `false`).
    *
    * In case of a `done` event the future completes with the given
    * [futureValue].
    */
   Future asFuture([var futureValue]);
 }
 
 
 /**
  * An interface that abstracts creation or handling of [Stream] events.
@@ skipping 14 matching lines @@
 
   StreamView(this._stream);
 
   bool get isBroadcast => _stream.isBroadcast;
 
   Stream<T> asBroadcastStream() => _stream.asBroadcastStream();
 
   StreamSubscription<T> listen(void onData(T value),
                                { void onError(AsyncError error),
                                  void onDone(),
-                                 bool unsubscribeOnError }) {
+                                 bool cancelOnError }) {
     return _stream.listen(onData, onError: onError, onDone: onDone,
-                          unsubscribeOnError: unsubscribeOnError);
+                          cancelOnError: cancelOnError);
   }
 }
 
 /**
  * [EventSink] wrapper that only exposes the [EventSink] interface.
  */
 class _EventSinkView<T> extends EventSink<T> {
   final EventSink<T> _sink;
 
   _EventSinkView(this._sink);
@@ skipping 189 matching lines @@
 class EventTransformStream<S, T> extends Stream<T> {
   final Stream<S> _source;
   final StreamEventTransformer _transformer;
   EventTransformStream(Stream<S> source,
                        StreamEventTransformer<S, T> transformer)
       : _source = source, _transformer = transformer;
 
   StreamSubscription<T> listen(void onData(T data),
                                { void onError(AsyncError error),
                                  void onDone(),
-                                 bool unsubscribeOnError }) {
-    unsubscribeOnError = identical(true, unsubscribeOnError);
+                                 bool cancelOnError }) {
+    cancelOnError = identical(true, cancelOnError);
     return new _EventTransformStreamSubscription(_source, _transformer,
                                                  onData, onError, onDone,
-                                                 unsubscribeOnError);
+                                                 cancelOnError);
   }
 }
 
 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. */

[Lasse Reichstein Nielsen, 2013/04/15 16:11:03]

unsubscribe -> cancel :)

[floitsch, 2013/04/15 16:13:52]

Please file a bug. I will do it later. ;P

-  final bool _unsubscribeOnError;
+  final bool _cancelOnError;
   /** 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(),
-                                    this._unsubscribeOnError)
+                                    this._cancelOnError)
       : 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;
 
@@ skipping 41 matching lines @@
 
   // EventOutputSink interface.
   void _sendData(T data) {
     if (_isClosed) return;
     _onData(data);
   }
 
   void _sendError(AsyncError error) {
     if (_isClosed) return;
     _onError(error);
-    if (_unsubscribeOnError) {
+    if (_cancelOnError) {
       cancel();
     }
   }
 
   void _sendDone() {
     if (_isClosed) throw new StateError("Already closed.");
     _isClosed = true;
     if (_isSubscribed) {
       _subscription.cancel();
       _subscription = null;
     }
     _onDone();
   }
 }
 
 class _EventOutputSinkWrapper<T> extends EventSink<T> {
   _EventOutputSink _sink;
   _EventOutputSinkWrapper(this._sink);
 
   void add(T data) { _sink._sendData(data); }
   void addError(AsyncError error) { _sink._sendError(error); }
   void close() { _sink._sendDone(); }
 }