add anyOf

Author: zhongkechen

examples/src/main/java/software/amazon/lambda/durable/examples/WaitExample.java

@@ -4,6 +4,7 @@
 
 import java.time.Duration;
 import software.amazon.lambda.durable.DurableContext;
+import software.amazon.lambda.durable.DurableFuture;
 import software.amazon.lambda.durable.DurableHandler;
 
 /**
@@ -31,11 +32,19 @@ public String handleRequest(GreetingRequest input, DurableContext context) {
         context.wait(null, Duration.ofSeconds(10));
 
         // Step 2: Continue processing
-        var continued =
-                context.step("continue-processing", String.class, stepCtx -> started + " - continued after 10s");
-
-        // Wait 5 seconds
-        context.wait(null, Duration.ofSeconds(5));
+        var continued = context.stepAsync("continue-processing", String.class, stepCtx -> {
+            try {
+                Thread.sleep(10000);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+            return started + " - continued after 10s";
+        });
+
+        // Wait at most seconds
+        var wait5seconds = context.waitAsync(null, Duration.ofSeconds(5));
+
+        DurableFuture.anyOf(continued, wait5seconds);
 
         // Step 3: Complete
         var result =

sdk/src/main/java/software/amazon/lambda/durable/DurableFuture.java

@@ -4,6 +4,8 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import software.amazon.lambda.durable.operation.BaseDurableOperation;
 
 /**
  * A future representing the result of an asynchronous durable operation.
@@ -52,4 +54,17 @@ static <T> List<T> allOf(DurableFuture<T>... futures) {
     static <T> List<T> allOf(List<DurableFuture<T>> futures) {
         return futures.stream().map(DurableFuture::get).toList();
     }
+
+    /**
+     * Waits for any of the provided futures to complete and returns its result.
+     *
+     * @param futures the futures to wait for
+     * @return the result of the first future to complete
+     */
+    static Object anyOf(DurableFuture<?>... futures) {
+        return CompletableFuture.anyOf(Arrays.stream(futures)
+                        .map(f -> ((BaseDurableOperation) f).getCompletionFuture())
+                        .toArray(CompletableFuture[]::new))
+                .join();
+    }
 }

sdk/src/main/java/software/amazon/lambda/durable/operation/BaseDurableOperation.java

@@ -67,7 +67,7 @@ protected BaseDurableOperation(OperationIdentifier operationIdentifier, DurableC
         executionManager.registerOperation(this);
     }
 
-    protected CompletableFuture<BaseDurableOperation> getCompletionFuture() {
+    public CompletableFuture<BaseDurableOperation> getCompletionFuture() {
         return completionFuture;
     }