Sdk new changes

Author: siri-varma

dapr-spring/dapr-spring-workflows/src/main/java/io/dapr/spring/workflows/config/DaprWorkflowsConfiguration.java

@@ -46,10 +46,9 @@ private void registerWorkflowsAndActivities(ApplicationContext applicationContex
       workflowRuntimeBuilder.registerActivity(activity);
     }
 
-    try (WorkflowRuntime runtime = workflowRuntimeBuilder.build()) {
-      LOGGER.info("Starting workflow runtime ... ");
-      runtime.start(false);
-    }
+    WorkflowRuntime runtime = workflowRuntimeBuilder.build();
+    LOGGER.info("Starting workflow runtime ... ");
+    runtime.start(false);
   }
 
   @Override

examples/src/main/java/io/dapr/examples/workflows/chain/DemoChainWorker.java

@@ -29,9 +29,8 @@ public static void main(String[] args) throws Exception {
     builder.registerActivity(ToUpperCaseActivity.class);
 
     // Build and then start the workflow runtime pulling and executing tasks
-    try (WorkflowRuntime runtime = builder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start();
-    }
+    WorkflowRuntime runtime = builder.build();
+    System.out.println("Start workflow runtime");
+    runtime.start();
   }
 }

examples/src/main/java/io/dapr/examples/workflows/childworkflow/DemoChildWorkflowWorker.java

@@ -31,9 +31,7 @@ public static void main(String[] args) throws Exception {
     builder.registerActivity(ReverseActivity.class);
 
     // Build and then start the workflow runtime pulling and executing tasks
-    try (WorkflowRuntime runtime = builder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start();
-    }
+    WorkflowRuntime runtime = builder.build();
+    System.out.println("Start workflow runtime");
   }
 }

examples/src/main/java/io/dapr/examples/workflows/continueasnew/DemoContinueAsNewWorker.java

@@ -16,6 +16,9 @@
 import io.dapr.workflows.runtime.WorkflowRuntime;
 import io.dapr.workflows.runtime.WorkflowRuntimeBuilder;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 public class DemoContinueAsNewWorker {
   /**
    * The main method of this app.
@@ -25,13 +28,14 @@ public class DemoContinueAsNewWorker {
    */
   public static void main(String[] args) throws Exception {
     // 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);
 
     // Build and then start the workflow runtime pulling and executing tasks
-    try (WorkflowRuntime runtime = builder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start();
-    }
+    WorkflowRuntime runtime = builder.build()
+    System.out.println("Start workflow runtime");
+    runtime.start();
   }
 }

examples/src/main/java/io/dapr/examples/workflows/externalevent/DemoExternalEventWorker.java

@@ -30,9 +30,8 @@ public static void main(String[] args) throws Exception {
     builder.registerActivity(DenyActivity.class);
 
     // Build and then start the workflow runtime pulling and executing tasks
-    try (WorkflowRuntime runtime = builder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start();
-    }
+    WorkflowRuntime runtime = builder.build();
+    System.out.println("Start workflow runtime");
+    runtime.start();
   }
 }

examples/src/main/java/io/dapr/examples/workflows/faninout/DemoFanInOutWorker.java

@@ -29,9 +29,8 @@ public static void main(String[] args) throws Exception {
     builder.registerActivity(CountWordsActivity.class);
 
     // Build and then start the workflow runtime pulling and executing tasks
-    try (WorkflowRuntime runtime = builder.build()) {
-      System.out.println("Start workflow runtime");
-      runtime.start();
-    }
+    WorkflowRuntime runtime = builder.build();
+    System.out.println("Start workflow runtime");
+    runtime.start(false);
   }
 }

examples/src/main/java/io/dapr/examples/workflows/faninout/DemoFanInOutWorkflow.java

@@ -37,8 +37,13 @@ public WorkflowStub create() {
           .map(input -> ctx.callActivity(CountWordsActivity.class.getName(), input.toString(), Integer.class))
           .collect(Collectors.toList());
 
+      List<Task<Integer>> tasks2 = inputs.stream()
+              .map(input -> ctx.callActivity(CountWordsActivity2.class.getName(), input.toString(), Integer.class))
+              .collect(Collectors.toList());
+
       // Fan-in to get the total word count from all the individual activity results.
       List<Integer> allWordCountResults = ctx.allOf(tasks).await();
+      List<Integer> allWordCountResults2 = ctx.allOf(tasks2).await();
       int totalWordCount = allWordCountResults.stream().mapToInt(Integer::intValue).sum();
 
       ctx.getLogger().info("Workflow finished with result: " + totalWordCount);

sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntime.java

@@ -18,7 +18,7 @@
 /**
  * Contains methods to register workflows and activities.
  */
-public class WorkflowRuntime implements AutoCloseable {
+public class WorkflowRuntime {
 
   private DurableTaskGrpcWorker worker;
 
@@ -46,15 +46,4 @@ public void start(boolean block) {
       this.worker.start();
     }
   }
-
-  /**
-   * {@inheritDoc}
-   */
-  @Override
-  public void close() {
-    if (this.worker != null) {
-      this.worker.close();
-      this.worker = null;
-    }
-  }
 }

sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilder.java

@@ -26,6 +26,8 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 public class WorkflowRuntimeBuilder {
   private static final ClientInterceptor WORKFLOW_INTERCEPTOR = new ApiTokenClientInterceptor();
@@ -59,7 +61,9 @@ public WorkflowRuntimeBuilder(Logger logger) {
 
   private WorkflowRuntimeBuilder(Properties properties, Logger logger) {
     ManagedChannel managedChannel = NetworkUtils.buildGrpcManagedChannel(properties, WORKFLOW_INTERCEPTOR);
-    this.builder = new DurableTaskGrpcWorkerBuilder().grpcChannel(managedChannel);
+    this.builder = new DurableTaskGrpcWorkerBuilder()
+            .withExecutorService(Executors.newCachedThreadPool())
+            .grpcChannel(managedChannel);
     this.logger = logger;
   }
 
@@ -84,6 +88,17 @@ public WorkflowRuntime build() {
     return instance;
   }
 
+  /**
+   * Register Executor Service to use with workflow.
+   *
+   * @param executorService to be used.
+   * @return {@link WorkflowRuntimeBuilder}.
+   */
+  public WorkflowRuntimeBuilder withExecutorService(ExecutorService executorService) {
+    this.builder.withExecutorService(executorService);
+    return this;
+  }
+
   /**
    * Registers a Workflow object.
    *

sdk-workflows/src/test/java/io/dapr/workflows/runtime/WorkflowRuntimeBuilderTest.java

@@ -67,7 +67,8 @@ public void registerValidWorkflowActivityInstance() {
   @Test
   public void buildTest() {
     assertDoesNotThrow(() -> {
-      try (WorkflowRuntime runtime = new WorkflowRuntimeBuilder().build()) {
+      try {
+        WorkflowRuntime runtime = new WorkflowRuntimeBuilder().build();
         System.out.println("WorkflowRuntime created");
       } catch (Exception e) {
         throw new RuntimeException(e);
@@ -88,13 +89,11 @@ public void loggingOutputTest() {
 
     WorkflowRuntimeBuilder workflowRuntimeBuilder = new WorkflowRuntimeBuilder();
 
-    try (WorkflowRuntime runtime = workflowRuntimeBuilder.build()) {
-      verify(testLogger, times(1))
-          .info(eq("Registered Workflow: {}"), eq("TestWorkflow"));
+    WorkflowRuntime runtime = workflowRuntimeBuilder.build();
+    verify(testLogger, times(1))
+        .info(eq("Registered Workflow: {}"), eq("TestWorkflow"));
 
-      verify(testLogger, times(1))
-          .info(eq("Registered Activity: {}"), eq("TestActivity"));
-    }
+    verify(testLogger, times(1))
+        .info(eq("Registered Activity: {}"), eq("TestActivity"));
   }
-
 }