[Fix #744] Improving EventPublisher/EventConsumer

There might be more that one EventPublisher

EventPublishers are either added programatically or through service
loader.

There will be only one consumer. If user want to handler more than one
consuming event broker, he should provide an specific implementation
that deals with them.

The one consumer is added programatically (priortiy) or default to
service loader.

If not eventConsumer was provides, InMemoryEvents is used as event
consumer and publisher.

ChatBotIT and LifeCycleEventTest has been changed accordingly.

Signed-off-by: Francisco Javier Tirado Sarti <[email protected]>

Author: fjtirado

fluent/agentic/src/test/java/io/serverlessworkflow/fluent/agentic/ChatBotIT.java

@@ -28,13 +28,12 @@
 import dev.langchain4j.memory.chat.MessageWindowChatMemory;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.v1.CloudEventBuilder;
-import io.serverlessworkflow.api.types.EventFilter;
-import io.serverlessworkflow.api.types.EventProperties;
 import io.serverlessworkflow.api.types.Workflow;
 import io.serverlessworkflow.impl.WorkflowApplication;
 import io.serverlessworkflow.impl.WorkflowInstance;
 import io.serverlessworkflow.impl.WorkflowModel;
 import io.serverlessworkflow.impl.WorkflowStatus;
+import io.serverlessworkflow.impl.events.InMemoryEvents;
 import java.net.URI;
 import java.time.OffsetDateTime;
 import java.util.Map;
@@ -103,25 +102,15 @@ void chat_bot() {
                         .emit(emit -> emit.event(e -> e.type("org.acme.chatbot.finished"))))
             .build();
 
-    try (WorkflowApplication app = WorkflowApplication.builder().build()) {
-      app.eventConsumer()
-          .register(
-              app.eventConsumer()
-                  .listen(
-                      new EventFilter()
-                          .withWith(new EventProperties().withType("org.acme.chatbot.reply")),
-                      app),
-              ce -> replyEvents.add((CloudEvent) ce));
-
-      app.eventConsumer()
-          .register(
-              app.eventConsumer()
-                  .listen(
-                      new EventFilter()
-                          .withWith(new EventProperties().withType("org.acme.chatbot.finished")),
-                      app),
-              ce -> finishedEvents.add((CloudEvent) ce));
+    InMemoryEvents eventBroker = new InMemoryEvents();
+    eventBroker.register("org.acme.chatbot.reply", ce -> replyEvents.add((CloudEvent) ce));
+    eventBroker.register("org.acme.chatbot.finished", ce -> finishedEvents.add((CloudEvent) ce));
 
+    try (WorkflowApplication app =
+        WorkflowApplication.builder()
+            .withEventConsumer(eventBroker)
+            .withEventPublisher(eventBroker)
+            .build()) {
       final WorkflowInstance waitingInstance =
           app.workflowDefinition(listenWorkflow).instance(Map.of());
       final CompletableFuture<WorkflowModel> runningModel = waitingInstance.start();
@@ -130,12 +119,12 @@ void chat_bot() {
       assertEquals(WorkflowStatus.WAITING, waitingInstance.status());
 
       // Publish the events
-      app.eventPublisher().publish(newMessageEvent("Hello World!"));
+      eventBroker.publish(newMessageEvent("Hello World!"));
       CloudEvent reply = replyEvents.poll(60, TimeUnit.SECONDS);
       assertNotNull(reply);
 
       // Empty message completes the workflow
-      app.eventPublisher().publish(newMessageEvent("", "org.acme.chatbot.finalize"));
+      eventBroker.publish(newMessageEvent("", "org.acme.chatbot.finalize"));
       CloudEvent finished = finishedEvents.poll(60, TimeUnit.SECONDS);
       assertNotNull(finished);
       assertThat(finishedEvents).isEmpty();
@@ -145,6 +134,7 @@ void chat_bot() {
 
     } catch (InterruptedException e) {
       fail(e.getMessage());
+    } finally {
     }
   }
 

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

@@ -32,6 +32,7 @@
 import io.serverlessworkflow.impl.resources.StaticResource;
 import io.serverlessworkflow.impl.schema.SchemaValidator;
 import io.serverlessworkflow.impl.schema.SchemaValidatorFactory;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Map;
@@ -58,7 +59,7 @@ public class WorkflowApplication implements AutoCloseable {
   private final ExecutorServiceFactory executorFactory;
   private final RuntimeDescriptorFactory runtimeDescriptorFactory;
   private final EventConsumer<?, ?> eventConsumer;
-  private final EventPublisher eventPublisher;
+  private final Collection<EventPublisher> eventPublishers;
   private final boolean lifeCycleCEPublishingEnabled;
 
   private WorkflowApplication(Builder builder) {
@@ -73,7 +74,7 @@ private WorkflowApplication(Builder builder) {
     this.listeners = builder.listeners != null ? builder.listeners : Collections.emptySet();
     this.definitions = new ConcurrentHashMap<>();
     this.eventConsumer = builder.eventConsumer;
-    this.eventPublisher = builder.eventPublisher;
+    this.eventPublishers = builder.eventPublishers;
     this.lifeCycleCEPublishingEnabled = builder.lifeCycleCEPublishingEnabled;
   }
 
@@ -101,8 +102,8 @@ public Collection<WorkflowExecutionListener> listeners() {
     return listeners;
   }
 
-  public EventPublisher eventPublisher() {
-    return eventPublisher;
+  public Collection<EventPublisher> eventPublishers() {
+    return eventPublishers;
   }
 
   public WorkflowIdFactory idFactory() {
@@ -144,7 +145,7 @@ public SchemaValidator getValidator(SchemaInline inline) {
     private WorkflowIdFactory idFactory = () -> UlidCreator.getMonotonicUlid().toString();
     private ExecutorServiceFactory executorFactory = new DefaultExecutorServiceFactory();
     private EventConsumer<?, ?> eventConsumer;
-    private EventPublisher eventPublisher;
+    private Collection<EventPublisher> eventPublishers = new ArrayList<>();
     private RuntimeDescriptorFactory descriptorFactory =
         () -> new RuntimeDescriptor("reference impl", "1.0.0_alpha", Collections.emptyMap());
     private boolean lifeCycleCEPublishingEnabled = true;
@@ -201,10 +202,13 @@ public Builder withDescriptorFactory(RuntimeDescriptorFactory factory) {
       return this;
     }
 
-    public Builder withEventHandler(
-        EventPublisher eventPublisher, EventConsumer<?, ?> eventConsumer) {
+    public Builder withEventConsumer(EventConsumer<?, ?> eventConsumer) {
       this.eventConsumer = eventConsumer;
-      this.eventPublisher = eventPublisher;
+      return this;
+    }
+
+    public Builder withEventPublisher(EventPublisher eventPublisher) {
+      this.eventPublishers.add(eventPublisher);
       return this;
     }
 
@@ -227,10 +231,19 @@ public WorkflowApplication build() {
                 .findFirst()
                 .orElseGet(() -> DefaultTaskExecutorFactory.get());
       }
-      if (eventConsumer == null && eventPublisher == null) {
-        InMemoryEvents inMemory = new InMemoryEvents(executorFactory);
-        eventPublisher = inMemory;
-        eventConsumer = inMemory;
+      ServiceLoader.load(EventPublisher.class).forEach(e -> eventPublishers.add(e));
+      if (eventConsumer == null) {
+        eventConsumer =
+            ServiceLoader.load(EventConsumer.class)
+                .findFirst()
+                .orElseGet(
+                    () -> {
+                      InMemoryEvents inMemory = new InMemoryEvents(executorFactory);
+                      if (eventPublishers.isEmpty()) {
+                        eventPublishers.add(inMemory);
+                      }
+                      return inMemory;
+                    });
       }
       return new WorkflowApplication(this);
     }
@@ -250,8 +263,11 @@ public WorkflowDefinition workflowDefinition(Workflow workflow) {
   @Override
   public void close() {
     safeClose(executorFactory);
-    safeClose(eventPublisher);
+    for (EventPublisher eventPublisher : eventPublishers) {
+      safeClose(eventPublisher);
+    }
     safeClose(eventConsumer);
+
     for (WorkflowDefinition definition : definitions.values()) {
       safeClose(definition);
     }

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

@@ -16,6 +16,7 @@
 package io.serverlessworkflow.impl.events;
 
 import io.cloudevents.CloudEvent;
+import io.serverlessworkflow.impl.DefaultExecutorServiceFactory;
 import io.serverlessworkflow.impl.ExecutorServiceFactory;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -29,6 +30,10 @@
  */
 public class InMemoryEvents extends AbstractTypeConsumer implements EventPublisher {
 
+  public InMemoryEvents() {
+    this(new DefaultExecutorServiceFactory());
+  }
+
   public InMemoryEvents(ExecutorServiceFactory serviceFactory) {
     this.serviceFactory = serviceFactory;
   }
@@ -40,7 +45,7 @@ public InMemoryEvents(ExecutorServiceFactory serviceFactory) {
   private AtomicReference<Consumer<CloudEvent>> allConsumerRef = new AtomicReference<>();
 
   @Override
-  protected void register(String topicName, Consumer<CloudEvent> consumer) {
+  public void register(String topicName, Consumer<CloudEvent> consumer) {
     topicMap.put(topicName, consumer);
   }
 
@@ -77,6 +82,7 @@ protected void unregisterFromAll() {
 
   @Override
   public void close() {
+    topicMap.clear();
     serviceFactory.close();
   }
 }

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

@@ -33,10 +33,13 @@
 import io.serverlessworkflow.impl.WorkflowUtils;
 import io.serverlessworkflow.impl.WorkflowValueResolver;
 import io.serverlessworkflow.impl.events.CloudEventUtils;
+import io.serverlessworkflow.impl.events.EventPublisher;
 import io.serverlessworkflow.impl.expressions.ExpressionDescriptor;
 import io.serverlessworkflow.impl.resources.ResourceLoader;
 import java.net.URI;
 import java.time.OffsetDateTime;
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -74,11 +77,15 @@ private EmitExecutor(EmitExecutorBuilder builder) {
   @Override
   protected CompletableFuture<WorkflowModel> internalExecute(
       WorkflowContext workflow, TaskContext taskContext) {
-    return workflow
-        .definition()
-        .application()
-        .eventPublisher()
-        .publish(buildCloudEvent(workflow, taskContext))
+    Collection<EventPublisher> eventPublishers =
+        workflow.definition().application().eventPublishers();
+    CloudEvent ce = buildCloudEvent(workflow, taskContext);
+    Collection<CompletableFuture<Void>> allCompletables = new ArrayList<>();
+    for (EventPublisher eventPublisher : eventPublishers) {
+      allCompletables.add(eventPublisher.publish(ce));
+    }
+    return CompletableFuture.allOf(
+            allCompletables.toArray(new CompletableFuture[allCompletables.size()]))
         .thenApply(v -> taskContext.input());
   }