[incubator-kie-issues#1457] Allow grouping of events

api/kogito-events-core/src/main/java/org/kie/kogito/event/AbstractDataEvent.java

@@ -177,6 +177,15 @@ public abstract class AbstractDataEvent<T> implements DataEvent<T> {
     protected AbstractDataEvent() {
     }
 
+    protected AbstractDataEvent(String type, URI source, T body) {
+        this.specVersion = SpecVersion.parse(SPEC_VERSION);
+        this.id = UUID.randomUUID().toString();
+        this.source = source;
+        this.type = type;
+        this.time = ZonedDateTime.now().toOffsetDateTime();
+        this.data = body;
+    }
+
     protected AbstractDataEvent(String type,
             String source,
             T body,
@@ -201,12 +210,7 @@ protected AbstractDataEvent(String type,
             String subject,
             String dataContentType,
             String dataSchema) {
-        this.specVersion = SpecVersion.parse(SPEC_VERSION);
-        this.id = UUID.randomUUID().toString();
-        this.source = Optional.ofNullable(source).map(URI::create).orElse(null);
-        this.type = type;
-        this.time = ZonedDateTime.now().toOffsetDateTime();
-        this.data = body;
+        this(type, Optional.ofNullable(source).map(URI::create).orElse(null), body);
         setKogitoProcessInstanceId(kogitoProcessInstanceId);
         setKogitoRootProcessInstanceId(kogitoRootProcessInstanceId);
         setKogitoProcessId(kogitoProcessId);

api/kogito-events-core/src/main/java/org/kie/kogito/event/process/MultipleProcessInstanceDataEvent.java

@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.kie.kogito.event.process;
+
+import java.net.URI;
+import java.util.Collection;
+
+public class MultipleProcessInstanceDataEvent extends ProcessInstanceDataEvent<Collection<ProcessInstanceDataEvent<?>>> {
+
+    public static final String TYPE = "MultipleProcessInstanceDataEvent";
+
+    public MultipleProcessInstanceDataEvent() {
+    }
+
+    public MultipleProcessInstanceDataEvent(URI source, Collection<ProcessInstanceDataEvent<?>> body) {
+        super(TYPE, source, body);
+    }
+}

api/kogito-events-core/src/main/java/org/kie/kogito/event/process/ProcessInstanceDataEvent.java

@@ -18,6 +18,8 @@
  */
 package org.kie.kogito.event.process;
 
+import java.net.URI;
+
 import org.kie.kogito.event.AbstractDataEvent;
 
 public class ProcessInstanceDataEvent<T> extends AbstractDataEvent<T> {
@@ -29,6 +31,10 @@ public ProcessInstanceDataEvent(T body) {
         setData(body);
     }
 
+    protected ProcessInstanceDataEvent(String type, URI source, T body) {
+        super(type, source, body);
+    }
+
     public ProcessInstanceDataEvent(String type,
             String source,
             T body,

api/kogito-events-core/src/main/java/org/kie/kogito/event/usertask/MultipleUserTaskInstanceDataEvent.java

@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.kie.kogito.event.usertask;
+
+import java.net.URI;
+import java.util.Collection;
+
+public class MultipleUserTaskInstanceDataEvent extends UserTaskInstanceDataEvent<Collection<UserTaskInstanceDataEvent<?>>> {
+
+    public static final String TYPE = "MultipleUserTaskInstanceDataEvent";
+
+    public MultipleUserTaskInstanceDataEvent() {
+    }
+
+    public MultipleUserTaskInstanceDataEvent(URI source, Collection<UserTaskInstanceDataEvent<?>> body) {
+        super(TYPE, source, body);
+    }
+}

api/kogito-events-core/src/main/java/org/kie/kogito/event/usertask/UserTaskInstanceDataEvent.java

@@ -18,6 +18,7 @@
  */
 package org.kie.kogito.event.usertask;
 
+import java.net.URI;
 import java.util.Set;
 
 import org.kie.kogito.event.AbstractDataEvent;
@@ -48,6 +49,10 @@ public UserTaskInstanceDataEvent(T body) {
         setData(body);
     }
 
+    protected UserTaskInstanceDataEvent(String type, URI source, T body) {
+        super(type, source, body);
+    }
+
     public UserTaskInstanceDataEvent(String type,
             String source,
             T body,

kogito-build/kogito-dependencies-bom/pom.xml

@@ -52,7 +52,7 @@
     <version.io.quarkiverse.jackson-jq>2.0.2</version.io.quarkiverse.jackson-jq>
     <version.io.quarkiverse.openapi.generator>2.4.1</version.io.quarkiverse.openapi.generator>
     <version.io.quarkiverse.asyncapi>0.3.0</version.io.quarkiverse.asyncapi>
-    <version.io.quarkiverse.reactivemessaging.http>2.2.0</version.io.quarkiverse.reactivemessaging.http>
+    <version.io.quarkiverse.reactivemessaging.http>2.4.1</version.io.quarkiverse.reactivemessaging.http>
     <version.io.quarkiverse.embedded.postgresql>0.2.3</version.io.quarkiverse.embedded.postgresql>
     <version.com.github.haifengl.smile>1.5.2</version.com.github.haifengl.smile>
     <version.com.github.javaparser>3.25.8</version.com.github.javaparser>

quarkus/addons/events/process/runtime/pom.xml

@@ -78,6 +78,17 @@
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
     </dependency>
+
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>io.quarkus</groupId>
+      <artifactId>quarkus-junit5-mockito</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

quarkus/addons/events/process/runtime/src/main/java/org/kie/kogito/events/process/AbstractMessagingEventPublisher.java

@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.kie.kogito.events.process;
+
+import java.util.Collection;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Consumer;
+
+import org.eclipse.microprofile.reactive.messaging.Channel;
+import org.eclipse.microprofile.reactive.messaging.Message;
+import org.eclipse.microprofile.reactive.messaging.OnOverflow;
+import org.eclipse.microprofile.reactive.messaging.OnOverflow.Strategy;
+import org.kie.kogito.addon.quarkus.common.reactive.messaging.MessageDecoratorProvider;
+import org.kie.kogito.event.DataEvent;
+import org.kie.kogito.event.EventPublisher;
+import org.kie.kogito.events.config.EventsRuntimeConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import io.smallrye.reactive.messaging.MutinyEmitter;
+import io.smallrye.reactive.messaging.providers.locals.ContextAwareMessage;
+
+import jakarta.annotation.PostConstruct;
+import jakarta.enterprise.inject.Instance;
+import jakarta.inject.Inject;
+
+public abstract class AbstractMessagingEventPublisher implements EventPublisher {
+
+    private static final Logger logger = LoggerFactory.getLogger(AbstractMessagingEventPublisher.class);
+
+    @Inject
+    ObjectMapper json;
+
+    @Inject
+    @Channel(PROCESS_INSTANCES_TOPIC_NAME)
+    @OnOverflow(Strategy.UNBOUNDED_BUFFER)
+    MutinyEmitter<String> processInstancesEventsEmitter;
+    private AbstractMessageEmitter processInstanceConsumer;
+
+    @Inject
+    @Channel(PROCESS_DEFINITIONS_TOPIC_NAME)
+    MutinyEmitter<String> processDefinitionEventsEmitter;
+    private AbstractMessageEmitter processDefinitionConsumer;
+
+    @Inject
+    @Channel(USER_TASK_INSTANCES_TOPIC_NAME)
+    MutinyEmitter<String> userTasksEventsEmitter;
+    private AbstractMessageEmitter userTaskConsumer;
+    @Inject
+    EventsRuntimeConfig eventsRuntimeConfig;
+
+    @Inject
+    Instance<MessageDecoratorProvider> decoratorProviderInstance;
+
+    private MessageDecoratorProvider decoratorProvider;
+
+    @PostConstruct
+    public void init() {
+        decoratorProvider = decoratorProviderInstance.isResolvable() ? decoratorProviderInstance.get() : null;
+        processDefinitionConsumer = eventsRuntimeConfig.isProcessInstancesPropagateError() ? new BlockingMessageEmitter(processDefinitionEventsEmitter, PROCESS_DEFINITIONS_TOPIC_NAME)
+                : new ReactiveMessageEmitter(processDefinitionEventsEmitter, PROCESS_DEFINITIONS_TOPIC_NAME);
+        processInstanceConsumer = eventsRuntimeConfig.isProcessDefinitionPropagateError() ? new BlockingMessageEmitter(processInstancesEventsEmitter, PROCESS_INSTANCES_TOPIC_NAME)
+                : new ReactiveMessageEmitter(processInstancesEventsEmitter, PROCESS_INSTANCES_TOPIC_NAME);
+        userTaskConsumer = eventsRuntimeConfig.isUserTasksPropagateError() ? new BlockingMessageEmitter(userTasksEventsEmitter, USER_TASK_INSTANCES_TOPIC_NAME)
+                : new ReactiveMessageEmitter(userTasksEventsEmitter, USER_TASK_INSTANCES_TOPIC_NAME);
+    }
+
+    protected Optional<AbstractMessageEmitter> getConsumer(DataEvent<?> event) {
+        if (event == null) {
+            return Optional.empty();
+        }
+        switch (event.getType()) {
+            case "ProcessDefinitionEvent":
+                return eventsRuntimeConfig.isProcessDefinitionEventsEnabled() ? Optional.of(processDefinitionConsumer) : Optional.empty();
+
+            case "ProcessInstanceErrorDataEvent":
+            case "ProcessInstanceNodeDataEvent":
+            case "ProcessInstanceSLADataEvent":
+            case "ProcessInstanceStateDataEvent":
+            case "ProcessInstanceVariableDataEvent":
+                return eventsRuntimeConfig.isProcessInstancesEventsEnabled() ? Optional.of(processInstanceConsumer) : Optional.empty();
+
+            case "UserTaskInstanceAssignmentDataEvent":
+            case "UserTaskInstanceAttachmentDataEvent":
+            case "UserTaskInstanceCommentDataEvent":
+            case "UserTaskInstanceDeadlineDataEvent":
+            case "UserTaskInstanceStateDataEvent":
+            case "UserTaskInstanceVariableDataEvent":
+                return eventsRuntimeConfig.isUserTasksEventsEnabled() ? Optional.of(userTaskConsumer) : Optional.empty();
+
+            default:
+                return Optional.empty();
+        }
+    }
+
+    @Override
+    public void publish(Collection<DataEvent<?>> events) {
+        for (DataEvent<?> event : events) {
+            publish(event);
+        }
+    }
+
+    protected void publishToTopic(AbstractMessageEmitter emitter, Object event) {
+        logger.debug("About to publish event {} to topic {}", event, emitter.topic);
+        Message<String> message = null;
+        try {
+            String eventString = json.writeValueAsString(event);
+            logger.debug("Event payload '{}'", eventString);
+            message = decorateMessage(ContextAwareMessage.of(eventString));
+        } catch (Exception e) {
+            logger.error("Error while creating event to topic {} for event {}", emitter.topic, event);
+        }
+        if (message != null) {
+            emitter.accept(message);
+        }
+    }
+
+    protected Message<String> decorateMessage(Message<String> message) {
+        return decoratorProvider != null ? decoratorProvider.decorate(message) : message;
+    }
+
+    protected static abstract class AbstractMessageEmitter implements Consumer<Message<String>> {
+
+        protected final String topic;
+        protected final MutinyEmitter<String> emitter;
+
+        protected AbstractMessageEmitter(MutinyEmitter<String> emitter, String topic) {
+            this.emitter = emitter;
+            this.topic = topic;
+        }
+    }
+
+    private static class BlockingMessageEmitter extends AbstractMessageEmitter {
+        protected BlockingMessageEmitter(MutinyEmitter<String> emitter, String topic) {
+            super(emitter, topic);
+        }
+
+        @Override
+        public void accept(Message<String> message) {
+            emitter.sendMessageAndAwait(message);
+            logger.debug("Successfully published message {}", message.getPayload());
+        }
+    }
+
+    private static class ReactiveMessageEmitter extends AbstractMessageEmitter {
+        protected ReactiveMessageEmitter(MutinyEmitter<String> emitter, String topic) {
+            super(emitter, topic);
+        }
+
+        @Override
+        public void accept(Message<String> message) {
+            emitter.sendMessageAndForget(message
+                    .withAck(() -> onAck(message))
+                    .withNack(reason -> onNack(reason, message)));
+        }
+
+        private CompletionStage<Void> onAck(Message<String> message) {
+            logger.debug("Successfully published message {}", message.getPayload());
+            return CompletableFuture.completedFuture(null);
+        }
+
+        private CompletionStage<Void> onNack(Throwable reason, Message<String> message) {
+            logger.error("Error while publishing message {}", message, reason);
+            return CompletableFuture.completedFuture(null);
+        }
+
+    }
+}