Add min and max to Iterable and Stream.

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 251 matching lines @@
     subscribe(
       onData: (_) { 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) {
+        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

[Sean Eagan, 2013/01/02 16:35:13]

greater -> less ?

+   * 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) {
+        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();
         future._setValue(false);
       },
       onError: future._setError,
@@ skipping 501 matching lines @@
     sink.signalError(error);
   }
 
   /**
    * Handle an incoming done event.
    */
   void handleDone(StreamSink<T> sink) {
     sink.close();
   }
 }