Skip to content

Commit a5f4354

Browse files
zhongkechenzhongkechen
authored
[feat]: make initialState optional (aws#254)
* remove initialState from waitForCondition * add WaitForConditionFailedException and remove unused exceptions * format code * remove test case for unused exception * make initialState optional in config * update example to use optional config * remove unused variable from example * revert some doc changes * update doc with the new config * revert another doc change * update comment in example
1 parent 263864e commit a5f4354

16 files changed

Lines changed: 111 additions & 266 deletions

File tree

‎docs/advanced/error-handling.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ DurableExecutionException - General durable exception
1717
├── CallbackException - General callback exception
1818
│ ├── CallbackFailedException - External system sent an error response to the callback. Handle the error or propagate failure
1919
│ └── CallbackTimeoutException - Callback exceeded its timeout duration. Handle the error or propagate the failure
20-
├── WaitForConditionException - waitForCondition exceeded max polling attempts or failed. Catch to implement fallback logic.
20+
├── WaitForConditionFailedException- waitForCondition exceeded max polling attempts or failed. Catch to implement fallback logic.
2121
└── ChildContextFailedException - Child context failed and the original exception could not be reconstructed
2222
```
2323

‎docs/spec/waitForCondition.md‎

Lines changed: 15 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88

99
### How it works
1010

11-
1. User calls `ctx.waitForCondition(name, resultType, checkFunc, initialState)` (or with optional config)
11+
1. User calls `ctx.waitForCondition(name, resultType, checkFunc)` (or with optional config)
1212
2. A `WaitForConditionOperation` is created with a unique operation ID
1313
3. On first execution:
1414
- Checkpoint START with subtype `WAIT_FOR_CONDITION`
@@ -43,14 +43,14 @@ sdk/src/main/java/software/amazon/lambda/durable/
4343

4444
```
4545
DurableContext (interface)
46-
├── waitForCondition(name, Class<T>, checkFunc, initialState) → T
47-
├── waitForCondition(name, Class<T>, checkFunc, initialState, config) → T
48-
├── waitForCondition(name, TypeToken<T>, checkFunc, initialState) → T
49-
├── waitForCondition(name, TypeToken<T>, checkFunc, initialState, config) → T
50-
├── waitForConditionAsync(name, Class<T>, checkFunc, initialState) → DurableFuture<T>
51-
├── waitForConditionAsync(name, Class<T>, checkFunc, initialState, config) → DurableFuture<T>
52-
├── waitForConditionAsync(name, TypeToken<T>, checkFunc, initialState) → DurableFuture<T>
53-
└── waitForConditionAsync(name, TypeToken<T>, checkFunc, initialState, config) → DurableFuture<T>
46+
├── waitForCondition(name, Class<T>, checkFunc) → T
47+
├── waitForCondition(name, Class<T>, checkFunc, config) → T
48+
├── waitForCondition(name, TypeToken<T>, checkFunc) → T
49+
├── waitForCondition(name, TypeToken<T>, checkFunc, config) → T
50+
├── waitForConditionAsync(name, Class<T>, checkFunc) → DurableFuture<T>
51+
├── waitForConditionAsync(name, Class<T>, checkFunc, config) → DurableFuture<T>
52+
├── waitForConditionAsync(name, TypeToken<T>, checkFunc) → DurableFuture<T>
53+
└── waitForConditionAsync(name, TypeToken<T>, checkFunc, config) → DurableFuture<T>
5454
│
5555
▼
5656
WaitForConditionOperation<T> extends BaseDurableOperation<T>
@@ -131,29 +131,31 @@ Validation: maxAttempts > 0, initialDelay >= 1s, maxDelay >= 1s, backoffRate >=
131131
public class WaitForConditionConfig<T> {
132132
public static <T> Builder<T> builder();
133133

134-
public WaitForConditionWaitStrategy<T> waitStrategy(); // defaults to WaitStrategies.defaultStrategy()
134+
public WaitForConditionWaitStrategy<T> waitStrategy(); // defaults to WaitStrategies.defaultStrategy()
135135
public SerDes serDes(); // defaults to null (uses handler default)
136+
public T initialState(); // defaults to null
136137
public Builder<T> toBuilder(); // for internal SerDes injection
137138

138139
public static class Builder<T> {
139140
public Builder<T> waitStrategy(WaitForConditionWaitStrategy<T> waitStrategy);
140141
public Builder<T> serDes(SerDes serDes);
142+
public Builder<T> initialState(T state);
141143
public WaitForConditionConfig<T> build();
142144
}
143145
}
144146
```
145147

146-
Holds only optional parameters. Required parameters (`initialState`, `checkFunc`) are direct method arguments on `DurableContext.waitForCondition()`.
148+
Holds only optional parameters. The required parameter `checkFunc` is direct method argument on `DurableContext.waitForCondition()`.
147149

148150
### DurableContext API (8 signatures)
149151

150152
Delegation chain (same pattern as `step()`):
151153
- All sync methods → corresponding async method → `.get()`
152154
- All Class-based methods → TypeToken-based via `TypeToken.get(resultType)`
153155
- All no-config methods → config method with `WaitForConditionConfig.builder().build()`
154-
- Core method: `waitForConditionAsync(name, TypeToken<T>, checkFunc, initialState, config)`
156+
- Core method: `waitForConditionAsync(name, TypeToken<T>, checkFunc, config)`
155157

156-
The core method validates: `name` (via `ParameterValidator`), `typeToken` not null, `checkFunc` not null, `initialState` not null, `config` not null.
158+
The core method validates: `name` (via `ParameterValidator`), `typeToken` not null, `checkFunc` not null, `config` not null.
157159

158160
### WaitForConditionOperation\<T\>
159161

‎examples/src/main/java/software/amazon/lambda/durable/examples/WaitForConditionExample.java‎

Lines changed: 10 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -2,38 +2,34 @@
22
// SPDX-License-Identifier: Apache-2.0
33
package software.amazon.lambda.durable.examples;
44

5-
import java.util.concurrent.atomic.AtomicInteger;
65
import software.amazon.lambda.durable.DurableContext;
76
import software.amazon.lambda.durable.DurableHandler;
7+
import software.amazon.lambda.durable.config.WaitForConditionConfig;
88
import software.amazon.lambda.durable.model.WaitForConditionResult;
99

1010
/**
1111
* Example demonstrating the waitForCondition operation.
1212
*
1313
* <p>This example simulates waiting for an order to ship, by repeatedly calling a check function.
1414
*/
15-
public class WaitForConditionExample extends DurableHandler<String, String> {
16-
17-
private final AtomicInteger callCount = new AtomicInteger(0);
15+
public class WaitForConditionExample extends DurableHandler<Integer, Integer> {
1816

1917
@Override
20-
public String handleRequest(String input, DurableContext context) {
18+
public Integer handleRequest(Integer input, DurableContext context) {
2119
// Poll the shipment status until the order is shipped.
22-
// The check function simulates an order shipment status
23-
// which transitions from PENDING > PROCESSING > SHIPPED
20+
// The check function simulates an order shipment (0 -> 1 -> 2 -> 3 -> 4)
2421
return context.waitForCondition(
2522
"wait-for-shipment",
26-
String.class,
27-
(status, stepCtx) -> {
23+
Integer.class,
24+
(callCount, stepCtx) -> {
2825
// Simulate checking shipment status from an external service
29-
var count = callCount.incrementAndGet();
30-
if (count >= 3) {
26+
if (callCount >= 3) {
3127
// Order has shipped — stop polling
32-
return WaitForConditionResult.stopPolling(input + ": SHIPPED");
28+
return WaitForConditionResult.stopPolling(callCount + 1);
3329
}
3430
// Order still processing — continue polling
35-
return WaitForConditionResult.continuePolling(input + ": PROCESSING");
31+
return WaitForConditionResult.continuePolling(callCount + 1);
3632
},
37-
input + ": PENDING"); // Order pending - initial status
33+
WaitForConditionConfig.<Integer>builder().initialState(1).build()); // Order pending - initial status
3834
}
3935
}

‎examples/src/test/java/software/amazon/lambda/durable/examples/WaitForConditionExampleTest.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,11 +13,11 @@ class WaitForConditionExampleTest {
1313
@Test
1414
void testWaitForConditionExample() {
1515
var handler = new WaitForConditionExample();
16-
var runner = LocalDurableTestRunner.create(String.class, handler);
16+
var runner = LocalDurableTestRunner.create(Integer.class, handler);
1717

18-
var result = runner.runUntilComplete("order-123");
18+
var result = runner.runUntilComplete(123);
1919

2020
assertEquals(ExecutionStatus.SUCCEEDED, result.getStatus());
21-
assertEquals("order-123: SHIPPED", result.getResult(String.class));
21+
assertEquals(4, result.getResult(Integer.class));
2222
}
2323
}

‎sdk-integration-tests/src/test/java/software/amazon/lambda/durable/WaitForConditionIntegrationTest.java‎

Lines changed: 15 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -39,13 +39,15 @@ void testBasicPollingSucceedsAfterNChecks() {
3939
"poll-counter",
4040
Integer.class,
4141
(state, stepCtx) -> {
42+
if (state == null) {
43+
state = 0;
44+
}
4245
checkCount.incrementAndGet();
4346
var next = state + 1;
4447
return next >= targetCount
4548
? WaitForConditionResult.stopPolling(next)
4649
: WaitForConditionResult.continuePolling(next);
4750
},
48-
0,
4951
config);
5052
});
5153

@@ -70,12 +72,14 @@ void testCustomWaitStrategy() {
7072
"custom-strategy",
7173
String.class,
7274
(state, stepCtx) -> {
75+
if (state == null) {
76+
state = "pending";
77+
}
7378
if ("pending".equals(state)) {
7479
return WaitForConditionResult.continuePolling("processing");
7580
}
7681
return WaitForConditionResult.stopPolling("done");
7782
},
78-
"pending",
7983
config);
8084
});
8185

@@ -99,7 +103,6 @@ void testMaxAttemptsExceeded() {
99103
"max-attempts",
100104
String.class,
101105
(state, stepCtx) -> WaitForConditionResult.continuePolling(state),
102-
"initial",
103106
config);
104107
});
105108

@@ -124,7 +127,6 @@ void testCheckFunctionError() {
124127
(state, stepCtx) -> {
125128
throw new IllegalStateException("Check function failed");
126129
},
127-
"initial",
128130
config);
129131
});
130132

@@ -153,13 +155,15 @@ void testReplayAcrossInvocations() {
153155
"replay-poll",
154156
Integer.class,
155157
(state, stepCtx) -> {
158+
if (state == null) {
159+
state = 0;
160+
}
156161
checkCount.incrementAndGet();
157162
var next = state + 1;
158163
return next >= 2
159164
? WaitForConditionResult.stopPolling(next)
160165
: WaitForConditionResult.continuePolling(next);
161166
},
162-
0,
163167
config);
164168

165169
return result.toString();
@@ -197,12 +201,14 @@ void propertyStopPollingCompletesWithState() {
197201
"stop-polling-prop",
198202
Integer.class,
199203
(state, stepCtx) -> {
204+
if (state == null) {
205+
state = 0;
206+
}
200207
var next = state + 1;
201208
return next >= target
202209
? WaitForConditionResult.stopPolling(next)
203210
: WaitForConditionResult.continuePolling(next);
204211
},
205-
0,
206212
config);
207213
});
208214

@@ -238,12 +244,14 @@ void propertyWaitStrategyReceivesCorrectStateAndAttempt() {
238244
"state-attempt-prop",
239245
Integer.class,
240246
(state, stepCtx) -> {
247+
if (state == null) {
248+
state = 0;
249+
}
241250
var next = state + 1;
242251
return next >= totalChecks
243252
? WaitForConditionResult.stopPolling(next)
244253
: WaitForConditionResult.continuePolling(next);
245254
},
246-
0,
247255
config);
248256
});
249257

‎sdk/src/main/java/software/amazon/lambda/durable/DurableContext.java‎

Lines changed: 8 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -628,19 +628,14 @@ <T> DurableFuture<T> waitForCallbackAsync(
628628
* @param name the unique operation name within this context
629629
* @param resultType the result class for deserialization
630630
* @param checkFunc the function that evaluates the condition and returns a {@link WaitForConditionResult}
631-
* @param initialState the initial state passed to the first check invocation
632631
* @return the final state value when the condition is met
633632
*/
634633
default <T> T waitForCondition(
635-
String name,
636-
Class<T> resultType,
637-
BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc,
638-
T initialState) {
634+
String name, Class<T> resultType, BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc) {
639635
return waitForConditionAsync(
640636
name,
641637
TypeToken.get(resultType),
642638
checkFunc,
643-
initialState,
644639
WaitForConditionConfig.<T>builder().build())
645640
.get();
646641
}
@@ -650,23 +645,17 @@ default <T> T waitForCondition(
650645
String name,
651646
Class<T> resultType,
652647
BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc,
653-
T initialState,
654648
WaitForConditionConfig<T> config) {
655-
return waitForConditionAsync(name, resultType, checkFunc, initialState, config)
656-
.get();
649+
return waitForConditionAsync(name, resultType, checkFunc, config).get();
657650
}
658651

659652
/** Polls a condition function until it signals done, using a {@link TypeToken}, blocking until complete. */
660653
default <T> T waitForCondition(
661-
String name,
662-
TypeToken<T> resultType,
663-
BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc,
664-
T initialState) {
654+
String name, TypeToken<T> resultType, BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc) {
665655
return waitForConditionAsync(
666656
name,
667657
resultType,
668658
checkFunc,
669-
initialState,
670659
WaitForConditionConfig.<T>builder().build())
671660
.get();
672661
}
@@ -679,23 +668,17 @@ default <T> T waitForCondition(
679668
String name,
680669
TypeToken<T> resultType,
681670
BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc,
682-
T initialState,
683671
WaitForConditionConfig<T> config) {
684-
return waitForConditionAsync(name, resultType, checkFunc, initialState, config)
685-
.get();
672+
return waitForConditionAsync(name, resultType, checkFunc, config).get();
686673
}
687674

688675
/** Asynchronously polls a condition function until it signals done. */
689676
default <T> DurableFuture<T> waitForConditionAsync(
690-
String name,
691-
Class<T> resultType,
692-
BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc,
693-
T initialState) {
677+
String name, Class<T> resultType, BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc) {
694678
return waitForConditionAsync(
695679
name,
696680
TypeToken.get(resultType),
697681
checkFunc,
698-
initialState,
699682
WaitForConditionConfig.<T>builder().build());
700683
}
701684

@@ -704,23 +687,15 @@ default <T> DurableFuture<T> waitForConditionAsync(
704687
String name,
705688
Class<T> resultType,
706689
BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc,
707-
T initialState,
708690
WaitForConditionConfig<T> config) {
709-
return waitForConditionAsync(name, TypeToken.get(resultType), checkFunc, initialState, config);
691+
return waitForConditionAsync(name, TypeToken.get(resultType), checkFunc, config);
710692
}
711693

712694
/** Asynchronously polls a condition function until it signals done, using a {@link TypeToken}. */
713695
default <T> DurableFuture<T> waitForConditionAsync(
714-
String name,
715-
TypeToken<T> resultType,
716-
BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc,
717-
T initialState) {
696+
String name, TypeToken<T> resultType, BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc) {
718697
return waitForConditionAsync(
719-
name,
720-
resultType,
721-
checkFunc,
722-
initialState,
723-
WaitForConditionConfig.<T>builder().build());
698+
name, resultType, checkFunc, WaitForConditionConfig.<T>builder().build());
724699
}
725700

726701
/**
@@ -734,15 +709,13 @@ default <T> DurableFuture<T> waitForConditionAsync(
734709
* @param name the unique operation name within this context
735710
* @param resultType the type token for deserialization of generic types
736711
* @param checkFunc the function that evaluates the condition and returns a {@link WaitForConditionResult}
737-
* @param initialState the initial state passed to the first check invocation
738712
* @param config the waitForCondition configuration (wait strategy, custom SerDes)
739713
* @return a future representing the final state value
740714
*/
741715
<T> DurableFuture<T> waitForConditionAsync(
742716
String name,
743717
TypeToken<T> resultType,
744718
BiFunction<T, StepContext, WaitForConditionResult<T>> checkFunc,
745-
T initialState,
746719
WaitForConditionConfig<T> config);
747720

748721
/**

0 commit comments

Comments
 (0)