Skip to content

Commit ea8e2cd

Browse files
zhongkechenzhongkechen
authored
clean up Class type in operations (#43)
1 parent d00cbc0 commit ea8e2cd

3 files changed

Lines changed: 22 additions & 47 deletions

File tree

‎sdk/src/main/java/com/amazonaws/lambda/durable/DurableContext.java‎

Lines changed: 1 addition & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -248,18 +248,7 @@ public ExecutionContext getExecutionContext() {
248248
// ========== createCallback methods ==========
249249

250250
public <T> DurableCallbackFuture<T> createCallback(String name, Class<T> resultType, CallbackConfig config) {
251-
var operationId = nextOperationId();
252-
253-
var existing = executionManager.getOperation(operationId);
254-
if (existing != null) {
255-
validateReplay(operationId, OperationType.CALLBACK, name, existing);
256-
}
257-
258-
var operation = new CallbackOperation<>(
259-
operationId, name, resultType, config, executionManager, durableConfig.getSerDes());
260-
operation.execute();
261-
262-
return new DurableCallbackFuture<>(operation.getCallbackId(), operation);
251+
return createCallback(name, TypeToken.get(resultType), config);
263252
}
264253

265254
public <T> DurableCallbackFuture<T> createCallback(String name, Class<T> resultType) {

‎sdk/src/main/java/com/amazonaws/lambda/durable/operation/CallbackOperation.java‎

Lines changed: 1 addition & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@ public class CallbackOperation<T> implements DurableOperation<T> {
2525

2626
private final String operationId;
2727
private final String name;
28-
private final Class<T> resultType;
2928
private final TypeToken<T> resultTypeToken;
3029
private final CallbackConfig config;
3130
private final ExecutionManager executionManager;
@@ -38,34 +37,12 @@ public class CallbackOperation<T> implements DurableOperation<T> {
3837
public CallbackOperation(
3938
String operationId,
4039
String name,
41-
Class<T> resultType,
42-
CallbackConfig config,
43-
ExecutionManager executionManager,
44-
SerDes serDes) {
45-
this(operationId, name, resultType, null, config, executionManager, serDes);
46-
}
47-
48-
public CallbackOperation(
49-
String operationId,
50-
String name,
51-
TypeToken<T> resultTypeToken,
52-
CallbackConfig config,
53-
ExecutionManager executionManager,
54-
SerDes serDes) {
55-
this(operationId, name, null, resultTypeToken, config, executionManager, serDes);
56-
}
57-
58-
private CallbackOperation(
59-
String operationId,
60-
String name,
61-
Class<T> resultType,
6240
TypeToken<T> resultTypeToken,
6341
CallbackConfig config,
6442
ExecutionManager executionManager,
6543
SerDes serDes) {
6644
this.operationId = operationId;
6745
this.name = name;
68-
this.resultType = resultType;
6946
this.resultTypeToken = resultTypeToken;
7047
this.config = config;
7148
this.executionManager = executionManager;
@@ -159,8 +136,7 @@ public T get() {
159136
case SUCCEEDED -> {
160137
var result = op.callbackDetails().result();
161138
try {
162-
var typeToken = resultTypeToken != null ? resultTypeToken : TypeToken.get(resultType);
163-
yield serDes.deserialize(result, typeToken);
139+
yield serDes.deserialize(result, resultTypeToken);
164140
} catch (SerDesException e) {
165141
logger.warn(
166142
"Failed to deserialize callback result for callback ID '{}'. "

‎sdk/src/test/java/com/amazonaws/lambda/durable/operation/CallbackOperationTest.java‎

Lines changed: 20 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,8 @@ void executeCreatesCheckpointAndGetsCallbackId() {
7373
var executionManager = createExecutionManager(List.of(executionOp));
7474
var serDes = new JacksonSerDes();
7575

76-
var operation = new CallbackOperation<>("1", "approval", String.class, null, executionManager, serDes);
76+
var operation =
77+
new CallbackOperation<>("1", "approval", TypeToken.get(String.class), null, executionManager, serDes);
7778
operation.execute();
7879

7980
assertNotNull(operation.getCallbackId());
@@ -93,7 +94,8 @@ void executeWithConfigSetsOptions() {
9394
.heartbeatTimeout(Duration.ofSeconds(30))
9495
.build();
9596

96-
var operation = new CallbackOperation<>("1", "approval", String.class, config, executionManager, serDes);
97+
var operation =
98+
new CallbackOperation<>("1", "approval", TypeToken.get(String.class), config, executionManager, serDes);
9799
operation.execute();
98100

99101
assertNotNull(operation.getCallbackId());
@@ -114,7 +116,8 @@ void replayReturnsExistingCallbackIdWhenSucceeded() {
114116
var executionManager = createExecutionManager(List.of(existingCallback));
115117
var serDes = new JacksonSerDes();
116118

117-
var operation = new CallbackOperation<>("1", "approval", String.class, null, executionManager, serDes);
119+
var operation =
120+
new CallbackOperation<>("1", "approval", TypeToken.get(String.class), null, executionManager, serDes);
118121
operation.execute();
119122

120123
assertEquals("existing-callback-id", operation.getCallbackId());
@@ -135,7 +138,8 @@ void getReturnsDeserializedResultWhenSucceeded() {
135138
var executionManager = createExecutionManager(List.of(existingCallback));
136139
var serDes = new JacksonSerDes();
137140

138-
var operation = new CallbackOperation<>("1", "approval", String.class, null, executionManager, serDes);
141+
var operation =
142+
new CallbackOperation<>("1", "approval", TypeToken.get(String.class), null, executionManager, serDes);
139143
operation.execute();
140144
var result = operation.get();
141145

@@ -160,7 +164,8 @@ void getThrowsCallbackExceptionWhenFailed() {
160164
var executionManager = createExecutionManager(List.of(existingCallback));
161165
var serDes = new JacksonSerDes();
162166

163-
var operation = new CallbackOperation<>("1", "approval", String.class, null, executionManager, serDes);
167+
var operation =
168+
new CallbackOperation<>("1", "approval", TypeToken.get(String.class), null, executionManager, serDes);
164169
operation.execute();
165170

166171
var exception = assertThrows(CallbackFailedException.class, operation::get);
@@ -180,7 +185,8 @@ void getThrowsCallbackTimeoutExceptionWhenTimedOut() {
180185
var executionManager = createExecutionManager(List.of(existingCallback));
181186
var serDes = new JacksonSerDes();
182187

183-
var operation = new CallbackOperation<>("1", "approval", String.class, null, executionManager, serDes);
188+
var operation =
189+
new CallbackOperation<>("1", "approval", TypeToken.get(String.class), null, executionManager, serDes);
184190
operation.execute();
185191

186192
var exception = assertThrows(CallbackTimeoutException.class, operation::get);
@@ -205,7 +211,8 @@ void operationUsesCustomSerDesWhenConfigContainsOne() {
205211
var executionManager = createExecutionManager(List.of(existingCallback));
206212

207213
var config = CallbackConfig.builder().serDes(customSerDes).build();
208-
var operation = new CallbackOperation<>("1", "approval", String.class, config, executionManager, defaultSerDes);
214+
var operation = new CallbackOperation<>(
215+
"1", "approval", TypeToken.get(String.class), config, executionManager, defaultSerDes);
209216
operation.execute();
210217
var result = operation.get();
211218

@@ -230,7 +237,8 @@ void operationUsesDefaultSerDesWhenConfigIsNull() {
230237
.build();
231238
var executionManager = createExecutionManager(List.of(existingCallback));
232239

233-
var operation = new CallbackOperation<>("1", "approval", String.class, null, executionManager, customSerDes);
240+
var operation = new CallbackOperation<>(
241+
"1", "approval", TypeToken.get(String.class), null, executionManager, customSerDes);
234242
operation.execute();
235243
var result = operation.get();
236244

@@ -256,7 +264,8 @@ void operationUsesDefaultSerDesWhenConfigSerDesIsNull() {
256264
var executionManager = createExecutionManager(List.of(existingCallback));
257265

258266
var config = CallbackConfig.builder().serDes(null).build();
259-
var operation = new CallbackOperation<>("1", "approval", String.class, config, executionManager, customSerDes);
267+
var operation = new CallbackOperation<>(
268+
"1", "approval", TypeToken.get(String.class), config, executionManager, customSerDes);
260269
operation.execute();
261270
var result = operation.get();
262271

@@ -281,7 +290,8 @@ void getThrowsSerDesExceptionWithHelpfulMessageWhenDeserializationFails() {
281290
.build();
282291
var executionManager = createExecutionManager(List.of(existingCallback));
283292

284-
var operation = new CallbackOperation<>("1", "approval", String.class, null, executionManager, failingSerDes);
293+
var operation = new CallbackOperation<>(
294+
"1", "approval", TypeToken.get(String.class), null, executionManager, failingSerDes);
285295
operation.execute();
286296

287297
var exception = assertThrows(SerDesException.class, operation::get);

0 commit comments

Comments
 (0)