Add cleanUp function to Future.wait.

The cleanUp function is called on the values of successful futures,
so it can release any resources that were successfully allocated.

R=sgjesse@google.com

Review URL: https://codereview.chromium.org//815773002

git-svn-id: https://dart.googlecode.com/svn/branches/bleeding_edge/dart@42730 260f80e4-7a28-3924-810f-c04153c831b5

pkg/async_helper/lib/async_helper.dart

@@ -91,4 +91,4 @@ void asyncSuccess(_) => asyncEnd();
 void asyncTest(f()) {
   asyncStart();
   f().then(asyncSuccess);
-}
+}

sdk/lib/async/future.dart

@@ -239,14 +239,24 @@ abstract class Future<T> {
    * Returns a future which will complete once all the futures in a list are
    * complete. If any of the futures in the list completes with an error,
    * the resulting future also completes with an error. Otherwise the value
-   * of the returned future will be a list of all the values that were produced.
+   * of the returned future will be a list of all the values that were
+   * produced.
    *
    * If `eagerError` is true, the future completes with an error immediately on
    * the first error from one of the futures. Otherwise all futures must
    * complete before the returned future is completed (still with the first
    * error to occur, the remaining errors are silently dropped).
+   *
+   * If [cleanUp] is provided, in the case of an error, any non-null result of
+   * a successful future is passed to `cleanUp`, which can then release any
+   * resources that the successful operation allocated.
+   *
+   * The call to `cleanUp` should not throw. If it does, the error will be an
+   * uncaught asynchronous error.
    */
-  static Future<List> wait(Iterable<Future> futures, {bool eagerError: false}) {
+  static Future<List> wait(Iterable<Future> futures,
+                           {bool eagerError: false,
+                            void cleanUp(successValue)}) {
     final _Future<List> result = new _Future<List>();
     List values;  // Collects the values. Set to null on error.
     int remaining = 0;  // How many futures are we waiting for.
@@ -254,11 +264,18 @@ abstract class Future<T> {
     StackTrace stackTrace;  // The stackTrace that came with the error.
 
     // Handle an error from any of the futures.
-    handleError(theError, theStackTrace) {
-      final bool isFirstError = (values != null);
-      values = null;
+    void handleError(theError, theStackTrace) {
       remaining--;
-      if (isFirstError) {
+      if (values != null) {
+        if (cleanUp != null) {
+          for (var value in values) {
+            if (value != null) {
+              // Ensure errors from cleanUp are uncaught.
+              new Future.sync(() { cleanUp(value); });
+            }
+          }
+        }
+        values = null;
         if (remaining == 0 || eagerError) {
           result._completeError(theError, theStackTrace);
         } else {
@@ -281,8 +298,14 @@ abstract class Future<T> {
           if (remaining == 0) {
             result._completeWithValue(values);
           }
-        } else if (remaining == 0 && !eagerError) {
-          result._completeError(error, stackTrace);
+        } else {
+          if (cleanUp != null && value != null) {
+            // Ensure errors from cleanUp are uncaught.
+            new Future.sync(() { cleanUp(value); });
+          }
+          if (remaining == 0 && !eagerError) {
+            result._completeError(error, stackTrace);
+          }
         }
       }, onError: handleError);
     }

tests/lib/async/future_test.dart

@@ -749,6 +749,133 @@ void testSyncFuture_i13368() {
     });
 }
 
+void testWaitCleanUp() {
+  asyncStart();
+  // Creates three futures with different completion times, and where some fail.
+  // The `mask` specifies which futures fail (values 1-7),
+  // and `permute` defines the order of completion. values 0-5.
+  void doTest(int mask, int permute) {
+    asyncStart();
+    String stringId = "waitCleanup-$mask-$permute";
+    List futures = new List(3);
+    List cleanup = new List(3);
+    int permuteTmp = permute;
+    for (int i = 0; i < 3; i++) {
+      bool throws = (mask & (1 << i)) != 0;
+      var future = new Future.delayed(
+          new Duration(milliseconds: 100 * (i + 1)),
+          () => (throws ? throw "Error $i($mask-$permute)" : i));
+      int mod = 3 - i;
+      int position = permuteTmp % mod;
+      permuteTmp = permuteTmp ~/ mod;
+      while (futures[position] != null) position++;
+      futures[position] = future;
+      cleanup[i] = throws;
+    }
+    void cleanUp(index) {
+      Expect.isFalse(cleanup[index]);
+      cleanup[index] = true;
+    }
+    Future.wait(futures, cleanUp: cleanUp)
+          .then((_) { Expect.fail("No error: $stringId"); },
+                onError: (e, s) {
+                  Expect.listEquals([true, true, true], cleanup);
+                  asyncEnd();
+                });
+  }
+
+  for (int i = 1; i < 8; i++) {
+    for (int j = 0; j < 6; j++) {
+      doTest(i, j);
+    }
+  }
+  asyncEnd();
+}
+
+void testWaitCleanUpEager() {
+  asyncStart();
+  // Creates three futures with different completion times, and where some fail.
+  // The `mask` specifies which futures fail (values 1-7),
+  // and `permute` defines the order of completion. values 0-5.
+  void doTest(int mask, int permute) {
+    asyncStart();
+    asyncStart();
+    bool done = false;
+    String stringId = "waitCleanup-$mask-$permute";
+    List futures = new List(3);
+    List cleanup = new List(3);
+    int permuteTmp = permute;
+    for (int i = 0; i < 3; i++) {
+      bool throws = (mask & (1 << i)) != 0;
+      var future = new Future.delayed(
+          new Duration(milliseconds: 100 * (i + 1)),
+          () => (throws ? throw "Error $i($mask-$permute)" : i));
+      int mod = 3 - i;
+      int position = permuteTmp % mod;
+      permuteTmp = permuteTmp ~/ mod;
+      while (futures[position] != null) position++;
+      futures[position] = future;
+      cleanup[i] = throws;
+    }
+    void checkDone() {
+      if (done) return;
+      if (cleanup.every((v) => v)) {
+        done = true;
+        asyncEnd();
+      }
+    }
+    void cleanUp(index) {
+      Expect.isFalse(cleanup[index]);
+      cleanup[index] = true;
+      // Cleanup might happen before and after the wait().then() callback.
+      checkDone();
+    }
+    Future.wait(futures, eagerError: true, cleanUp: cleanUp)
+          .then((_) { Expect.fail("No error: $stringId"); },
+                onError: (e, s) {
+                  asyncEnd();
+                  checkDone();
+                });
+  }
+
+  for (int i = 1; i < 8; i++) {
+    for (int j = 0; j < 6; j++) {
+      doTest(i, j);
+    }
+  }
+  asyncEnd();
+}
+
+void testWaitCleanUpError() {
+  var cms = const Duration(milliseconds: 100);
+  var cleanups = new List.filled(3, false);
+  var uncaughts = new List.filled(3, false);
+  asyncStart();
+  asyncStart();
+  asyncStart();
+  runZoned(() {
+    Future.wait([new Future.delayed(cms, () => 0),
+                 new Future.delayed(cms * 2, ()=> throw 1),
+                 new Future.delayed(cms * 3, () => 2)],
+                 cleanUp: (index) {
+                   Expect.isTrue(index == 0 || index == 2, "$index");
+                   Expect.isFalse(cleanups[index]);
+                   cleanups[index] = true;
+                   throw index;
+                 })
+          .catchError((e) {
+            Expect.equals(e, 1);
+            asyncEnd();
+          });
+
+  }, onError: (int index, s) {
+    Expect.isTrue(index == 0 || index == 2, "$index");
+    Expect.isFalse(uncaughts[index]);
+    uncaughts[index] = true;
+    asyncEnd();
+  });
+}
+
 main() {
   asyncStart();
 
@@ -805,6 +932,9 @@ main() {
 
   testSyncFuture_i13368();
 
+  testWaitCleanUp();
+  testWaitCleanUpError();
+
   asyncEnd();
 }