Rename unsubscribe to cancel.

sdk/lib/async/stream.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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 44 matching lines @@
    * 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
    * the first error is reported. The default is false.
    */
-  StreamSubscription<T> subscribe({void onData(T event),
-                                   void onError(AsyncError error),
-                                   void onDone(),
-                                   bool unsubscribeOnError});
+  StreamSubscription<T> listen(void onData(T event),
+                               { void onError(AsyncError error),
+                                 void onDone(),
+                                 bool unsubscribeOnError});
 
   /**
    * 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 this.chain(new WhereStream<T>(test));
   }
@@ skipping 63 matching lines @@
    */
   Stream transform(StreamTransformer transformer) {
     return this.chain(new TransformStream(transformer));
   }
 
 
   /** Reduces a sequence of values by repeatedly applying [combine]. */
   Future reduce(var initialValue, combine(var previous, T element)) {
     Completer completer = new Completer();
     var value = initialValue;
-    StreamSubscription subscription = this.subscribe(unsubscribeOnError: true);
-    subscription..onData((T element) {
-      try {
-        value = combine(value, element);
-      } catch (e, s) {
-        subscription.unsubscribe();
-        completer.completeError(e, s);
-      }
-    })..onError((AsyncError e) {
-      completer.completeError(e.error, e.stackTrace);
-    })..onDone(() {
-      completer.complete(value);
-    });
+    StreamSubscription subscription;
+    subscription = this.listen(
+      (T element) {
+        try {
+          value = combine(value, element);
+        } catch (e, s) {
+          subscription.cancel();
+          completer.completeError(e, s);
+        }
+      },
+      unsubscribeOnError: true,
+      onError: (AsyncError e) {
+        completer.completeError(e.error, e.stackTrace);
+      },
+      onDone: () {
+        completer.complete(value);
+      });
     return completer.future;
   }
 
   // Deprecated method, previously called 'pipe', retained for compatibility.
   Signal pipeInto(Sink<T> sink,
                   {void onError(AsyncError error),
                    bool unsubscribeOnError}) {
     SignalCompleter completer = new SignalCompleter();
-    this.subscribe(
-        onData: sink.add,
+    this.listen(
+        sink.add,
         onError: onError,
         onDone: () {
           sink.close();
           completer.complete();
         },
         unsubscribeOnError: unsubscribeOnError);
     return completer.signal;
   }
 
 
   /**
    * Check 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) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
-    subscription = subscribe(
-        onData: (T element) {
+    subscription = listen(

[Lasse Reichstein Nielsen, 2013/01/04 08:17:55]

Not your code (probably mine), but ...
This uses "listen" and the above cases uses "this.listen". Should it be
consistent?

[floitsch, 2013/01/04 15:51:36]

Added this. here at the other use-sites.

+        (T element) {
           if (element == match) {
-            subscription.unsubscribe();
+            subscription.cancel();
             future._setValue(true);
           }
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
         unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Check 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)) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
-    subscription = subscribe(
-        onData: (T element) {
+    subscription = listen(
+        (T element) {
           if (!test(element)) {
-            subscription.unsubscribe();
+            subscription.cancel();
             future._setValue(false);
           }
         },
         onError: future._setError,
         onDone: () {
           future._setValue(true);
         },
         unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Check 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)) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
-    subscription = subscribe(
-        onData: (T element) {
+    subscription = listen(
+        (T element) {
           if (test(element)) {
-            subscription.unsubscribe();
+            subscription.cancel();
             future._setValue(true);
           }
         },
         onError: future._setError,
         onDone: () {
           future._setValue(false);
         },
         unsubscribeOnError: true);
     return future;
   }
 
 
   /** Counts the elements in the stream. */
   Future<int> get length {
     _FutureImpl<int> future = new _FutureImpl<int>();
     int count = 0;
-    subscribe(
-      onData: (_) { count++; },
+    listen(
+      (_) { count++; },
       onError: future._setError,
       onDone: () {
         future._setValue(count);
       },
       unsubscribeOnError: true);
     return future;
   }
 
   /**
    * Finds the least element in the stream.
    *
    * If the stream is empty, the result is [:null:].
    * Otherwise the result is a value from the stream that is not greater
    * than any other value from the stream (according to [compare], which must
    * be a [Comparator]).
    *
    * If [compare] is omitted, it defaults to [Comparable.compare].
    */
   Future<T> min([int compare(T a, T b)]) {
     _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     T min = null;
-    subscription = subscribe(
-      onData: (T value) {
+    subscription = listen(
+      (T value) {
         min = value;
         subscription.onData = (T value) {
           if (compare(min, value) > 0) min = value;
         };
       },
       onError: future.setError,
       onDone: () {
         future._setValue(min);
       },
       unsubscribeOnError: true
     );
   }
 
   /**
    * Finds the least element in the stream.
    *
    * If the stream is emtpy, the result is [:null:].
    * Otherwise the result is an value from the stream that is not greater
    * than any other value from the stream (according to [compare], which must
    * be a [Comparator]).
    *
    * If [compare] is omitted, it defaults to [Comparable.compare].
    */
   Future<T> max([int compare(T a, T b)]) {
     _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
     T max = null;
-    subscription = subscribe(
-      onData: (T value) {
+    subscription = listen(
+      (T value) {
         max = value;
         subscription.onData = (T value) {
           if (compare(max, value) < 0) max = value;
         };
       },
       onError: future.setError,
       onDone: () {
         future._setValue(max);
       },
       unsubscribeOnError: true
     );
   }
 
   /** Reports whether this stream contains any elements. */
   Future<bool> get isEmpty {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
-    subscription = subscribe(
-      onData: (_) {
-        subscription.unsubscribe();
+    subscription = listen(
+      (_) {
+        subscription.cancel();
         future._setValue(false);
       },
       onError: future._setError,
       onDone: () {
         future._setValue(true);
       },
       unsubscribeOnError: true);
     return future;
   }
 
   /** Collect the data of this stream in a [List]. */
   Future<List<T>> toList() {
     List<T> result = <T>[];
     _FutureImpl<List<T>> future = new _FutureImpl<List<T>>();
-    subscribe(
-      onData: (T data) {
+    listen(
+      (T data) {
         result.add(data);
       },
       onError: future._setError,
       onDone: () {
         future._setValue(result);
       },
       unsubscribeOnError: true);
     return future;
   }
 
   /** Collect 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>>();
-    subscribe(
-      onData: (T data) {
+    listen(
+      (T data) {
         result.add(data);
       },
       onError: future._setError,
       onDone: () {
         future._setValue(result);
       },
       unsubscribeOnError: true);
     return future;
   }
 
@@ skipping 56 matching lines @@
 
   /**
    * Returns the first element.
    *
    * 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();
     StreamSubscription subscription;
-    subscription = subscribe(
-      onData: (T value) {
+    subscription = listen(
+      (T value) {
         future._setValue(value);
-        subscription.unsubscribe();
+        subscription.cancel();
         return;
       },
       onError: future._setError,
       onDone: () {
         future._setError(new AsyncError(new StateError("No elements")));
       },
       unsubscribeOnError: 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 = subscribe(
-      onData: (T value) {
+    subscription = listen(
+      (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);
     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>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
-    subscription = subscribe(
-      onData: (T value) {
+    subscription = listen(
+      (T value) {
         if (foundResult) {
           // This is the second element we get.
           Error error = new StateError("More than one element");
           future._setError(new AsyncError(error));
-          subscription.unsubscribe();
+          subscription.cancel();
           return;
         }
         foundResult = true;
         result = value;
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
           return;
@@ skipping 14 matching lines @@
    * [defaultValue] function is provided, the result of calling [defaultValue]
    * becomes the value of the future.
    *
    * If an error occurs, or if this stream ends without finding a match and
    * with no [defaultValue] function provided, the future will receive an
    * error.
    */
   Future<T> firstMatching(bool test(T value), {T defaultValue()}) {
     _FutureImpl<T> future = new _FutureImpl<T>();
     StreamSubscription subscription;
-    subscription = subscribe(
-      onData: (T value) {
+    subscription = listen(
+      (T value) {
         bool matches;
         try {
           matches = (true == test(value));
         } catch (e, s) {
           future._setError(new AsyncError(e, s));
-          subscription.unsubscribe();
+          subscription.cancel();
           return;
         }
         if (matches) {
           future._setValue(value);
-          subscription.unsubscribe();
+          subscription.cancel();
         }
       },
       onError: future._setError,
       onDone: () {
         if (defaultValue != null) {
           T value;
           try {
             value = defaultValue();
           } catch (e, s) {
             future._setError(new AsyncError(e, s));
@@ skipping 14 matching lines @@
    *
    * As [firstMatching], except that the last matching element is found.
    * That means that the result cannot be provided before this stream
    * is done.
    */
   Future<T> lastMatching(bool test(T value), {T defaultValue()}) {
     _FutureImpl<T> future = new _FutureImpl<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
-    subscription = subscribe(
-      onData: (T value) {
+    subscription = listen(
+      (T value) {
         bool matches;
         try {
           matches = (true == test(value));
         } catch (e, s) {
           future._setError(new AsyncError(e, s));
-          subscription.unsubscribe();
+          subscription.cancel();
           return;
         }
         if (matches) {
           foundResult = true;
           result = value;
         }
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
@@ skipping 22 matching lines @@
    * 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> singleMatching(bool test(T value)) {
     _FutureImpl<T> future = new _FutureImpl<T>();
     T result = null;
     bool foundResult = false;
     StreamSubscription subscription;
-    subscription = subscribe(
-      onData: (T value) {
+    subscription = listen(
+      (T value) {
         bool matches;
         try {
           matches = (true == test(value));
         } catch (e, s) {
           future._setError(new AsyncError(e, s));
-          subscription.unsubscribe();
+          subscription.cancel();
           return;
         }
         if (matches) {
           if (foundResult) {
             future._setError(new AsyncError(
                 new StateError('Multiple matches for "single"')));
-            subscription.unsubscribe();
+            subscription.cancel();
             return;
           }
           foundResult = true;
           result = value;
         }
       },
       onError: future._setError,
       onDone: () {
         if (foundResult) {
           future._setValue(result);
@@ skipping 10 matching lines @@
    * 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.
    */
   Future<T> elementAt(int index) {
     _FutureImpl<T> future = new _FutureImpl();
     StreamSubscription subscription;
-    subscription = subscribe(
-      onData: (T value) {
+    subscription = listen(
+      (T value) {
         if (index == 0) {
           future._setValue(value);
-          subscription.unsubscribe();
+          subscription.cancel();
           return;
         }
         index -= 1;
       },
       onError: future._setError,
       onDone: () {
         future._setError(new AsyncError(
             new StateError("Not enough elements for elementAt")));
       },
       unsubscribeOnError: true);
     return future;
   }
 }
 
 /**
  * A control object for the subscription on a [Stream].
  *
  * When you subscribe on a [Stream] using [Stream.subscribe],
  * a [StreamSubscription] object is returned. This object
  * is used to later unsubscribe again, or to temporarily pause
  * the stream's events.
  */
 abstract class StreamSubscription<T> {
   /**
    * Cancels this subscription. It will no longer receive events.
    *
    * If an event is currently firing, this unsubscription will only
    * take effect after all subscribers have received the current event.
    */
-  void unsubscribe();
+  void cancel();
 
   /** Set or override the data event handler of this subscription. */
   void onData(void handleData(T data));
 
   /** Set or override the error event handler of this subscription. */
   void onError(void handleError(AsyncError error));
 
   /** Set or override the done event handler of this subscription. */
   void onDone(void handleDone());
 
@@ skipping 25 matching lines @@
   void signalError(AsyncError errorEvent);
   void close();
 }
 
 /** [Stream] wrapper that only exposes the [Stream] interface. */
 class StreamView<T> extends Stream<T> {
   Stream<T> _stream;
 
   StreamView(this._stream);
 
-  StreamSubscription<T> subscribe({void onData(T value),
-                                   void onError(AsyncError error),
-                                   void onDone(),
-                                   bool unsubscribeOnError}) {
-    return _stream.subscribe(onData: onData, onError: onError, onDone: onDone,
+  StreamSubscription<T> listen(void onData(T value),
+                               { void onError(AsyncError error),
+                                 void onDone(),
+                                 bool unsubscribeOnError}) {
+    return _stream.listen(onData, onError: onError, onDone: onDone,
                              unsubscribeOnError: unsubscribeOnError);
   }
 }
 
 /**
  * [StreamSink] wrapper that only exposes the [StreamSink] interface.
  */
 class StreamSinkView<T> implements StreamSink<T> {
   final StreamSink<T> _sink;
 
@@ skipping 67 matching lines @@
     sink.signalError(error);
   }
 
   /**
    * Handle an incoming done event.
    */
   void handleDone(StreamSink<T> sink) {
     sink.close();
   }
 }