Add more methods to 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 246 matching lines @@
       onData: (T data) {
         result.add(data);
       },
       onError: future._setError,
       onDone: () {
         future._setValue(result);
       },
       unsubscribeOnError: true);
     return future;
   }
+
+  /**
+   * Provide 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 less than [n] values before it's done,

[Bob Nystrom, 2012/11/26 16:10:19]

Nit: "less" -> "fewer". :)

+   * so will the returned stream.
+   */
+  Stream<T> take(int n) {
+    return new TakeStream(this, n);
+  }
+
+  /**
+   * Forwards data events while [test] is successful.
+   *
+   * The returned stream provides the same events as this stream as long
+   * as [test] returns [:true:] for the event data. The stream is done
+   * when either this stream is done, or when this stream first provides
+   * a value that [test] doesn't accept.
+   */
+  Stream<T> takeWhile(bool test(T value)) {
+    return new TakeWhileStream(this, test);
+  }
+
+  /**
+   * Skips the first [n] data events from this stream.
+   */
+  Stream<T> skip(int n) {
+    return new SkipStream(this, n);
+  }
+
+  /**
+   * Skip data events from this stream while they are matched by [test].
+   *
+   * Error and done events are provided by the returned stream unmodified.
+   *
+   * Starting with the first data event where [test] returns true for the
+   * event data, the returned stream will have the same events as this stream.
+   */
+  Stream<T> skipWhile(bool test(T value)) {
+    return new SkipWhileStream(this, test);
+  }
+
+  /**
+   * Skip data events if they are equal to the previous data event.
+   *
+   * The returned stream provides the same events as this stream, except
+   * that it never provides two consequtive data events that are equal.
+   *
+   * Equality is determined by the provided [equals] method. If that is
+   * omitted, the '==' operator on the last provided data element is used.
+   */
+  Stream<T> distinct([bool equals(T a, T b)]) {
+    return new DistinctStream(this, compare);
+  }
+
+  /**
+   * Find 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
+   * [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> firstMatch(bool test(T value), {T defaultValue()}) {
+    _FutureImpl<T> future = new _FutureImpl<T>();
+    StreamSubscription subscription;
+    subscription = subscribe(
+      onData: (T value) {
+        bool matches;
+        try {
+          matches = (true == test(value));
+        } catch (e, s) {
+          future._setError(new AsyncError(e, s));
+          subscription.unsubscribe();
+          return;
+        }
+        if (matches) {
+          future._setValue(value);
+          subscription.unsubscribe();
+        }
+      },
+      onError: future._setError,
+      onDone: () {
+        if (defaultValue != null) {
+          T value;
+          try {
+            value = defaultValue();
+          } catch (e, s) {
+            future._setError(new AsyncError(e, s));
+            return;
+          }
+          future._setValue(value);
+          return;
+        }
+        future._setError(
+            new AsyncError(new StateError("firstMatch ended without match")));
+      },
+      unsubscribeOnError: true);
+    return future;
+  }
+
+  /**
+   * Finds the last element in this stream matching [test].
+   *
+   * As [firstMatch], except that the last matching element is found.
+   * That means that the result cannot be provided before this stream
+   * is done.
+   */
+  Future<T> lastMatch(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) {
+        bool matches;
+        try {
+          matches = (true == test(value));
+        } catch (e, s) {
+          future._setError(new AsyncError(e, s));
+          subscription.unsubscribe();
+          return;
+        }
+        if (matches) {
+          foundResult = true;
+          result = value;
+        }
+      },
+      onError: future._setError,
+      onDone: () {
+        if (foundResult) {
+          future._setValue(result);
+          return;
+        }
+        if (defaultValue != null) {
+          T value;
+          try {
+            value = defaultValue();
+          } catch (e, s) {
+            future._setError(new AsyncError(e, s));
+            return;
+          }
+          future._setValue(value);
+          return;
+        }
+        future._setError(
+            new AsyncError(new StateError("lastMatch ended without match")));
+      },
+      unsubscribeOnError: 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> single(bool test(T value)) {
+    _FutureImpl<T> future = new _FutureImpl<T>();
+    T result = null;
+    bool foundResult = false;
+    StreamSubscription subscription;
+    subscription = subscribe(
+      onData: (T value) {
+        bool matches;
+        try {
+          matches = (true == test(value));
+        } catch (e, s) {
+          future._setError(new AsyncError(e, s));
+          subscription.unsubscribe();
+          return;
+        }
+        if (matches) {
+          if (foundResult) {
+            future._setError(new AsyncError(
+                new StateError('Multiple matches for "single"')));
+            subscription.unsubscribe();
+            return;
+          }
+          foundResult = true;
+          result = value;
+        }
+      },
+      onError: future._setError,
+      onDone: () {
+        if (foundResult) {
+          future._setValue(result);
+          return;
+        }
+        future._setError(
+            new AsyncError(new StateError("single ended without match")));
+      },
+      unsubscribeOnError: 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.
+   */
+  Future<T> elementAt(int index) {
+    _FutureImpl<T> future = new _FutureImpl();
+    StreamSubscription subscription;
+    subscription = subscribe(
+      onData: (T value) {
+        if (index == 0) {
+          future._setValue(value);
+          subscription.unsubscribe();
+          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.
  */
@@ skipping 97 matching lines @@
     sink.signalError(error);
   }
 
   /**
    * Handle an incoming done event.
    */
   void handleDone(StreamSink<T> sink) {
     sink.close(data);
   }
 }