feat: durable future allof

* feat: all of

* feat: all of

Author: maschnetwork

examples/src/main/java/com/amazonaws/lambda/durable/examples/ManyAsyncStepsExample.java

@@ -16,7 +16,7 @@
  * <ul>
  *   <li>Creates async steps in a loop
  *   <li>Each step performs a simple computation
- *   <li>All results are collected and summed
+ *   <li>All results are collected using {@link DurableFuture#allOf}
  * </ul>
  */
 public class ManyAsyncStepsExample extends DurableHandler<ManyAsyncStepsExample.Input, String> {
@@ -32,7 +32,7 @@ public String handleRequest(Input input, DurableContext context) {
 
         context.getLogger().info("Starting {} async steps with multiplier {}", STEP_COUNT, multiplier);
 
-        // Create 100 async steps
+        // Create async steps
         var futures = new ArrayList<DurableFuture<Integer>>(STEP_COUNT);
         for (var i = 0; i < STEP_COUNT; i++) {
             var index = i;
@@ -42,11 +42,9 @@ public String handleRequest(Input input, DurableContext context) {
 
         context.getLogger().info("All {} async steps created, collecting results", STEP_COUNT);
 
-        // Collect all results
-        var totalSum = 0L;
-        for (var future : futures) {
-            totalSum += future.get();
-        }
+        // Collect all results using allOf
+        var results = DurableFuture.allOf(futures);
+        var totalSum = results.stream().mapToInt(Integer::intValue).sum();
 
         var executionTimeMs = System.currentTimeMillis() - startTime;
         context.getLogger()

sdk/src/main/java/com/amazonaws/lambda/durable/DurableFuture.java

@@ -3,6 +3,8 @@
 package com.amazonaws.lambda.durable;
 
 import com.amazonaws.lambda.durable.operation.DurableOperation;
+import java.util.Arrays;
+import java.util.List;
 
 public class DurableFuture<T> {
     private final DurableOperation<T> operation;
@@ -22,4 +24,33 @@ public DurableFuture(DurableOperation<T> operation) {
     public T get() {
         return operation.get();
     }
+
+    /**
+     * Waits for all provided futures to complete and returns their results in order.
+     *
+     * <p>The futures are resolved sequentially, but since the underlying operations run concurrently, this effectively
+     * waits for all operations to complete. During replay, completed operations return immediately.
+     *
+     * @param futures the futures to wait for
+     * @param <T> the result type of the futures
+     * @return a list of results in the same order as the input futures
+     */
+    @SafeVarargs
+    public static <T> List<T> allOf(DurableFuture<T>... futures) {
+        return Arrays.stream(futures).map(DurableFuture::get).toList();
+    }
+
+    /**
+     * Waits for all provided futures to complete and returns their results in order.
+     *
+     * <p>The futures are resolved sequentially, but since the underlying operations run concurrently, this effectively
+     * waits for all operations to complete. During replay, completed operations return immediately.
+     *
+     * @param futures the list of futures to wait for
+     * @param <T> the result type of the futures
+     * @return a list of results in the same order as the input futures
+     */
+    public static <T> List<T> allOf(List<DurableFuture<T>> futures) {
+        return futures.stream().map(DurableFuture::get).toList();
+    }
 }

sdk/src/test/java/com/amazonaws/lambda/durable/DurableFutureTest.java

@@ -0,0 +1,88 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+package com.amazonaws.lambda.durable;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import com.amazonaws.lambda.durable.operation.DurableOperation;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+class DurableFutureTest {
+
+    @Test
+    void allOfVarargsReturnsResultsInOrder() {
+        var op1 = mockOperation("first");
+        var op2 = mockOperation("second");
+        var op3 = mockOperation("third");
+
+        var future1 = new DurableFuture<>(op1);
+        var future2 = new DurableFuture<>(op2);
+        var future3 = new DurableFuture<>(op3);
+
+        var results = DurableFuture.allOf(future1, future2, future3);
+
+        assertEquals(List.of("first", "second", "third"), results);
+        verify(op1).get();
+        verify(op2).get();
+        verify(op3).get();
+    }
+
+    @Test
+    void allOfListReturnsResultsInOrder() {
+        var op1 = mockOperation(1);
+        var op2 = mockOperation(2);
+        var op3 = mockOperation(3);
+
+        var futures = List.of(new DurableFuture<>(op1), new DurableFuture<>(op2), new DurableFuture<>(op3));
+
+        var results = DurableFuture.allOf(futures);
+
+        assertEquals(List.of(1, 2, 3), results);
+    }
+
+    @Test
+    void allOfVarargsEmptyReturnsEmptyList() {
+        var results = DurableFuture.<String>allOf();
+
+        assertTrue(results.isEmpty());
+    }
+
+    @Test
+    void allOfListEmptyReturnsEmptyList() {
+        var results = DurableFuture.allOf(List.<DurableFuture<String>>of());
+
+        assertTrue(results.isEmpty());
+    }
+
+    @Test
+    void allOfSingleFutureReturnsSingleResult() {
+        var op = mockOperation("only");
+        var future = new DurableFuture<>(op);
+
+        var results = DurableFuture.allOf(future);
+
+        assertEquals(List.of("only"), results);
+    }
+
+    @Test
+    void allOfPropagatesException() {
+        var op1 = mockOperation("first");
+        @SuppressWarnings("unchecked")
+        DurableOperation<String> op2 = mock(DurableOperation.class);
+        when(op2.get()).thenThrow(new RuntimeException("Step failed"));
+
+        var future1 = new DurableFuture<>(op1);
+        var future2 = new DurableFuture<>(op2);
+
+        assertThrows(RuntimeException.class, () -> DurableFuture.allOf(future1, future2));
+    }
+
+    @SuppressWarnings("unchecked")
+    private <T> DurableOperation<T> mockOperation(T result) {
+        DurableOperation<T> op = mock(DurableOperation.class);
+        when(op.get()).thenReturn(result);
+        return op;
+    }
+}