Make the executor service in ParallelFlow configurable

Resolves #12

Author: fmbenhassine

src/main/java/org/jeasy/flows/workflow/ParallelFlow.java

@@ -30,9 +30,15 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.ExecutorService;
 
 /**
- * A parallel flow executes a set of work units in parallel.
+ * A parallel flow executes a set of work units in parallel. A {@link ParallelFlow}
+ * requires a {@link ExecutorService} to run work units in parallel using multiple
+ * threads.
+ * 
+ * <strong>It is the responsibility of the caller to manage the lifecycle of the
+ * executor service.</strong>
  *
  * The status of a parallel flow execution is defined as:
  *
@@ -68,14 +74,27 @@ public static class Builder {
 
         private String name;
         private List<Work> works;
+        private ExecutorService executorService;
 
-        private Builder() {
+        private Builder(ExecutorService executorService) {
             this.name = UUID.randomUUID().toString();
             this.works = new ArrayList<>();
+            this.executorService = executorService;
         }
 
-        public static ParallelFlow.Builder aNewParallelFlow() {
-            return new ParallelFlow.Builder();
+        /**
+         *  Create a new {@link ParallelFlow} builder. A {@link ParallelFlow}
+         *  requires a {@link ExecutorService} to run work units in parallel
+         *  using multiple threads.
+         *  
+         *  <strong>It is the responsibility of the caller to manage the lifecycle
+         *  of the executor service.</strong>
+         *  
+         * @param executorService to use to run work units in parallel
+         * @return a new {@link ParallelFlow} builder
+         */
+        public static ParallelFlow.Builder aNewParallelFlow(ExecutorService executorService) {
+            return new ParallelFlow.Builder(executorService);
         }
 
         public ParallelFlow.Builder named(String name) {
@@ -89,7 +108,7 @@ public ParallelFlow.Builder execute(Work... works) {
         }
 
         public ParallelFlow build() {
-            return new ParallelFlow(name, works, new ParallelFlowExecutor());
+            return new ParallelFlow(name, works, new ParallelFlowExecutor(executorService));
         }
     }
 }

src/main/java/org/jeasy/flows/workflow/ParallelFlowExecutor.java

@@ -32,7 +32,6 @@
 import java.util.Map;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -41,26 +40,14 @@ class ParallelFlowExecutor {
 
     private static final Logger LOGGER = Logger.getLogger(ParallelFlowExecutor.class.getName());
 
-    /*
-     * TODO Making the executor configurable requires to answer the following questions first:
-     *
-     * 1. If the user provides a custom executor, when should it be shutdown? -> Could be documented so the user shuts it down himself
-     * 2. If the user provides a custom executor which is shared by multiple parallel flow, shutting it down here (as currently done) may impact other flows
-     * 3. If it is decided to shut down the executor at the end of the parallel flow, the parallel flow could not be re-run (in a repeat flow for example) since the executor will be in an illegal state
-     */
     private ExecutorService workExecutor;
 
-    ParallelFlowExecutor() {
-        this.workExecutor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());;
+    ParallelFlowExecutor(ExecutorService workExecutor) {
+        this.workExecutor = workExecutor;
     }
 
     List<WorkReport> executeInParallel(List<Work> works) {
-        // re-init in case it has been shut down in a previous run (See question 3)
-        if(workExecutor.isShutdown()) {
-            workExecutor = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());
-        }
-
-        // submit works to be executed in parallel
+        // submit work units to be executed in parallel
         Map<Work, Future<WorkReport>> reportFutures = new HashMap<>();
         for (Work work : works) {
             Future<WorkReport> reportFuture = workExecutor.submit(work);
@@ -69,7 +56,7 @@ List<WorkReport> executeInParallel(List<Work> works) {
 
         // poll for work completion
         int finishedWorks = works.size();
-        // FIXME polling futures for completion, not sure this is the best way to run callables in parallel and wait them for completion (use CompletionService??)
+        // FIXME polling futures for completion, not sure this is the best way to run callables in parallel and wait for them to complete (use CompletionService??)
         while (finishedWorks > 0) {
             for (Future<WorkReport> future : reportFutures.values()) {
                 if (future != null && future.isDone()) {
@@ -88,7 +75,6 @@ List<WorkReport> executeInParallel(List<Work> works) {
             }
         }
 
-        workExecutor.shutdown(); // because if not, the workflow engine may run forever.. (See question 2).
         return workReports;
     }
 }

src/test/java/org/jeasy/flows/engine/WorkFlowEngineImplTest.java

@@ -23,6 +23,9 @@
  */
 package org.jeasy.flows.engine;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 import org.jeasy.flows.work.DefaultWorkReport;
 import org.jeasy.flows.work.Work;
 import org.jeasy.flows.work.WorkReport;
@@ -73,7 +76,8 @@ public void composeWorkFlowFromSeparateFlowsAndExecuteIt() {
                 .times(3)
                 .build();
 
-        ParallelFlow parallelFlow = aNewParallelFlow()
+        ExecutorService executorService = Executors.newFixedThreadPool(2);
+        ParallelFlow parallelFlow = aNewParallelFlow(executorService)
                 .named("print 'hello' and 'world' in parallel")
                 .execute(work2, work3)
                 .build();
@@ -91,6 +95,7 @@ public void composeWorkFlowFromSeparateFlowsAndExecuteIt() {
 
         WorkFlowEngine workFlowEngine = aNewWorkFlowEngine().build();
         WorkReport workReport = workFlowEngine.run(sequentialFlow);
+        executorService.shutdown();
         assertThat(workReport.getStatus()).isEqualTo(WorkStatus.COMPLETED);
         System.out.println("workflow report = " + workReport);
     }
@@ -103,14 +108,15 @@ public void defineWorkFlowInlineAndExecuteIt() {
         PrintMessageWork work3 = new PrintMessageWork("world");
         PrintMessageWork work4 = new PrintMessageWork("done");
 
+        ExecutorService executorService = Executors.newFixedThreadPool(2);
         WorkFlow workflow = aNewSequentialFlow()
                 .execute(aNewRepeatFlow()
                             .named("print foo 3 times")
                             .repeat(work1)
                             .times(3)
                             .build())
                 .then(aNewConditionalFlow()
-                        .execute(aNewParallelFlow()
+                        .execute(aNewParallelFlow(executorService)
                                     .named("print 'hello' and 'world' in parallel")
                                     .execute(work2, work3)
                                     .build())
@@ -121,6 +127,7 @@ public void defineWorkFlowInlineAndExecuteIt() {
 
         WorkFlowEngine workFlowEngine = aNewWorkFlowEngine().build();
         WorkReport workReport = workFlowEngine.run(workflow);
+        executorService.shutdown();
         assertThat(workReport.getStatus()).isEqualTo(WorkStatus.COMPLETED);
         System.out.println("workflow report = " + workReport);
     }

src/test/java/org/jeasy/flows/workflow/ParallelFlowExecutorTest.java

@@ -32,19 +32,23 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 public class ParallelFlowExecutorTest {
 
     @Test
     public void call() {
 
         // given
+        ExecutorService executorService = Executors.newFixedThreadPool(2);
         HelloWorldWork work1 = new HelloWorldWork("work1", WorkStatus.COMPLETED);
         HelloWorldWork work2 = new HelloWorldWork("work2", WorkStatus.FAILED);
-        ParallelFlowExecutor parallelFlowExecutor = new ParallelFlowExecutor();
+        ParallelFlowExecutor parallelFlowExecutor = new ParallelFlowExecutor(executorService);
 
         // when
         List<WorkReport> workReports = parallelFlowExecutor.executeInParallel(Arrays.asList(work1, work2));
+        executorService.shutdown();
 
         // then
         Assertions.assertThat(workReports).hasSize(2);