Improve Process result handling

Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com>

Author: mcruzdev

impl/core/src/main/java/io/serverlessworkflow/impl/executors/RunShellExecutor.java

@@ -22,12 +22,18 @@
 import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowContext;
 import io.serverlessworkflow.impl.WorkflowDefinition;
+import io.serverlessworkflow.impl.WorkflowError;
+import io.serverlessworkflow.impl.WorkflowException;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowModelFactory;
 import io.serverlessworkflow.impl.WorkflowUtils;
 import io.serverlessworkflow.impl.expressions.ExpressionUtils;
+import java.io.BufferedReader;
 import java.io.IOException;
-import java.io.OutputStream;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.StringWriter;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
@@ -39,19 +45,20 @@ public class RunShellExecutor implements RunnableTask<RunShell> {
   @FunctionalInterface
   private interface ShellResultSupplier {
     WorkflowModel apply(
-        TaskContext taskContext, ProcessBuilder processBuilder, WorkflowModel input);
+        TaskContext taskContext, WorkflowModel input, ProcessBuilder processBuilder);
   }
 
+  @FunctionalInterface
   private interface ProcessBuilderSupplier {
-    ProcessBuilder apply(TaskContext taskContext, WorkflowContext workflowContext);
+    ProcessBuilder apply(WorkflowContext workflowContext, TaskContext taskContext);
   }
 
   @Override
   public CompletableFuture<WorkflowModel> apply(
       WorkflowContext workflowContext, TaskContext taskContext, WorkflowModel input) {
-    ProcessBuilder processBuilder = this.processBuilderSupplier.apply(taskContext, workflowContext);
+    ProcessBuilder processBuilder = this.processBuilderSupplier.apply(workflowContext, taskContext);
     WorkflowModel workflowModel =
-        this.shellResultSupplier.apply(taskContext, processBuilder, input);
+        this.shellResultSupplier.apply(taskContext, input, processBuilder);
     return CompletableFuture.completedFuture(workflowModel);
   }
 
@@ -60,8 +67,11 @@ public void init(RunShell taskConfiguration, WorkflowDefinition definition) {
     Shell shell = taskConfiguration.getShell();
     final String shellCommand = shell.getCommand();
 
+    if (shellCommand == null || shellCommand.isBlank()) {
+      throw new IllegalStateException("Missing shell command in RunShell task configuration");
+    }
     this.processBuilderSupplier =
-        (taskContext, workflowContext) -> {
+        (workflowContext, taskContext) -> {
           WorkflowApplication application = definition.application();
 
           String command =
@@ -94,49 +104,50 @@ public void init(RunShell taskConfiguration, WorkflowDefinition definition) {
         };
 
     this.shellResultSupplier =
-        (taskContext, processBuilder, input) -> {
-          if (shellCommand == null || shellCommand.isBlank()) {
-            throw new IllegalStateException(
-                "Missing shell command in RunShell task: " + taskContext.taskName());
-          }
-
+        (taskContext, input, processBuilder) -> {
           try {
             Process process = processBuilder.start();
 
             if (taskConfiguration.isAwait()) {
-              return executeAwaiting(taskConfiguration, definition, process);
+              return waitForResult(taskConfiguration, definition, process);
             } else {
               return input;
             }
 
           } catch (IOException | InterruptedException e) {
-            throw new RuntimeException(e);
+            throw new WorkflowException(WorkflowError.runtime(taskContext, e).build(), e);
           }
         };
   }
 
-  private WorkflowModel executeAwaiting(
+  private WorkflowModel waitForResult(
       RunShell taskConfiguration, WorkflowDefinition definition, Process process)
       throws IOException, InterruptedException {
-    StringBuilder stdout = new StringBuilder();
-    StringBuilder stderr = new StringBuilder();
 
-    process.getInputStream().transferTo(the(stdout));
-    process.getErrorStream().transferTo(the(stderr));
+    CompletableFuture<String> futureStdout =
+        CompletableFuture.supplyAsync(() -> readInputStream(process.getInputStream()));
+    CompletableFuture<String> futureStderr =
+        CompletableFuture.supplyAsync(() -> readInputStream(process.getErrorStream()));
+
     int exitCode = process.waitFor();
 
+    CompletableFuture<Void> allStd = CompletableFuture.allOf(futureStdout, futureStderr);
+
+    allStd.join();
+
+    String stdout = futureStdout.join();
+    String stderr = futureStderr.join();
+
     RunTaskConfiguration.ProcessReturnType returnType = taskConfiguration.getReturn();
 
     WorkflowModelFactory modelFactory = definition.application().modelFactory();
 
     return switch (returnType) {
-      case ALL ->
-          modelFactory.fromAny(
-              new ProcessResult(exitCode, stdout.toString().trim(), stderr.toString().trim()));
+      case ALL -> modelFactory.fromAny(new ProcessResult(exitCode, stdout.trim(), stderr.trim()));
       case NONE -> modelFactory.fromNull();
       case CODE -> modelFactory.from(exitCode);
-      case STDOUT -> modelFactory.from(stdout.toString().trim());
-      case STDERR -> modelFactory.from(stderr.toString().trim());
+      case STDOUT -> modelFactory.from(stdout.trim());
+      case STDERR -> modelFactory.from(stderr.trim());
     };
   }
 
@@ -145,18 +156,18 @@ public boolean accept(Class<? extends RunTaskConfiguration> clazz) {
     return RunShell.class.equals(clazz);
   }
 
-  /**
-   * Helper to create an OutputStream that writes to a StringBuilder
-   *
-   * @param output the {@link StringBuilder} to write to
-   * @return the {@link OutputStream}
-   */
-  private static OutputStream the(StringBuilder output) {
-    return new OutputStream() {
-      @Override
-      public void write(int b) {
-        output.append((char) b);
+  public static String readInputStream(InputStream inputStream) {
+    StringWriter writer = new StringWriter();
+    try (BufferedReader reader =
+        new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))) {
+      char[] buffer = new char[1024];
+      int charsRead;
+      while ((charsRead = reader.read(buffer)) != -1) {
+        writer.write(buffer, 0, charsRead);
       }
-    };
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+    return writer.toString();
   }
 }

impl/test/src/test/java/io/serverlessworkflow/impl/test/RunShellExecutorTest.java

@@ -112,9 +112,7 @@ void testMissingShellCommand() throws IOException {
                     () -> {
                       appl.workflowDefinition(workflow).instance(Map.of()).start().join();
                     })
-                .hasCauseInstanceOf(IllegalStateException.class)
-                .hasMessageContaining(
-                    "Missing shell command in RunShell task: missingShellCommand");
+                .hasMessageContaining("Missing shell command in RunShell task configuration");
           });
     }
   }