WIP

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

Author: mcruzdev

impl/grpc/src/main/java/io/serverlessworkflow/impl/executors/grpc/FileDescriptorReader.java

@@ -15,24 +15,20 @@
  */
 package io.serverlessworkflow.impl.executors.grpc;
 
-import com.github.os72.protocjar.Protoc;
 import com.google.protobuf.DescriptorProtos;
 import io.serverlessworkflow.impl.resources.ExternalResourceHandler;
+import io.serverlessworkflow.impl.scripts.ScriptUtils;
+
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Optional;
-import java.util.stream.Stream;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class FileDescriptorReader {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(FileDescriptorReader.class);
-
-  public FileDescriptorContext readDescriptor(ExternalResourceHandler externalResourceHandler) {
+    public FileDescriptorContext readDescriptor(ExternalResourceHandler externalResourceHandler) {
     Path grpcDir =
         tryCreateTempGrpcDir()
             .orElseThrow(
@@ -93,34 +89,16 @@ private static void generateFileDescriptor(Path grpcDir, Path protoFile, Path de
         };
 
     try {
-      int status = Protoc.runProtoc(protocArgs);
+
+      ProcessBuilder processBuilder = new ProcessBuilder(protocArgs);
+      int status = ScriptUtils.uncheckedStart(processBuilder).waitFor();
       if (status != 0) {
         throw new RuntimeException(
             "Unable to generate file descriptor, 'protoc' execution failed with status " + status);
       }
-    } catch (IOException e) {
-      throw new UncheckedIOException("Unable to generate file descriptor", e);
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
       throw new RuntimeException("Unable to generate file descriptor", e);
-    } finally {
-      cleanUp(grpcDir);
-    }
-  }
-
-  private static void cleanUp(Path grpcDir) {
-    try (Stream<Path> pathStream = Files.walk(grpcDir)) {
-      pathStream
-          .map(Path::toFile)
-          .forEach(
-              file -> {
-                boolean deleted = file.delete();
-                if (!deleted) {
-                  LOGGER.warn("Unable to deleted the file: {}", file.getName());
-                }
-              });
-    } catch (IOException ignored) {
-      LOGGER.warn("Unable to clean temporary directory: {}", grpcDir);
     }
   }
 }

impl/grpc/src/main/java/io/serverlessworkflow/impl/executors/grpc/GrpcExecutorBuilder.java

@@ -150,53 +150,17 @@ public void init(CallGRPC task, WorkflowDefinition definition, WorkflowMutablePo
                         .build(),
                     CallOptions.DEFAULT.withWaitForReady());
 
-            Map<String, Object> parameters = Map.of();
-
-            if (methodType == MethodDescriptor.MethodType.CLIENT_STREAMING) {
-              JsonNode jsonNode =
-                  ProtobufMessageUtils.asyncStreamingCall(
-                      parameters,
-                      methodDescriptor,
-                      responseObserver ->
-                          ClientCalls.asyncClientStreamingCall(call, responseObserver),
-                      nodes -> nodes.isEmpty() ? NullNode.instance : nodes.get(0));
-              return workflowContext.definition().application().modelFactory().fromAny(jsonNode);
-
-            } else if (methodType == MethodDescriptor.MethodType.BIDI_STREAMING) {
-              return workflowContext
-                  .definition()
-                  .application()
-                  .modelFactory()
-                  .fromAny(
-                      ProtobufMessageUtils.asyncStreamingCall(
-                          parameters,
-                          methodDescriptor,
-                          responseObserver ->
-                              ClientCalls.asyncBidiStreamingCall(call, responseObserver),
-                          v -> {
-                            Collection<JsonNode> nodes = v;
-                            List<JsonNode> list = new ArrayList<>(nodes);
-                            return JsonUtils.fromValue(list);
-                          }));
-            } else if (methodType == MethodDescriptor.MethodType.SERVER_STREAMING) {
-              Message.Builder builder =
-                  ProtobufMessageUtils.buildMessage(methodDescriptor, parameters);
-              List<JsonNode> nodes = new ArrayList<>();
-              ClientCalls.blockingServerStreamingCall(call, builder.build())
-                  .forEachRemaining(message -> nodes.add(ProtobufMessageUtils.convert(message)));
-              return workflowContext.definition().application().modelFactory().fromAny(nodes);
-            } else {
-
-              Message.Builder builder =
-                  ProtobufMessageUtils.buildMessage(methodDescriptor, parameters);
-
-              Message message = ClientCalls.blockingUnaryCall(call, builder.build());
-              return workflowContext
-                  .definition()
-                  .application()
-                  .modelFactory()
-                  .fromAny(ProtobufMessageUtils.convert(message));
-            }
+            return switch (methodType) {
+              case CLIENT_STREAMING ->
+                  handleClientStreaming(workflowContext, arguments, methodDescriptor, call);
+              case BIDI_STREAMING ->
+                  handleBidiStreaming(workflowContext, arguments, methodDescriptor, call);
+              case SERVER_STREAMING ->
+                  handleServerStreaming(workflowContext, methodDescriptor, arguments, call);
+              case UNARY, UNKNOWN ->
+                  handleBlockingUnary(workflowContext, methodDescriptor, arguments, call);
+            };
+
           } catch (Descriptors.DescriptorValidationException
               | InvalidProtocolBufferException
               | JsonProcessingException e) {
@@ -205,6 +169,70 @@ public void init(CallGRPC task, WorkflowDefinition definition, WorkflowMutablePo
         };
   }
 
+  private static WorkflowModel handleClientStreaming(
+      WorkflowContext workflowContext,
+      Map<String, Object> parameters,
+      Descriptors.MethodDescriptor methodDescriptor,
+      ClientCall<Message, Message> call) {
+    JsonNode jsonNode =
+        ProtobufMessageUtils.asyncStreamingCall(
+            parameters,
+            methodDescriptor,
+            responseObserver -> ClientCalls.asyncClientStreamingCall(call, responseObserver),
+            nodes -> nodes.isEmpty() ? NullNode.instance : nodes.get(0));
+    return workflowContext.definition().application().modelFactory().fromAny(jsonNode);
+  }
+
+  private static WorkflowModel handleServerStreaming(
+      WorkflowContext workflowContext,
+      Descriptors.MethodDescriptor methodDescriptor,
+      Map<String, Object> parameters,
+      ClientCall<Message, Message> call)
+      throws InvalidProtocolBufferException, JsonProcessingException {
+    Message.Builder builder = ProtobufMessageUtils.buildMessage(methodDescriptor, parameters);
+    List<JsonNode> nodes = new ArrayList<>();
+    ClientCalls.blockingServerStreamingCall(call, builder.build())
+        .forEachRemaining(message -> nodes.add(ProtobufMessageUtils.convert(message)));
+    return workflowContext.definition().application().modelFactory().fromAny(nodes);
+  }
+
+  private static WorkflowModel handleBlockingUnary(
+      WorkflowContext workflowContext,
+      Descriptors.MethodDescriptor methodDescriptor,
+      Map<String, Object> parameters,
+      ClientCall<Message, Message> call)
+      throws InvalidProtocolBufferException, JsonProcessingException {
+    Message.Builder builder = ProtobufMessageUtils.buildMessage(methodDescriptor, parameters);
+
+    Message message = ClientCalls.blockingUnaryCall(call, builder.build());
+    return workflowContext
+        .definition()
+        .application()
+        .modelFactory()
+        .fromAny(ProtobufMessageUtils.convert(message));
+  }
+
+  private static WorkflowModel handleBidiStreaming(
+      WorkflowContext workflowContext,
+      Map<String, Object> parameters,
+      Descriptors.MethodDescriptor methodDescriptor,
+      ClientCall<Message, Message> call) {
+    return workflowContext
+        .definition()
+        .application()
+        .modelFactory()
+        .fromAny(
+            ProtobufMessageUtils.asyncStreamingCall(
+                parameters,
+                methodDescriptor,
+                responseObserver -> ClientCalls.asyncBidiStreamingCall(call, responseObserver),
+                v -> {
+                  Collection<JsonNode> nodes = v;
+                  List<JsonNode> list = new ArrayList<>(nodes);
+                  return JsonUtils.fromValue(list);
+                }));
+  }
+
   @Override
   public CallableTask build() {
     return new GrpcExecutor(