Skip to content

Commit 3078c27

Browse files
kaibocaikaibocai
authored
enable resume and suspend (#151)
* enable resume and suspend * update changelog.md * update the end2end tests with polling * clean up end2end tests
1 parent c272b0a commit 3078c27

File tree

7 files changed

+239
-141
lines changed

7 files changed

+239
-141
lines changed

‎CHANGELOG.md‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
## v1.2.0
2-
* Add `thenAccept` and `thenApply` to `Task` interface ([#148](https://github.com/microsoft/durabletask-java/pull/148))
3-
2+
* Add `thenAccept` and `thenApply` to `Task` interface ([#148](https://github.com/microsoft/durabletask-java/pull/148))
3+
* Support Suspend and Resume Client APIs ([#151](https://github.com/microsoft/durabletask-java/pull/151))
4+
* Support restartInstance and pass restartPostUri in HttpManagementPayload ([#108](https://github.com/microsoft/durabletask-java/issues/108))
45

56
## v1.1.1
67

‎azurefunctions/src/main/java/com/microsoft/durabletask/azurefunctions/HttpManagementPayload.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@ public class HttpManagementPayload {
1616
private final String sendEventPostUri;
1717
private final String statusQueryGetUri;
1818
private final String terminatePostUri;
19+
private final String resumePostUri;
20+
private final String suspendPostUri;
1921

2022
/**
2123
* Creates a {@link HttpManagementPayload} to manage orchestration instances
@@ -34,6 +36,8 @@ public HttpManagementPayload(
3436
this.sendEventPostUri = instanceStatusURL + "/raiseEvent/{eventName}?" + requiredQueryStringParameters;
3537
this.statusQueryGetUri = instanceStatusURL + "?" + requiredQueryStringParameters;
3638
this.terminatePostUri = instanceStatusURL + "/terminate?reason={text}&" + requiredQueryStringParameters;
39+
this.resumePostUri = instanceStatusURL + "/resume?reason={text}&" + requiredQueryStringParameters;
40+
this.suspendPostUri = instanceStatusURL + "/suspend?reason={text}&" + requiredQueryStringParameters;
3741
}
3842

3943
/**

‎client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java‎

Lines changed: 29 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -292,33 +292,33 @@ public abstract OrchestrationMetadata waitForInstanceCompletion(
292292
*/
293293
public abstract String restartInstance(String instanceId, boolean restartWithNewInstanceId);
294294

295-
// /**
296-
// * Suspends a running orchestration instance.
297-
// * @param instanceId the ID of the orchestration instance to suspend
298-
// */
299-
// public void suspendInstance (String instanceId) {
300-
// this.suspendInstance(instanceId, null);
301-
// }
302-
//
303-
// /**
304-
// * Resumes a running orchestration instance.
305-
// * @param instanceId the ID of the orchestration instance to resume
306-
// */
307-
// public void resumeInstance(String instanceId) {
308-
// this.resumeInstance(instanceId, null);
309-
// }
310-
//
311-
// /**
312-
// * Suspends a running orchestration instance.
313-
// * @param instanceId the ID of the orchestration instance to suspend
314-
// * @param reason the reason for suspending the orchestration instance
315-
// */
316-
// public abstract void suspendInstance(String instanceId, @Nullable String reason);
317-
//
318-
// /**
319-
// * Resumes a running orchestration instance.
320-
// * @param instanceId the ID of the orchestration instance to resume
321-
// * @param reason the reason for resuming the orchestration instance
322-
// */
323-
// public abstract void resumeInstance(String instanceId, @Nullable String reason);
295+
/**
296+
* Suspends a running orchestration instance.
297+
* @param instanceId the ID of the orchestration instance to suspend
298+
*/
299+
public void suspendInstance (String instanceId) {
300+
this.suspendInstance(instanceId, null);
301+
}
302+
303+
/**
304+
* Resumes a running orchestration instance.
305+
* @param instanceId the ID of the orchestration instance to resume
306+
*/
307+
public void resumeInstance(String instanceId) {
308+
this.resumeInstance(instanceId, null);
309+
}
310+
311+
/**
312+
* Suspends a running orchestration instance.
313+
* @param instanceId the ID of the orchestration instance to suspend
314+
* @param reason the reason for suspending the orchestration instance
315+
*/
316+
public abstract void suspendInstance(String instanceId, @Nullable String reason);
317+
318+
/**
319+
* Resumes a running orchestration instance.
320+
* @param instanceId the ID of the orchestration instance to resume
321+
* @param reason the reason for resuming the orchestration instance
322+
*/
323+
public abstract void resumeInstance(String instanceId, @Nullable String reason);
324324
}

‎client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java‎

Lines changed: 19 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -283,25 +283,25 @@ public PurgeResult purgeInstances(PurgeInstanceCriteria purgeInstanceCriteria) t
283283
}
284284
}
285285

286-
// @Override
287-
// public void suspendInstance(String instanceId, @Nullable String reason) {
288-
// SuspendRequest.Builder suspendRequestBuilder = SuspendRequest.newBuilder();
289-
// suspendRequestBuilder.setInstanceId(instanceId);
290-
// if (reason != null) {
291-
// suspendRequestBuilder.setReason(StringValue.of(reason));
292-
// }
293-
// this.sidecarClient.suspendInstance(suspendRequestBuilder.build());
294-
// }
295-
//
296-
// @Override
297-
// public void resumeInstance(String instanceId, @Nullable String reason) {
298-
// ResumeRequest.Builder resumeRequestBuilder = ResumeRequest.newBuilder();
299-
// resumeRequestBuilder.setInstanceId(instanceId);
300-
// if (reason != null) {
301-
// resumeRequestBuilder.setReason(StringValue.of(reason));
302-
// }
303-
// this.sidecarClient.resumeInstance(resumeRequestBuilder.build());
304-
// }
286+
@Override
287+
public void suspendInstance(String instanceId, @Nullable String reason) {
288+
SuspendRequest.Builder suspendRequestBuilder = SuspendRequest.newBuilder();
289+
suspendRequestBuilder.setInstanceId(instanceId);
290+
if (reason != null) {
291+
suspendRequestBuilder.setReason(StringValue.of(reason));
292+
}
293+
this.sidecarClient.suspendInstance(suspendRequestBuilder.build());
294+
}
295+
296+
@Override
297+
public void resumeInstance(String instanceId, @Nullable String reason) {
298+
ResumeRequest.Builder resumeRequestBuilder = ResumeRequest.newBuilder();
299+
resumeRequestBuilder.setInstanceId(instanceId);
300+
if (reason != null) {
301+
resumeRequestBuilder.setReason(StringValue.of(reason));
302+
}
303+
this.sidecarClient.resumeInstance(resumeRequestBuilder.build());
304+
}
305305

306306
@Override
307307
public String restartInstance(String instanceId, boolean restartWithNewInstanceId) {

‎client/src/test/java/com/microsoft/durabletask/IntegrationTests.java‎

Lines changed: 65 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -457,71 +457,71 @@ void restartOrchestrationThrowsException() {
457457

458458
}
459459

460-
// @Test
461-
// void suspendResumeOrchestration() throws TimeoutException, InterruptedException {
462-
// final String orchestratorName = "suspend";
463-
// final String eventName = "MyEvent";
464-
// final String eventPayload = "testPayload";
465-
// final Duration suspendTimeout = Duration.ofSeconds(5);
466-
//
467-
// DurableTaskGrpcWorker worker = this.createWorkerBuilder()
468-
// .addOrchestrator(orchestratorName, ctx -> {
469-
// String payload = ctx.waitForExternalEvent(eventName, String.class).await();
470-
// ctx.complete(payload);
471-
// })
472-
// .buildAndStart();
473-
//
474-
// DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
475-
// try (worker; client) {
476-
// String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
477-
// client.suspendInstance(instanceId);
478-
// OrchestrationMetadata instance = client.waitForInstanceStart(instanceId, defaultTimeout);
479-
// assertNotNull(instance);
480-
// assertEquals(OrchestrationRuntimeStatus.SUSPENDED, instance.getRuntimeStatus());
481-
//
482-
// client.raiseEvent(instanceId, eventName, eventPayload);
483-
//
484-
// assertThrows(
485-
// TimeoutException.class,
486-
// () -> client.waitForInstanceCompletion(instanceId, suspendTimeout, false),
487-
// "Expected to throw TimeoutException, but it didn't"
488-
// );
489-
//
490-
// String resumeReason = "Resume for testing.";
491-
// client.resumeInstance(instanceId, resumeReason);
492-
// instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
493-
// assertNotNull(instance);
494-
// assertEquals(instanceId, instance.getInstanceId());
495-
// assertEquals(eventPayload, instance.readOutputAs(String.class));
496-
// assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
497-
// }
498-
// }
499-
//
500-
// @Test
501-
// void terminateSuspendOrchestration() throws TimeoutException, InterruptedException {
502-
// final String orchestratorName = "suspendResume";
503-
// final String eventName = "MyEvent";
504-
// final String eventPayload = "testPayload";
505-
//
506-
// DurableTaskGrpcWorker worker = this.createWorkerBuilder()
507-
// .addOrchestrator(orchestratorName, ctx -> {
508-
// String payload = ctx.waitForExternalEvent(eventName, String.class).await();
509-
// ctx.complete(payload);
510-
// })
511-
// .buildAndStart();
512-
//
513-
// DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
514-
// try (worker; client) {
515-
// String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
516-
// String suspendReason = "Suspend for testing.";
517-
// client.suspendInstance(instanceId, suspendReason);
518-
// client.terminate(instanceId, null);
519-
// OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, false);
520-
// assertNotNull(instance);
521-
// assertEquals(instanceId, instance.getInstanceId());
522-
// assertEquals(OrchestrationRuntimeStatus.TERMINATED, instance.getRuntimeStatus());
523-
// }
524-
// }
460+
@Test
461+
void suspendResumeOrchestration() throws TimeoutException, InterruptedException {
462+
final String orchestratorName = "suspend";
463+
final String eventName = "MyEvent";
464+
final String eventPayload = "testPayload";
465+
final Duration suspendTimeout = Duration.ofSeconds(5);
466+
467+
DurableTaskGrpcWorker worker = this.createWorkerBuilder()
468+
.addOrchestrator(orchestratorName, ctx -> {
469+
String payload = ctx.waitForExternalEvent(eventName, String.class).await();
470+
ctx.complete(payload);
471+
})
472+
.buildAndStart();
473+
474+
DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
475+
try (worker; client) {
476+
String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
477+
client.suspendInstance(instanceId);
478+
OrchestrationMetadata instance = client.waitForInstanceStart(instanceId, defaultTimeout);
479+
assertNotNull(instance);
480+
assertEquals(OrchestrationRuntimeStatus.SUSPENDED, instance.getRuntimeStatus());
481+
482+
client.raiseEvent(instanceId, eventName, eventPayload);
483+
484+
assertThrows(
485+
TimeoutException.class,
486+
() -> client.waitForInstanceCompletion(instanceId, suspendTimeout, false),
487+
"Expected to throw TimeoutException, but it didn't"
488+
);
489+
490+
String resumeReason = "Resume for testing.";
491+
client.resumeInstance(instanceId, resumeReason);
492+
instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true);
493+
assertNotNull(instance);
494+
assertEquals(instanceId, instance.getInstanceId());
495+
assertEquals(eventPayload, instance.readOutputAs(String.class));
496+
assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
497+
}
498+
}
499+
500+
@Test
501+
void terminateSuspendOrchestration() throws TimeoutException, InterruptedException {
502+
final String orchestratorName = "suspendResume";
503+
final String eventName = "MyEvent";
504+
final String eventPayload = "testPayload";
505+
506+
DurableTaskGrpcWorker worker = this.createWorkerBuilder()
507+
.addOrchestrator(orchestratorName, ctx -> {
508+
String payload = ctx.waitForExternalEvent(eventName, String.class).await();
509+
ctx.complete(payload);
510+
})
511+
.buildAndStart();
512+
513+
DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
514+
try (worker; client) {
515+
String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
516+
String suspendReason = "Suspend for testing.";
517+
client.suspendInstance(instanceId, suspendReason);
518+
client.terminate(instanceId, null);
519+
OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, false);
520+
assertNotNull(instance);
521+
assertEquals(instanceId, instance.getInstanceId());
522+
assertEquals(OrchestrationRuntimeStatus.TERMINATED, instance.getRuntimeStatus());
523+
}
524+
}
525525

526526
@Test
527527
void activityFanOut() throws IOException, TimeoutException {

‎samples-azure-functions/src/main/java/com/functions/ResumeSuspend.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
package com.functions;
2+
3+
import com.microsoft.azure.functions.ExecutionContext;
4+
import com.microsoft.azure.functions.HttpMethod;
5+
import com.microsoft.azure.functions.HttpRequestMessage;
6+
import com.microsoft.azure.functions.HttpResponseMessage;
7+
import com.microsoft.azure.functions.annotation.AuthorizationLevel;
8+
import com.microsoft.azure.functions.annotation.FunctionName;
9+
import com.microsoft.azure.functions.annotation.HttpTrigger;
10+
import com.microsoft.durabletask.DurableTaskClient;
11+
import com.microsoft.durabletask.TaskOrchestrationContext;
12+
import com.microsoft.durabletask.azurefunctions.DurableClientContext;
13+
import com.microsoft.durabletask.azurefunctions.DurableClientInput;
14+
import com.microsoft.durabletask.azurefunctions.DurableOrchestrationTrigger;
15+
16+
import java.util.Optional;
17+
18+
public class ResumeSuspend {
19+
20+
@FunctionName("StartResumeSuspendOrchestration")
21+
public HttpResponseMessage startResumeSuspendOrchestration(
22+
@HttpTrigger(name = "req", methods = {HttpMethod.GET, HttpMethod.POST}, authLevel = AuthorizationLevel.ANONYMOUS) HttpRequestMessage<Optional<String>> request,
23+
@DurableClientInput(name = "durableContext") DurableClientContext durableContext,
24+
final ExecutionContext context) {
25+
context.getLogger().info("Java HTTP trigger processed a request.");
26+
27+
DurableTaskClient client = durableContext.getClient();
28+
String instanceId = client.scheduleNewOrchestrationInstance("ResumeSuspend");
29+
context.getLogger().info("Created new Java orchestration with instance ID = " + instanceId);
30+
return durableContext.createCheckStatusResponse(request, instanceId);
31+
}
32+
33+
@FunctionName("ResumeSuspend")
34+
public void resumeSuspend(
35+
@DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
36+
ctx.waitForExternalEvent("test").await();
37+
return;
38+
}
39+
}

0 commit comments

Comments
 (0)