Add Future.any and Stream.fromFutures.

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 95 matching lines @@
         controller._closeUnchecked();
       },
       onError: (error, stackTrace) {
         controller._addError(error, stackTrace);
         controller._closeUnchecked();
       });
     return controller.stream;
   }
 
   /**
+   * Create a stream from a group of futures.
+   *
+   * The stream reports the results of the futures on the stream in the order
+   * in which the futures complete.
+   *
+   * If some futures have completed before calling `Stream.fromFutures`,
+   * their result will be output on the created stream in some unspecified
+   * order.
+   *
+   * When all futures have completed, the stream is closed.
+   *
+   * If no future is passed, the stream closes as soon as possible.
+   */
+  factory Stream.fromFutures(Iterable<Future<T>> futures) {
+    var controller;
+    controller = new StreamController<T>(sync: true);

[floitsch, 2016/01/07 13:57:13]

Merge these two lines?

[Lasse Reichstein Nielsen, 2016/01/08 09:18:48]

Done.

+    int count = 0;
+    var onValue = (value) {
+      if (!controller.isClosed) {
+        controller._add(value);
+        if (--count == 0) controller._closeUnchecked();
+      }
+    };
+    var onError = (error, stack) {
+      if (!controller.isClosed) {
+        controller._addError(error, stack);
+        if (--count == 0) controller._closeUnchecked();
+      }
+    };
+    // The futures are already running, so start listening to them immediately
+    // (instead of waiting for the stream to be listened on).
+    // If we wait, we might not catch errors in the futures in time.
+    for (var future in futures) {
+      count++;
+      future.then(onValue, onError: onError);
+    }
+    // Use schedule microtask since controller is sync.
+    if (count == 0) scheduleMicrotask(controller.close);
+    return controller.stream;
+  }
+
+  /**
    * Creates a single-subscription stream that gets its data from [data].
    *
    * The iterable is iterated when the stream receives a listener, and stops
    * iterating if the listener cancels the subscription.
    *
    * If iterating [data] throws an error, the stream ends immediately with
    * that error. No done event will be sent (iteration is not complete), but no
    * further data events will be generated either, since iteration cannot
    * continue.
    */
@@ skipping 1625 matching lines @@
 class _ControllerEventSinkWrapper<T> implements EventSink<T> {
   EventSink _sink;
   _ControllerEventSinkWrapper(this._sink);
 
   void add(T data) { _sink.add(data); }
   void addError(error, [StackTrace stackTrace]) {
     _sink.addError(error, stackTrace);
   }
   void close() { _sink.close(); }
 }