Add Stream.join

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 310 matching lines @@
         result._setError(e);
       },
       onDone: () {
         result._setValue(value);
       },
       cancelOnError: true);
     return result;
   }
 
   /**
+   * Collects string of data events' string representations.
+   *
+   * If [separator] is provided, it is inserted between any two
+   * elements.
+   *
+   * Any error in the stream causes the future to complete with that
+   * error. Otherwise it completes with the collected string when
+   * the "done" event arrives.
+   */
+  Future<String> join([String separator = ""]) {
+    _FutureImpl<String> result = new _FutureImpl<String>();
+    StringBuffer buffer = new StringBuffer();
+    StreamSubscription subscription;
+    bool first = true;
+    subscription = this.listen(
+      (T element) {
+        if (!first) {
+          buffer.write(separator);
+        }
+        first = false;
+        try {
+          buffer.write(element);
+        } catch (e, s) {
+          subscription.cancel();
+          result._setError(_asyncError(e, s));
+        }
+      },
+      onError: (e) {
+        result._setError(e);
+      },
+      onDone: () {
+        result._setValue(buffer.toString());
+      },
+      cancelOnError: true);
+    return result;
+  }
+
+  /**
    * Checks whether [needle] 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(Object needle) {
     _FutureImpl<bool> future = new _FutureImpl<bool>();
     StreamSubscription subscription;
     subscription = this.listen(
         // TODO(ahe): Restore type when feature is implemented in dart2js
@@ skipping 910 matching lines @@
    *
    * If a [moveNext] call has been made, it will complete with `false` as value,
    * as will all further calls to [moveNext].
    *
    * If you need to stop listening for values before the stream iterator is
    * automatically closed, you must call [cancel] to ensure that the stream
    * is properly closed.
    */
   void cancel();
 }