mirror of https://github.com/dapr/java-sdk.git
Use dapr/durabletask-java (#1336)
* microsoft durabletask-java -> dapr durabletask-java Signed-off-by: Cassandra Coyle <cassie@diagrid.io> * update another ref Signed-off-by: Cassandra Coyle <cassie@diagrid.io> * 1.5.2 release Signed-off-by: Cassandra Coyle <cassie@diagrid.io> * fix import order Signed-off-by: Cassandra Coyle <cassie@diagrid.io> * Sdk new changes Signed-off-by: siri-varma <siri.varma@outlook.com> * Refine workflows Signed-off-by: siri-varma <siri.varma@outlook.com> * add ; Signed-off-by: Cassandra Coyle <cassie@diagrid.io> * rm try Signed-off-by: Cassandra Coyle <cassie@diagrid.io> --------- Signed-off-by: Cassandra Coyle <cassie@diagrid.io> Signed-off-by: siri-varma <siri.varma@outlook.com> Co-authored-by: siri-varma <siri.varma@outlook.com>
This commit is contained in:
parent
ecc94f5b94
commit
7ed4d9184c
|
@ -46,10 +46,9 @@ public class DaprWorkflowsConfiguration implements ApplicationContextAware {
|
||||||
workflowRuntimeBuilder.registerActivity(activity);
|
workflowRuntimeBuilder.registerActivity(activity);
|
||||||
}
|
}
|
||||||
|
|
||||||
try (WorkflowRuntime runtime = workflowRuntimeBuilder.build()) {
|
WorkflowRuntime runtime = workflowRuntimeBuilder.build();
|
||||||
LOGGER.info("Starting workflow runtime ... ");
|
LOGGER.info("Starting workflow runtime ... ");
|
||||||
runtime.start(false);
|
runtime.start(false);
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|
|
@ -13,8 +13,8 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.examples.unittesting;
|
package io.dapr.examples.unittesting;
|
||||||
|
|
||||||
import com.microsoft.durabletask.Task;
|
import io.dapr.durabletask.Task;
|
||||||
import com.microsoft.durabletask.TaskCanceledException;
|
import io.dapr.durabletask.TaskCanceledException;
|
||||||
import io.dapr.workflows.Workflow;
|
import io.dapr.workflows.Workflow;
|
||||||
import io.dapr.workflows.WorkflowContext;
|
import io.dapr.workflows.WorkflowContext;
|
||||||
import io.dapr.workflows.WorkflowStub;
|
import io.dapr.workflows.WorkflowStub;
|
||||||
|
|
|
@ -29,9 +29,8 @@ public class DemoChainWorker {
|
||||||
builder.registerActivity(ToUpperCaseActivity.class);
|
builder.registerActivity(ToUpperCaseActivity.class);
|
||||||
|
|
||||||
// Build and then start the workflow runtime pulling and executing tasks
|
// Build and then start the workflow runtime pulling and executing tasks
|
||||||
try (WorkflowRuntime runtime = builder.build()) {
|
WorkflowRuntime runtime = builder.build();
|
||||||
System.out.println("Start workflow runtime");
|
System.out.println("Start workflow runtime");
|
||||||
runtime.start();
|
runtime.start();
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -31,9 +31,7 @@ public class DemoChildWorkflowWorker {
|
||||||
builder.registerActivity(ReverseActivity.class);
|
builder.registerActivity(ReverseActivity.class);
|
||||||
|
|
||||||
// Build and then start the workflow runtime pulling and executing tasks
|
// Build and then start the workflow runtime pulling and executing tasks
|
||||||
try (WorkflowRuntime runtime = builder.build()) {
|
WorkflowRuntime runtime = builder.build();
|
||||||
System.out.println("Start workflow runtime");
|
System.out.println("Start workflow runtime");
|
||||||
runtime.start();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -16,6 +16,9 @@ package io.dapr.examples.workflows.continueasnew;
|
||||||
import io.dapr.workflows.runtime.WorkflowRuntime;
|
import io.dapr.workflows.runtime.WorkflowRuntime;
|
||||||
import io.dapr.workflows.runtime.WorkflowRuntimeBuilder;
|
import io.dapr.workflows.runtime.WorkflowRuntimeBuilder;
|
||||||
|
|
||||||
|
import java.util.concurrent.ExecutorService;
|
||||||
|
import java.util.concurrent.Executors;
|
||||||
|
|
||||||
public class DemoContinueAsNewWorker {
|
public class DemoContinueAsNewWorker {
|
||||||
/**
|
/**
|
||||||
* The main method of this app.
|
* The main method of this app.
|
||||||
|
@ -25,13 +28,14 @@ public class DemoContinueAsNewWorker {
|
||||||
*/
|
*/
|
||||||
public static void main(String[] args) throws Exception {
|
public static void main(String[] args) throws Exception {
|
||||||
// Register the Workflow with the builder.
|
// Register the Workflow with the builder.
|
||||||
WorkflowRuntimeBuilder builder = new WorkflowRuntimeBuilder().registerWorkflow(DemoContinueAsNewWorkflow.class);
|
WorkflowRuntimeBuilder builder = new WorkflowRuntimeBuilder().
|
||||||
|
registerWorkflow(DemoContinueAsNewWorkflow.class)
|
||||||
|
.withExecutorService(Executors.newFixedThreadPool(3));
|
||||||
builder.registerActivity(CleanUpActivity.class);
|
builder.registerActivity(CleanUpActivity.class);
|
||||||
|
|
||||||
// Build and then start the workflow runtime pulling and executing tasks
|
// Build and then start the workflow runtime pulling and executing tasks
|
||||||
try (WorkflowRuntime runtime = builder.build()) {
|
WorkflowRuntime runtime = builder.build();
|
||||||
System.out.println("Start workflow runtime");
|
System.out.println("Start workflow runtime");
|
||||||
runtime.start();
|
runtime.start();
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -30,9 +30,8 @@ public class DemoExternalEventWorker {
|
||||||
builder.registerActivity(DenyActivity.class);
|
builder.registerActivity(DenyActivity.class);
|
||||||
|
|
||||||
// Build and then start the workflow runtime pulling and executing tasks
|
// Build and then start the workflow runtime pulling and executing tasks
|
||||||
try (WorkflowRuntime runtime = builder.build()) {
|
WorkflowRuntime runtime = builder.build();
|
||||||
System.out.println("Start workflow runtime");
|
System.out.println("Start workflow runtime");
|
||||||
runtime.start();
|
runtime.start();
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -29,9 +29,8 @@ public class DemoFanInOutWorker {
|
||||||
builder.registerActivity(CountWordsActivity.class);
|
builder.registerActivity(CountWordsActivity.class);
|
||||||
|
|
||||||
// Build and then start the workflow runtime pulling and executing tasks
|
// Build and then start the workflow runtime pulling and executing tasks
|
||||||
try (WorkflowRuntime runtime = builder.build()) {
|
WorkflowRuntime runtime = builder.build();
|
||||||
System.out.println("Start workflow runtime");
|
System.out.println("Start workflow runtime");
|
||||||
runtime.start();
|
runtime.start(false);
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -13,7 +13,7 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.examples.workflows.faninout;
|
package io.dapr.examples.workflows.faninout;
|
||||||
|
|
||||||
import com.microsoft.durabletask.Task;
|
import io.dapr.durabletask.Task;
|
||||||
import io.dapr.workflows.Workflow;
|
import io.dapr.workflows.Workflow;
|
||||||
import io.dapr.workflows.WorkflowStub;
|
import io.dapr.workflows.WorkflowStub;
|
||||||
|
|
||||||
|
|
2
pom.xml
2
pom.xml
|
@ -36,7 +36,7 @@
|
||||||
<!--
|
<!--
|
||||||
manually declare durabletask-client's jackson dependencies for workflows sdk
|
manually declare durabletask-client's jackson dependencies for workflows sdk
|
||||||
which conflict with dapr-sdk's jackson dependencies
|
which conflict with dapr-sdk's jackson dependencies
|
||||||
https://github.com/microsoft/durabletask-java/blob/main/client/build.gradle#L16
|
https://github.com/dapr/durabletask-java/blob/main/client/build.gradle#L16
|
||||||
-->
|
-->
|
||||||
<jackson.version>2.16.1</jackson.version>
|
<jackson.version>2.16.1</jackson.version>
|
||||||
<gpg.skip>true</gpg.skip>
|
<gpg.skip>true</gpg.skip>
|
||||||
|
|
|
@ -91,10 +91,9 @@ public class DaprWorkflowsIT {
|
||||||
*/
|
*/
|
||||||
@BeforeEach
|
@BeforeEach
|
||||||
public void init() {
|
public void init() {
|
||||||
try (WorkflowRuntime runtime = workflowRuntimeBuilder.build()) {
|
WorkflowRuntime runtime = workflowRuntimeBuilder.build();
|
||||||
System.out.println("Start workflow runtime");
|
System.out.println("Start workflow runtime");
|
||||||
runtime.start(false);
|
runtime.start(false);
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
|
|
|
@ -45,14 +45,14 @@
|
||||||
<scope>test</scope>
|
<scope>test</scope>
|
||||||
</dependency>
|
</dependency>
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>com.microsoft</groupId>
|
<groupId>io.dapr</groupId>
|
||||||
<artifactId>durabletask-client</artifactId>
|
<artifactId>durabletask-client</artifactId>
|
||||||
<version>1.5.0</version>
|
<version>1.5.2</version>
|
||||||
</dependency>
|
</dependency>
|
||||||
<!--
|
<!--
|
||||||
manually declare durabletask-client's jackson dependencies
|
manually declare durabletask-client's jackson dependencies
|
||||||
which conflict with dapr-sdk's jackson dependencies
|
which conflict with dapr-sdk's jackson dependencies
|
||||||
https://github.com/microsoft/durabletask-java/blob/main/client/build.gradle#L16
|
https://github.com/dapr/durabletask-java/blob/main/client/build.gradle#L16
|
||||||
-->
|
-->
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>com.fasterxml.jackson.core</groupId>
|
<groupId>com.fasterxml.jackson.core</groupId>
|
||||||
|
|
|
@ -13,10 +13,10 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows;
|
package io.dapr.workflows;
|
||||||
|
|
||||||
import com.microsoft.durabletask.CompositeTaskFailedException;
|
import io.dapr.durabletask.CompositeTaskFailedException;
|
||||||
import com.microsoft.durabletask.Task;
|
import io.dapr.durabletask.Task;
|
||||||
import com.microsoft.durabletask.TaskCanceledException;
|
import io.dapr.durabletask.TaskCanceledException;
|
||||||
import com.microsoft.durabletask.TaskFailedException;
|
import io.dapr.durabletask.TaskFailedException;
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
|
|
||||||
import javax.annotation.Nullable;
|
import javax.annotation.Nullable;
|
||||||
|
|
|
@ -13,12 +13,12 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.client;
|
package io.dapr.workflows.client;
|
||||||
|
|
||||||
import com.microsoft.durabletask.DurableTaskClient;
|
|
||||||
import com.microsoft.durabletask.DurableTaskGrpcClientBuilder;
|
|
||||||
import com.microsoft.durabletask.NewOrchestrationInstanceOptions;
|
|
||||||
import com.microsoft.durabletask.OrchestrationMetadata;
|
|
||||||
import com.microsoft.durabletask.PurgeResult;
|
|
||||||
import io.dapr.config.Properties;
|
import io.dapr.config.Properties;
|
||||||
|
import io.dapr.durabletask.DurableTaskClient;
|
||||||
|
import io.dapr.durabletask.DurableTaskGrpcClientBuilder;
|
||||||
|
import io.dapr.durabletask.NewOrchestrationInstanceOptions;
|
||||||
|
import io.dapr.durabletask.OrchestrationMetadata;
|
||||||
|
import io.dapr.durabletask.PurgeResult;
|
||||||
import io.dapr.utils.NetworkUtils;
|
import io.dapr.utils.NetworkUtils;
|
||||||
import io.dapr.workflows.Workflow;
|
import io.dapr.workflows.Workflow;
|
||||||
import io.dapr.workflows.internal.ApiTokenClientInterceptor;
|
import io.dapr.workflows.internal.ApiTokenClientInterceptor;
|
||||||
|
|
|
@ -13,7 +13,7 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.TaskActivityContext;
|
import io.dapr.durabletask.TaskActivityContext;
|
||||||
import io.dapr.workflows.WorkflowActivityContext;
|
import io.dapr.workflows.WorkflowActivityContext;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
@ -13,12 +13,12 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.CompositeTaskFailedException;
|
import io.dapr.durabletask.CompositeTaskFailedException;
|
||||||
import com.microsoft.durabletask.RetryPolicy;
|
import io.dapr.durabletask.RetryPolicy;
|
||||||
import com.microsoft.durabletask.Task;
|
import io.dapr.durabletask.Task;
|
||||||
import com.microsoft.durabletask.TaskCanceledException;
|
import io.dapr.durabletask.TaskCanceledException;
|
||||||
import com.microsoft.durabletask.TaskOptions;
|
import io.dapr.durabletask.TaskOptions;
|
||||||
import com.microsoft.durabletask.TaskOrchestrationContext;
|
import io.dapr.durabletask.TaskOrchestrationContext;
|
||||||
import io.dapr.workflows.WorkflowContext;
|
import io.dapr.workflows.WorkflowContext;
|
||||||
import io.dapr.workflows.WorkflowTaskOptions;
|
import io.dapr.workflows.WorkflowTaskOptions;
|
||||||
import io.dapr.workflows.WorkflowTaskRetryPolicy;
|
import io.dapr.workflows.WorkflowTaskRetryPolicy;
|
||||||
|
|
|
@ -13,7 +13,7 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.FailureDetails;
|
import io.dapr.durabletask.FailureDetails;
|
||||||
import io.dapr.workflows.client.WorkflowFailureDetails;
|
import io.dapr.workflows.client.WorkflowFailureDetails;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
@ -13,9 +13,9 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.FailureDetails;
|
import io.dapr.durabletask.FailureDetails;
|
||||||
import com.microsoft.durabletask.OrchestrationMetadata;
|
import io.dapr.durabletask.OrchestrationMetadata;
|
||||||
import com.microsoft.durabletask.OrchestrationRuntimeStatus;
|
import io.dapr.durabletask.OrchestrationRuntimeStatus;
|
||||||
import io.dapr.workflows.client.WorkflowFailureDetails;
|
import io.dapr.workflows.client.WorkflowFailureDetails;
|
||||||
import io.dapr.workflows.client.WorkflowInstanceStatus;
|
import io.dapr.workflows.client.WorkflowInstanceStatus;
|
||||||
import io.dapr.workflows.client.WorkflowRuntimeStatus;
|
import io.dapr.workflows.client.WorkflowRuntimeStatus;
|
||||||
|
|
|
@ -13,8 +13,8 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.TaskActivity;
|
import io.dapr.durabletask.TaskActivity;
|
||||||
import com.microsoft.durabletask.TaskActivityFactory;
|
import io.dapr.durabletask.TaskActivityFactory;
|
||||||
import io.dapr.workflows.WorkflowActivity;
|
import io.dapr.workflows.WorkflowActivity;
|
||||||
|
|
||||||
import java.lang.reflect.Constructor;
|
import java.lang.reflect.Constructor;
|
||||||
|
|
|
@ -13,8 +13,8 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.TaskActivity;
|
import io.dapr.durabletask.TaskActivity;
|
||||||
import com.microsoft.durabletask.TaskActivityFactory;
|
import io.dapr.durabletask.TaskActivityFactory;
|
||||||
import io.dapr.workflows.WorkflowActivity;
|
import io.dapr.workflows.WorkflowActivity;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
@ -13,8 +13,8 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.TaskOrchestration;
|
import io.dapr.durabletask.TaskOrchestration;
|
||||||
import com.microsoft.durabletask.TaskOrchestrationFactory;
|
import io.dapr.durabletask.TaskOrchestrationFactory;
|
||||||
import io.dapr.workflows.Workflow;
|
import io.dapr.workflows.Workflow;
|
||||||
|
|
||||||
import java.lang.reflect.Constructor;
|
import java.lang.reflect.Constructor;
|
||||||
|
|
|
@ -13,8 +13,8 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.TaskOrchestration;
|
import io.dapr.durabletask.TaskOrchestration;
|
||||||
import com.microsoft.durabletask.TaskOrchestrationFactory;
|
import io.dapr.durabletask.TaskOrchestrationFactory;
|
||||||
import io.dapr.workflows.Workflow;
|
import io.dapr.workflows.Workflow;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
@ -13,17 +13,34 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.DurableTaskGrpcWorker;
|
import io.dapr.durabletask.DurableTaskGrpcWorker;
|
||||||
|
import io.grpc.ManagedChannel;
|
||||||
|
|
||||||
|
import java.util.concurrent.ExecutorService;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Contains methods to register workflows and activities.
|
* Contains methods to register workflows and activities.
|
||||||
*/
|
*/
|
||||||
public class WorkflowRuntime implements AutoCloseable {
|
public class WorkflowRuntime implements AutoCloseable {
|
||||||
|
|
||||||
private DurableTaskGrpcWorker worker;
|
private final DurableTaskGrpcWorker worker;
|
||||||
|
private final ManagedChannel managedChannel;
|
||||||
|
private final ExecutorService executorService;
|
||||||
|
|
||||||
public WorkflowRuntime(DurableTaskGrpcWorker worker) {
|
/**
|
||||||
|
* Constructor.
|
||||||
|
*
|
||||||
|
* @param worker grpcWorker processing activities.
|
||||||
|
* @param managedChannel grpc channel.
|
||||||
|
* @param executorService executor service responsible for running the threads.
|
||||||
|
*/
|
||||||
|
public WorkflowRuntime(DurableTaskGrpcWorker worker,
|
||||||
|
ManagedChannel managedChannel,
|
||||||
|
ExecutorService executorService) {
|
||||||
this.worker = worker;
|
this.worker = worker;
|
||||||
|
this.managedChannel = managedChannel;
|
||||||
|
this.executorService = executorService;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -50,11 +67,31 @@ public class WorkflowRuntime implements AutoCloseable {
|
||||||
/**
|
/**
|
||||||
* {@inheritDoc}
|
* {@inheritDoc}
|
||||||
*/
|
*/
|
||||||
@Override
|
|
||||||
public void close() {
|
public void close() {
|
||||||
if (this.worker != null) {
|
this.shutDownWorkerPool();
|
||||||
this.worker.close();
|
this.closeSideCarChannel();
|
||||||
this.worker = null;
|
}
|
||||||
|
|
||||||
|
private void closeSideCarChannel() {
|
||||||
|
this.managedChannel.shutdown();
|
||||||
|
|
||||||
|
try {
|
||||||
|
if (!this.managedChannel.awaitTermination(60, TimeUnit.SECONDS)) {
|
||||||
|
this.managedChannel.shutdownNow();
|
||||||
|
}
|
||||||
|
} catch (InterruptedException ex) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private void shutDownWorkerPool() {
|
||||||
|
this.executorService.shutdown();
|
||||||
|
try {
|
||||||
|
if (!this.executorService.awaitTermination(60, TimeUnit.SECONDS)) {
|
||||||
|
this.executorService.shutdownNow();
|
||||||
|
}
|
||||||
|
} catch (InterruptedException ex) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
|
@ -13,8 +13,8 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.DurableTaskGrpcWorkerBuilder;
|
|
||||||
import io.dapr.config.Properties;
|
import io.dapr.config.Properties;
|
||||||
|
import io.dapr.durabletask.DurableTaskGrpcWorkerBuilder;
|
||||||
import io.dapr.utils.NetworkUtils;
|
import io.dapr.utils.NetworkUtils;
|
||||||
import io.dapr.workflows.Workflow;
|
import io.dapr.workflows.Workflow;
|
||||||
import io.dapr.workflows.WorkflowActivity;
|
import io.dapr.workflows.WorkflowActivity;
|
||||||
|
@ -26,6 +26,8 @@ import org.slf4j.LoggerFactory;
|
||||||
import java.util.Collections;
|
import java.util.Collections;
|
||||||
import java.util.HashSet;
|
import java.util.HashSet;
|
||||||
import java.util.Set;
|
import java.util.Set;
|
||||||
|
import java.util.concurrent.ExecutorService;
|
||||||
|
import java.util.concurrent.Executors;
|
||||||
|
|
||||||
public class WorkflowRuntimeBuilder {
|
public class WorkflowRuntimeBuilder {
|
||||||
private static final ClientInterceptor WORKFLOW_INTERCEPTOR = new ApiTokenClientInterceptor();
|
private static final ClientInterceptor WORKFLOW_INTERCEPTOR = new ApiTokenClientInterceptor();
|
||||||
|
@ -36,6 +38,8 @@ public class WorkflowRuntimeBuilder {
|
||||||
private final Set<String> activitySet = Collections.synchronizedSet(new HashSet<>());
|
private final Set<String> activitySet = Collections.synchronizedSet(new HashSet<>());
|
||||||
private final Set<String> workflowSet = Collections.synchronizedSet(new HashSet<>());
|
private final Set<String> workflowSet = Collections.synchronizedSet(new HashSet<>());
|
||||||
private final DurableTaskGrpcWorkerBuilder builder;
|
private final DurableTaskGrpcWorkerBuilder builder;
|
||||||
|
private final ManagedChannel managedChannel;
|
||||||
|
private ExecutorService executorService;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Constructs the WorkflowRuntimeBuilder.
|
* Constructs the WorkflowRuntimeBuilder.
|
||||||
|
@ -58,8 +62,8 @@ public class WorkflowRuntimeBuilder {
|
||||||
}
|
}
|
||||||
|
|
||||||
private WorkflowRuntimeBuilder(Properties properties, Logger logger) {
|
private WorkflowRuntimeBuilder(Properties properties, Logger logger) {
|
||||||
ManagedChannel managedChannel = NetworkUtils.buildGrpcManagedChannel(properties, WORKFLOW_INTERCEPTOR);
|
this.managedChannel = NetworkUtils.buildGrpcManagedChannel(properties, WORKFLOW_INTERCEPTOR);
|
||||||
this.builder = new DurableTaskGrpcWorkerBuilder().grpcChannel(managedChannel);
|
this.builder = new DurableTaskGrpcWorkerBuilder().grpcChannel(this.managedChannel);
|
||||||
this.logger = logger;
|
this.logger = logger;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -71,8 +75,11 @@ public class WorkflowRuntimeBuilder {
|
||||||
public WorkflowRuntime build() {
|
public WorkflowRuntime build() {
|
||||||
if (instance == null) {
|
if (instance == null) {
|
||||||
synchronized (WorkflowRuntime.class) {
|
synchronized (WorkflowRuntime.class) {
|
||||||
|
this.executorService = this.executorService == null ? Executors.newCachedThreadPool() : this.executorService;
|
||||||
if (instance == null) {
|
if (instance == null) {
|
||||||
instance = new WorkflowRuntime(this.builder.build());
|
instance = new WorkflowRuntime(
|
||||||
|
this.builder.withExecutorService(this.executorService).build(),
|
||||||
|
this.managedChannel, this.executorService);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -84,6 +91,18 @@ public class WorkflowRuntimeBuilder {
|
||||||
return instance;
|
return instance;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Register Executor Service to use with workflow.
|
||||||
|
*
|
||||||
|
* @param executorService to be used.
|
||||||
|
* @return {@link WorkflowRuntimeBuilder}.
|
||||||
|
*/
|
||||||
|
public WorkflowRuntimeBuilder withExecutorService(ExecutorService executorService) {
|
||||||
|
this.executorService = executorService;
|
||||||
|
this.builder.withExecutorService(executorService);
|
||||||
|
return this;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Registers a Workflow object.
|
* Registers a Workflow object.
|
||||||
*
|
*
|
||||||
|
|
|
@ -13,7 +13,7 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.OrchestrationRuntimeStatus;
|
import io.dapr.durabletask.OrchestrationRuntimeStatus;
|
||||||
import io.dapr.workflows.client.WorkflowRuntimeStatus;
|
import io.dapr.workflows.client.WorkflowRuntimeStatus;
|
||||||
|
|
||||||
public class WorkflowRuntimeStatusConverter {
|
public class WorkflowRuntimeStatusConverter {
|
||||||
|
|
|
@ -13,11 +13,11 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows;
|
package io.dapr.workflows;
|
||||||
|
|
||||||
import com.microsoft.durabletask.CompositeTaskFailedException;
|
import io.dapr.durabletask.CompositeTaskFailedException;
|
||||||
import com.microsoft.durabletask.Task;
|
import io.dapr.durabletask.Task;
|
||||||
import com.microsoft.durabletask.TaskCanceledException;
|
import io.dapr.durabletask.TaskCanceledException;
|
||||||
import com.microsoft.durabletask.TaskOptions;
|
import io.dapr.durabletask.TaskOptions;
|
||||||
import com.microsoft.durabletask.TaskOrchestrationContext;
|
import io.dapr.durabletask.TaskOrchestrationContext;
|
||||||
|
|
||||||
import io.dapr.workflows.runtime.DefaultWorkflowContext;
|
import io.dapr.workflows.runtime.DefaultWorkflowContext;
|
||||||
|
|
||||||
|
|
|
@ -13,10 +13,10 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.client;
|
package io.dapr.workflows.client;
|
||||||
|
|
||||||
import com.microsoft.durabletask.DurableTaskClient;
|
import io.dapr.durabletask.DurableTaskClient;
|
||||||
import com.microsoft.durabletask.NewOrchestrationInstanceOptions;
|
import io.dapr.durabletask.NewOrchestrationInstanceOptions;
|
||||||
import com.microsoft.durabletask.OrchestrationMetadata;
|
import io.dapr.durabletask.OrchestrationMetadata;
|
||||||
import com.microsoft.durabletask.OrchestrationRuntimeStatus;
|
import io.dapr.durabletask.OrchestrationRuntimeStatus;
|
||||||
import io.dapr.workflows.Workflow;
|
import io.dapr.workflows.Workflow;
|
||||||
import io.dapr.workflows.WorkflowContext;
|
import io.dapr.workflows.WorkflowContext;
|
||||||
import io.dapr.workflows.WorkflowStub;
|
import io.dapr.workflows.WorkflowStub;
|
||||||
|
|
|
@ -13,9 +13,9 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.client;
|
package io.dapr.workflows.client;
|
||||||
|
|
||||||
import com.microsoft.durabletask.FailureDetails;
|
import io.dapr.durabletask.FailureDetails;
|
||||||
import com.microsoft.durabletask.OrchestrationMetadata;
|
import io.dapr.durabletask.OrchestrationMetadata;
|
||||||
import com.microsoft.durabletask.OrchestrationRuntimeStatus;
|
import io.dapr.durabletask.OrchestrationRuntimeStatus;
|
||||||
import io.dapr.workflows.runtime.DefaultWorkflowInstanceStatus;
|
import io.dapr.workflows.runtime.DefaultWorkflowInstanceStatus;
|
||||||
import org.junit.jupiter.api.BeforeEach;
|
import org.junit.jupiter.api.BeforeEach;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
|
@ -1,6 +1,6 @@
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.TaskActivityContext;
|
import io.dapr.durabletask.TaskActivityContext;
|
||||||
import io.dapr.workflows.WorkflowActivity;
|
import io.dapr.workflows.WorkflowActivity;
|
||||||
import io.dapr.workflows.WorkflowActivityContext;
|
import io.dapr.workflows.WorkflowActivityContext;
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
|
|
|
@ -1,6 +1,6 @@
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.TaskActivityContext;
|
import io.dapr.durabletask.TaskActivityContext;
|
||||||
import io.dapr.workflows.WorkflowActivity;
|
import io.dapr.workflows.WorkflowActivity;
|
||||||
import io.dapr.workflows.WorkflowActivityContext;
|
import io.dapr.workflows.WorkflowActivityContext;
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
|
|
|
@ -13,7 +13,7 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.TaskOrchestrationContext;
|
import io.dapr.durabletask.TaskOrchestrationContext;
|
||||||
import io.dapr.workflows.Workflow;
|
import io.dapr.workflows.Workflow;
|
||||||
import io.dapr.workflows.WorkflowContext;
|
import io.dapr.workflows.WorkflowContext;
|
||||||
import io.dapr.workflows.WorkflowStub;
|
import io.dapr.workflows.WorkflowStub;
|
||||||
|
|
|
@ -13,7 +13,7 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.TaskOrchestrationContext;
|
import io.dapr.durabletask.TaskOrchestrationContext;
|
||||||
import io.dapr.workflows.Workflow;
|
import io.dapr.workflows.Workflow;
|
||||||
import io.dapr.workflows.WorkflowContext;
|
import io.dapr.workflows.WorkflowContext;
|
||||||
import io.dapr.workflows.WorkflowStub;
|
import io.dapr.workflows.WorkflowStub;
|
||||||
|
|
|
@ -67,7 +67,8 @@ public class WorkflowRuntimeBuilderTest {
|
||||||
@Test
|
@Test
|
||||||
public void buildTest() {
|
public void buildTest() {
|
||||||
assertDoesNotThrow(() -> {
|
assertDoesNotThrow(() -> {
|
||||||
try (WorkflowRuntime runtime = new WorkflowRuntimeBuilder().build()) {
|
try {
|
||||||
|
WorkflowRuntime runtime = new WorkflowRuntimeBuilder().build();
|
||||||
System.out.println("WorkflowRuntime created");
|
System.out.println("WorkflowRuntime created");
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
throw new RuntimeException(e);
|
throw new RuntimeException(e);
|
||||||
|
@ -88,13 +89,11 @@ public class WorkflowRuntimeBuilderTest {
|
||||||
|
|
||||||
WorkflowRuntimeBuilder workflowRuntimeBuilder = new WorkflowRuntimeBuilder();
|
WorkflowRuntimeBuilder workflowRuntimeBuilder = new WorkflowRuntimeBuilder();
|
||||||
|
|
||||||
try (WorkflowRuntime runtime = workflowRuntimeBuilder.build()) {
|
WorkflowRuntime runtime = workflowRuntimeBuilder.build();
|
||||||
verify(testLogger, times(1))
|
verify(testLogger, times(1))
|
||||||
.info(eq("Registered Workflow: {}"), eq("TestWorkflow"));
|
.info(eq("Registered Workflow: {}"), eq("TestWorkflow"));
|
||||||
|
|
||||||
verify(testLogger, times(1))
|
verify(testLogger, times(1))
|
||||||
.info(eq("Registered Activity: {}"), eq("TestActivity"));
|
.info(eq("Registered Activity: {}"), eq("TestActivity"));
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -13,7 +13,7 @@ limitations under the License.
|
||||||
|
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
import com.microsoft.durabletask.OrchestrationRuntimeStatus;
|
import io.dapr.durabletask.OrchestrationRuntimeStatus;
|
||||||
import io.dapr.workflows.client.WorkflowRuntimeStatus;
|
import io.dapr.workflows.client.WorkflowRuntimeStatus;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
|
|
|
@ -14,10 +14,15 @@ limitations under the License.
|
||||||
package io.dapr.workflows.runtime;
|
package io.dapr.workflows.runtime;
|
||||||
|
|
||||||
|
|
||||||
import com.microsoft.durabletask.DurableTaskGrpcWorker;
|
import io.dapr.durabletask.DurableTaskGrpcWorker;
|
||||||
import com.microsoft.durabletask.DurableTaskGrpcWorkerBuilder;
|
import io.dapr.durabletask.DurableTaskGrpcWorkerBuilder;
|
||||||
|
import io.dapr.config.Properties;
|
||||||
|
import io.dapr.utils.NetworkUtils;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
|
|
||||||
|
import java.util.concurrent.ExecutorService;
|
||||||
|
import java.util.concurrent.Executors;
|
||||||
|
|
||||||
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
|
||||||
|
|
||||||
public class WorkflowRuntimeTest {
|
public class WorkflowRuntimeTest {
|
||||||
|
@ -25,15 +30,16 @@ public class WorkflowRuntimeTest {
|
||||||
@Test
|
@Test
|
||||||
public void startTest() {
|
public void startTest() {
|
||||||
DurableTaskGrpcWorker worker = new DurableTaskGrpcWorkerBuilder().build();
|
DurableTaskGrpcWorker worker = new DurableTaskGrpcWorkerBuilder().build();
|
||||||
try (WorkflowRuntime runtime = new WorkflowRuntime(worker)) {
|
WorkflowRuntime runtime = new WorkflowRuntime(worker, NetworkUtils.buildGrpcManagedChannel(new Properties()),
|
||||||
assertDoesNotThrow(() -> runtime.start(false));
|
Executors.newCachedThreadPool());
|
||||||
}
|
assertDoesNotThrow(() -> runtime.start(false));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void closeWithoutStarting() {
|
public void closeWithoutStarting() {
|
||||||
DurableTaskGrpcWorker worker = new DurableTaskGrpcWorkerBuilder().build();
|
DurableTaskGrpcWorker worker = new DurableTaskGrpcWorkerBuilder().build();
|
||||||
try (WorkflowRuntime runtime = new WorkflowRuntime(worker)) {
|
try (WorkflowRuntime runtime = new WorkflowRuntime(worker, NetworkUtils.buildGrpcManagedChannel(new Properties()),
|
||||||
|
Executors.newCachedThreadPool())) {
|
||||||
assertDoesNotThrow(runtime::close);
|
assertDoesNotThrow(runtime::close);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue