[Fix #722] Close executor service if created

Signed-off-by: fjtirado <[email protected]>

Author: fjtirado

impl/core/src/main/java/io/serverlessworkflow/impl/DefaultExecutorServiceFactory.java

@@ -17,23 +17,37 @@
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 public class DefaultExecutorServiceFactory implements ExecutorServiceFactory {
 
-  private static final ExecutorServiceFactory instance = new DefaultExecutorServiceFactory();
+  private ExecutorService service;
+  private Lock serviceLock = new ReentrantLock();
 
   public static ExecutorServiceFactory instance() {
-    return instance;
-  }
-
-  private static class ExecutorServiceHolder {
-    private static ExecutorService instance = Executors.newCachedThreadPool();
+    return new DefaultExecutorServiceFactory();
   }
 
   @Override
   public ExecutorService get() {
-    return ExecutorServiceHolder.instance;
+    try {
+      serviceLock.lock();
+      if (service == null) {
+        service = Executors.newCachedThreadPool();
+      }
+    } finally {
+      serviceLock.unlock();
+    }
+    return service;
   }
 
   private DefaultExecutorServiceFactory() {}
+
+  @Override
+  public void close() {
+    if (service != null) {
+      service.shutdown();
+    }
+  }
 }

impl/core/src/main/java/io/serverlessworkflow/impl/ExecutorServiceFactory.java

@@ -18,5 +18,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.function.Supplier;
 
-@FunctionalInterface
-public interface ExecutorServiceFactory extends Supplier<ExecutorService> {}
+public interface ExecutorServiceFactory extends Supplier<ExecutorService>, AutoCloseable {
+  void close();
+}

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java

@@ -39,11 +39,14 @@
 import java.util.ServiceLoader.Provider;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class WorkflowApplication implements AutoCloseable {
 
+  private static final Logger logger = LoggerFactory.getLogger(WorkflowApplication.class);
+
   private final TaskExecutorFactory taskFactory;
   private final ExpressionFactory exprFactory;
   private final ResourceLoaderFactory resourceLoaderFactory;
@@ -137,7 +140,7 @@ public SchemaValidator getValidator(SchemaInline inline) {
     private SchemaValidatorFactory schemaValidatorFactory;
     private WorkflowPositionFactory positionFactory = () -> new QueueWorkflowPosition();
     private WorkflowIdFactory idFactory = () -> UlidCreator.getMonotonicUlid().toString();
-    private ExecutorServiceFactory executorFactory = () -> Executors.newCachedThreadPool();
+    private ExecutorServiceFactory executorFactory = DefaultExecutorServiceFactory.instance();
     private EventConsumer<?, ?> eventConsumer = InMemoryEvents.get();
     private EventPublisher eventPublisher = InMemoryEvents.get();
     private RuntimeDescriptorFactory descriptorFactory =
@@ -236,12 +239,23 @@ public WorkflowDefinition workflowDefinition(Workflow workflow) {
 
   @Override
   public void close() {
+    safeClose(executorFactory);
+    safeClose(eventPublisher);
+    safeClose(eventConsumer);
     for (WorkflowDefinition definition : definitions.values()) {
-      definition.close();
+      safeClose(definition);
     }
     definitions.clear();
   }
 
+  private void safeClose(AutoCloseable closeable) {
+    try {
+      closeable.close();
+    } catch (Exception ex) {
+      logger.warn("Error closing resource {}", closeable.getClass().getName(), ex);
+    }
+  }
+
   public WorkflowPositionFactory positionFactory() {
     return positionFactory;
   }

impl/core/src/main/java/io/serverlessworkflow/impl/events/EventConsumer.java

@@ -21,7 +21,8 @@
 import java.util.Collection;
 import java.util.function.Consumer;
 
-public interface EventConsumer<T extends EventRegistration, V extends EventRegistrationBuilder> {
+public interface EventConsumer<T extends EventRegistration, V extends EventRegistrationBuilder>
+    extends AutoCloseable {
 
   V listen(EventFilter filter, WorkflowApplication workflowApplication);
 
@@ -30,4 +31,6 @@ public interface EventConsumer<T extends EventRegistration, V extends EventRegis
   T register(V builder, Consumer<CloudEvent> consumer);
 
   void unregister(T register);
+
+  void close();
 }

impl/core/src/main/java/io/serverlessworkflow/impl/events/EventPublisher.java

@@ -18,6 +18,8 @@
 import io.cloudevents.CloudEvent;
 import java.util.concurrent.CompletableFuture;
 
-public interface EventPublisher {
+public interface EventPublisher extends AutoCloseable {
   CompletableFuture<Void> publish(CloudEvent event);
+
+  void close();
 }

impl/core/src/main/java/io/serverlessworkflow/impl/events/InMemoryEvents.java

@@ -16,10 +16,11 @@
 package io.serverlessworkflow.impl.events;
 
 import io.cloudevents.CloudEvent;
-import io.serverlessworkflow.impl.DefaultExecutorServiceFactory;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
@@ -31,6 +32,8 @@ public class InMemoryEvents extends AbstractTypeConsumer implements EventPublish
 
   private static InMemoryEvents instance = new InMemoryEvents();
 
+  private ExecutorService service = Executors.newCachedThreadPool();
+
   private InMemoryEvents() {}
 
   public static InMemoryEvents get() {
@@ -64,7 +67,7 @@ public CompletableFuture<Void> publish(CloudEvent ce) {
             consumer.accept(ce);
           }
         },
-        DefaultExecutorServiceFactory.instance().get());
+        service);
   }
 
   @Override
@@ -76,4 +79,11 @@ protected void registerToAll(Consumer<CloudEvent> consumer) {
   protected void unregisterFromAll() {
     allConsumerRef.set(null);
   }
+
+  @Override
+  public void close() {
+    if (!service.isShutdown()) {
+      service.shutdown();
+    }
+  }
 }