Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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> {
Expand All @@ -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;
Expand All @@ -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()
Expand Down
31 changes: 31 additions & 0 deletions sdk/src/main/java/com/amazonaws/lambda/durable/DurableFuture.java
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think we should restrict the futures to have the same result types. Instead, maybe we should return a future of Void?

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();
}
}
Original file line number Diff line number Diff line change
@@ -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;
}
}