You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kie.apache.org by ri...@apache.org on 2025/06/27 18:55:27 UTC
(incubator-kie-kogito-runtimes) branch main updated: [Fix #3971] Adding default event broker and support for context attrs (#3972)
This is an automated email from the ASF dual-hosted git repository.
ricardozanini pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-kie-kogito-runtimes.git
The following commit(s) were added to refs/heads/main by this push:
new 642fcd80bc [Fix #3971] Adding default event broker and support for context attrs (#3972)
642fcd80bc is described below
commit 642fcd80bc55df7bf4a686f24c0e0c6157803664
Author: Francisco Javier Tirado Sarti <65...@users.noreply.github.com>
AuthorDate: Fri Jun 27 20:55:24 2025 +0200
[Fix #3971] Adding default event broker and support for context attrs (#3972)
* [Fix #3971] Adding default event broker and support for context attrs
* [Fix #3973] Adding codegen support
---
.../org/kie/kogito/event/DataEventFactory.java | 13 +++++-
.../kogito/event/impl/AbstractMessageProducer.java | 8 ++--
.../event/impl/MessageProducerWithContext.java | 20 ++++----
.../canonical/descriptors/ExpressionUtils.java | 10 ++--
.../instance/impl/actions/ProduceEventAction.java | 18 ++++++--
.../MessageProducerJavaTemplate.java | 4 +-
.../handlers/CompositeContextNodeHandler.java | 3 +-
.../workflow/parser/handlers/StateHandler.java | 5 +-
.../suppliers/ProduceEventActionSupplier.java | 36 +++++++--------
.../executor/events/CloudEventReceiver.java} | 30 ++++++------
.../executor/events/InMemoryEventEmitter.java | 28 ++++++++----
.../events/InMemoryEventEmitterFactory.java | 53 ++++++++++++++++++++++
.../events/InMemoryEventReceiverFactory.java | 23 +++++-----
.../executor/events/InMemoryEventShared.java | 25 +++++-----
.../org.kie.kogito.event.EventEmitterFactory | 20 ++++++++
.../org.kie.kogito.event.EventReceiverFactory | 20 ++++++++
.../StaticFluentWorkflowApplicationTest.java | 17 +++++++
.../executor/KafkaEventReceiverFactory.java | 7 +--
.../serverless/workflow/fluent/ActionBuilder.java | 13 +++++-
.../workflow/actions/SWFProduceEventAction.java | 19 +++++---
.../kie/kogito/jackson/utils/StringConverter.java | 2 +-
21 files changed, 266 insertions(+), 108 deletions(-)
diff --git a/api/kogito-events-core/src/main/java/org/kie/kogito/event/DataEventFactory.java b/api/kogito-events-core/src/main/java/org/kie/kogito/event/DataEventFactory.java
index f2317a5d66..1586ac31d7 100644
--- a/api/kogito-events-core/src/main/java/org/kie/kogito/event/DataEventFactory.java
+++ b/api/kogito-events-core/src/main/java/org/kie/kogito/event/DataEventFactory.java
@@ -21,6 +21,8 @@ package org.kie.kogito.event;
import java.io.IOException;
import java.net.URI;
import java.time.OffsetDateTime;
+import java.util.Collections;
+import java.util.Map;
import java.util.Optional;
import java.util.UUID;
@@ -61,7 +63,15 @@ public class DataEventFactory {
}
public static <T> DataEvent<T> from(T eventData, String trigger, KogitoProcessInstance pi) {
- return from(eventData, trigger, URI.create("/process/" + pi.getProcessId()), Optional.empty(), ProcessMeta.fromKogitoProcessInstance(pi));
+ return from(eventData, trigger, pi, Collections.emptyMap());
+ }
+
+ public static <T> DataEvent<T> from(T eventData, String trigger, KogitoProcessInstance pi, Map<String, Object> contextAttributes) {
+ AbstractDataEvent<T> ce = (AbstractDataEvent<T>) from(eventData, trigger, URI.create("/process/" + pi.getProcessId()), Optional.empty(), ProcessMeta.fromKogitoProcessInstance(pi));
+ if (contextAttributes != null) {
+ contextAttributes.forEach((k, v) -> ce.addExtensionAttribute(k, v));
+ }
+ return ce;
}
public static <T> DataEvent<T> from(T eventData, String type, URI source, Optional<String> subject, CloudEventExtension... extensions) {
@@ -82,5 +92,4 @@ public class DataEventFactory {
private DataEventFactory() {
}
-
}
diff --git a/api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/AbstractMessageProducer.java b/api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/AbstractMessageProducer.java
index 36981370ca..7a732a4258 100644
--- a/api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/AbstractMessageProducer.java
+++ b/api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/AbstractMessageProducer.java
@@ -18,13 +18,15 @@
*/
package org.kie.kogito.event.impl;
+import java.util.Map;
+
import org.kie.kogito.event.DataEventFactory;
import org.kie.kogito.event.EventEmitter;
import org.kie.kogito.internal.process.runtime.KogitoProcessInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public abstract class AbstractMessageProducer<D> implements MessageProducer<D> {
+public abstract class AbstractMessageProducer<D> implements MessageProducerWithContext<D> {
private static final Logger logger = LoggerFactory.getLogger(AbstractMessageProducer.class);
@@ -48,8 +50,8 @@ public abstract class AbstractMessageProducer<D> implements MessageProducer<D> {
}
@Override
- public void produce(KogitoProcessInstance pi, D eventData) {
- emitter.emit(DataEventFactory.from(eventData, trigger, pi))
+ public void produce(KogitoProcessInstance pi, D eventData, Map<String, Object> contextAttrs) {
+ emitter.emit(DataEventFactory.from(eventData, trigger, pi, contextAttrs))
.exceptionally(ex -> {
logger.error("An error was caught while process " + pi.getProcessId() + " produced message " + eventData, ex);
return null;
diff --git a/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java b/api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/MessageProducerWithContext.java
similarity index 61%
copy from kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
copy to api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/MessageProducerWithContext.java
index d2bf12f26d..89c1ca1c15 100644
--- a/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
+++ b/api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/MessageProducerWithContext.java
@@ -16,21 +16,17 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.kie.kogito.jackson.utils;
+package org.kie.kogito.event.impl;
-import java.util.function.Function;
+import java.util.Map;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
+import org.kie.kogito.internal.process.runtime.KogitoProcessInstance;
-public class StringConverter implements Function<JsonNode, String> {
+public interface MessageProducerWithContext<D> extends MessageProducer<D> {
- @Override
- public String apply(JsonNode t) {
- try {
- return JsonObjectUtils.toString(t);
- } catch (JsonProcessingException e) {
- throw new IllegalArgumentException("Invalid value for json node " + t);
- }
+ default void produce(KogitoProcessInstance pi, D eventData) {
+ produce(pi, eventData, Map.of());
}
+
+ void produce(KogitoProcessInstance pi, D eventData, Map<String, Object> contextAttrs);
}
diff --git a/jbpm/jbpm-flow-builder/src/main/java/org/jbpm/compiler/canonical/descriptors/ExpressionUtils.java b/jbpm/jbpm-flow-builder/src/main/java/org/jbpm/compiler/canonical/descriptors/ExpressionUtils.java
index f810334326..3d9cec7c71 100644
--- a/jbpm/jbpm-flow-builder/src/main/java/org/jbpm/compiler/canonical/descriptors/ExpressionUtils.java
+++ b/jbpm/jbpm-flow-builder/src/main/java/org/jbpm/compiler/canonical/descriptors/ExpressionUtils.java
@@ -141,10 +141,12 @@ public class ExpressionUtils {
}
if (objectClass != null) {
// will generate TypeConverterRegistry.get().forType("JsonNode.class").apply("{\"dog\":\"perro\"}"))
- return new CastExpr(StaticJavaParser.parseClassOrInterfaceType(object.getClass().getName()),
- new MethodCallExpr(new MethodCallExpr(new MethodCallExpr(new TypeExpr(StaticJavaParser.parseClassOrInterfaceType(TypeConverterRegistry.class.getName())), "get"), "forType",
- NodeList.nodeList(new StringLiteralExpr(objectClass.getName()))), "apply",
- NodeList.nodeList(new StringLiteralExpr().setString(TypeConverterRegistry.get().forTypeReverse(object).apply((object))))));
+ String str = TypeConverterRegistry.get().forTypeReverse(object).apply(object);
+ return str == null ? new NullLiteralExpr()
+ : new CastExpr(StaticJavaParser.parseClassOrInterfaceType(object.getClass().getName()),
+ new MethodCallExpr(new MethodCallExpr(new MethodCallExpr(new TypeExpr(StaticJavaParser.parseClassOrInterfaceType(TypeConverterRegistry.class.getName())), "get"), "forType",
+ NodeList.nodeList(new StringLiteralExpr(objectClass.getName()))), "apply",
+ NodeList.nodeList(new StringLiteralExpr().setString(str))));
} else {
return new StringLiteralExpr().setString(object.toString());
}
diff --git a/jbpm/jbpm-flow/src/main/java/org/jbpm/process/instance/impl/actions/ProduceEventAction.java b/jbpm/jbpm-flow/src/main/java/org/jbpm/process/instance/impl/actions/ProduceEventAction.java
index c9f9c3c385..6dc4fcb217 100644
--- a/jbpm/jbpm-flow/src/main/java/org/jbpm/process/instance/impl/actions/ProduceEventAction.java
+++ b/jbpm/jbpm-flow/src/main/java/org/jbpm/process/instance/impl/actions/ProduceEventAction.java
@@ -19,6 +19,7 @@
package org.jbpm.process.instance.impl.actions;
import java.io.Serializable;
+import java.util.Collections;
import java.util.Map;
import java.util.function.Supplier;
@@ -28,6 +29,7 @@ import org.jbpm.process.instance.impl.Action;
import org.jbpm.workflow.core.impl.NodeIoHelper;
import org.jbpm.workflow.instance.impl.NodeInstanceImpl;
import org.kie.kogito.event.impl.MessageProducer;
+import org.kie.kogito.event.impl.MessageProducerWithContext;
import org.kie.kogito.internal.process.runtime.KogitoProcessContext;
import org.kie.kogito.internal.process.runtime.KogitoProcessInstance;
@@ -37,9 +39,9 @@ public class ProduceEventAction<T> implements Action, Serializable {
private final String varName;
private final String triggerName;
- private final Supplier<MessageProducer<T>> supplier;
+ private final Supplier<? extends MessageProducer<T>> supplier;
- public ProduceEventAction(String triggerName, String varName, Supplier<MessageProducer<T>> supplier) {
+ public ProduceEventAction(String triggerName, String varName, Supplier<? extends MessageProducer<T>> supplier) {
this.triggerName = triggerName;
this.varName = varName;
this.supplier = supplier;
@@ -53,7 +55,17 @@ public class ProduceEventAction<T> implements Action, Serializable {
InternalKnowledgeRuntime runtime = (InternalKnowledgeRuntime) context.getKieRuntime();
InternalProcessRuntime process = (InternalProcessRuntime) runtime.getProcessRuntime();
process.getProcessEventSupport().fireOnMessage(pi, context.getNodeInstance(), runtime, triggerName, object);
- supplier.get().produce(pi, getObject(object, context));
+ final MessageProducer<T> producer = supplier.get();
+ final T obj = getObject(object, context);
+ if (producer instanceof MessageProducerWithContext) {
+ ((MessageProducerWithContext<T>) producer).produce(pi, obj, getContextAttrs(object, context));
+ } else {
+ producer.produce(pi, obj);
+ }
+ }
+
+ protected Map<String, Object> getContextAttrs(Object object, KogitoProcessContext context) {
+ return Collections.emptyMap();
}
protected T getObject(Object object, KogitoProcessContext context) {
diff --git a/kogito-codegen-modules/kogito-codegen-processes/src/main/resources/class-templates/MessageProducerJavaTemplate.java b/kogito-codegen-modules/kogito-codegen-processes/src/main/resources/class-templates/MessageProducerJavaTemplate.java
index 4eb7b7f110..f2ed6e5581 100644
--- a/kogito-codegen-modules/kogito-codegen-processes/src/main/resources/class-templates/MessageProducerJavaTemplate.java
+++ b/kogito-codegen-modules/kogito-codegen-processes/src/main/resources/class-templates/MessageProducerJavaTemplate.java
@@ -20,6 +20,7 @@ package com.myspace.demo;
import java.io.IOException;
import java.util.Optional;
+import java.util.Map;
import org.kie.kogito.event.EventMarshaller;
import org.kie.kogito.event.impl.StringEventMarshaller;
@@ -39,7 +40,8 @@ public class MessageProducer extends org.kie.kogito.event.impl.AbstractMessagePr
}
- public void produce(KogitoProcessInstance pi, $Type$ eventData) {
+ @Override
+ public void produce(KogitoProcessInstance pi, $Type$ eventData, Map<String, Object> contextAttrs) {
}
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/CompositeContextNodeHandler.java b/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/CompositeContextNodeHandler.java
index 49deb2d8e2..68791096b2 100644
--- a/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/CompositeContextNodeHandler.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/CompositeContextNodeHandler.java
@@ -187,7 +187,8 @@ public abstract class CompositeContextNodeHandler<S extends State> extends State
private NodeFactory<?, ?> getActionNode(RuleFlowNodeContainerFactory<?, ?> embeddedSubProcess,
EventRef eventRef, String inputVar) {
- return sendEventNode(embeddedSubProcess.actionNode(parserContext.newId()), eventDefinition(eventRef.getTriggerEventRef()), eventRef.getData(), inputVar);
+ return sendEventNode(
+ embeddedSubProcess.actionNode(parserContext.newId()), eventDefinition(eventRef.getTriggerEventRef()), eventRef.getData(), eventRef.getContextAttributes(), inputVar);
}
private NodeFactory<?, ?> getActionNode(RuleFlowNodeContainerFactory<?, ?> embeddedSubProcess,
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/StateHandler.java b/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/StateHandler.java
index 77fa36cdc3..51f90a079f 100644
--- a/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/StateHandler.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/parser/handlers/StateHandler.java
@@ -133,15 +133,16 @@ public abstract class StateHandler<S extends State> {
protected final <T extends NodeFactory<?, ?> & SupportsAction<?, ?>>
NodeFactory<?, ?> sendEventNode(T actionNode, ProduceEvent event) {
- return sendEventNode(actionNode, eventDefinition(event.getEventRef()), event.getData(), DEFAULT_WORKFLOW_VAR);
+ return sendEventNode(actionNode, eventDefinition(event.getEventRef()), event.getData(), event.getContextAttributes(), DEFAULT_WORKFLOW_VAR);
}
protected final <T extends NodeFactory<?, ?> & SupportsAction<?, ?>> NodeFactory<?, ?> sendEventNode(T actionNode,
EventDefinition eventDefinition,
JsonNode data,
+ Map<String, String> contextAttributes,
String defaultWorkflowVar) {
return NodeFactoryUtils.sendEventNode(
- actionNode.action(new ProduceEventActionSupplier(workflow, eventDefinition.getType(), defaultWorkflowVar, data)),
+ actionNode.action(new ProduceEventActionSupplier(workflow, eventDefinition.getType(), defaultWorkflowVar, data, contextAttributes)),
eventDefinition,
defaultWorkflowVar);
}
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/suppliers/ProduceEventActionSupplier.java b/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/suppliers/ProduceEventActionSupplier.java
index 7739f056ff..1ef5bc812b 100644
--- a/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/suppliers/ProduceEventActionSupplier.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-builder/src/main/java/org/kie/kogito/serverless/workflow/suppliers/ProduceEventActionSupplier.java
@@ -18,24 +18,23 @@
*/
package org.kie.kogito.serverless.workflow.suppliers;
-import java.io.UncheckedIOException;
+import java.util.Collections;
+import java.util.Map;
import java.util.function.Supplier;
import org.jbpm.compiler.canonical.AbstractNodeVisitor;
import org.jbpm.compiler.canonical.ExpressionSupplier;
import org.jbpm.compiler.canonical.ProcessMetaData;
import org.jbpm.compiler.canonical.TriggerMetaData;
+import org.jbpm.compiler.canonical.descriptors.ExpressionUtils;
import org.jbpm.process.core.event.StaticMessageProducer;
-import org.kie.kogito.event.impl.MessageProducer;
+import org.kie.kogito.event.impl.MessageProducerWithContext;
import org.kie.kogito.internal.process.runtime.KogitoNode;
import org.kie.kogito.jackson.utils.JsonObjectUtils;
import org.kie.kogito.serverless.workflow.actions.SWFProduceEventAction;
-import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.github.javaparser.ast.expr.Expression;
-import com.github.javaparser.ast.expr.MethodCallExpr;
-import com.github.javaparser.ast.expr.NullLiteralExpr;
import com.github.javaparser.ast.expr.StringLiteralExpr;
import io.serverlessworkflow.api.Workflow;
@@ -48,26 +47,25 @@ public class ProduceEventActionSupplier extends SWFProduceEventAction implements
private static final long serialVersionUID = 1L;
public ProduceEventActionSupplier(Workflow workflow, String trigger, String varName, JsonNode data) {
- super(trigger, varName, new MessageProducerSupplier(trigger), workflow.getExpressionLang(), data);
+ this(workflow, trigger, varName, data, Collections.emptyMap());
+ }
+
+ public ProduceEventActionSupplier(Workflow workflow, String trigger, String varName, JsonNode data, Map<String, String> contextAttributes) {
+ super(trigger, varName, new MessageProducerSupplier(trigger), workflow.getExpressionLang(), data, JsonObjectUtils.fromValue(contextAttributes));
}
@Override
public Expression get(KogitoNode node, ProcessMetaData metadata) {
- try {
- return AbstractNodeVisitor.buildProducerAction(parseClassOrInterfaceType(SWFProduceEventAction.class.getCanonicalName()), TriggerMetaData.of(node, (String) node.getMetaData()
- .get(MAPPING_VARIABLE_INPUT)), metadata)
- .addArgument(new StringLiteralExpr(exprLang))
- .addArgument(data != null ? new MethodCallExpr(JsonObjectUtils.class.getCanonicalName() + ".fromString").addArgument(
- JsonObjectUtils.toString(data))
- : new NullLiteralExpr());
- } catch (JsonProcessingException e) {
- throw new UncheckedIOException(e);
- }
+ return AbstractNodeVisitor.buildProducerAction(parseClassOrInterfaceType(SWFProduceEventAction.class.getCanonicalName()), TriggerMetaData.of(node, (String) node.getMetaData()
+ .get(MAPPING_VARIABLE_INPUT)), metadata)
+ .addArgument(new StringLiteralExpr(exprLang))
+ .addArgument(ExpressionUtils.getLiteralExpr(data))
+ .addArgument(ExpressionUtils.getLiteralExpr(contextAttrs));
}
- private static class MessageProducerSupplier implements Supplier<MessageProducer<JsonNode>> {
+ private static class MessageProducerSupplier implements Supplier<MessageProducerWithContext<JsonNode>> {
- private MessageProducer<JsonNode> producer;
+ private MessageProducerWithContext<JsonNode> producer;
private final String trigger;
public MessageProducerSupplier(String trigger) {
@@ -75,7 +73,7 @@ public class ProduceEventActionSupplier extends SWFProduceEventAction implements
}
@Override
- public MessageProducer<JsonNode> get() {
+ public MessageProducerWithContext<JsonNode> get() {
if (producer == null) {
producer = new StaticMessageProducer<>(trigger);
}
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-executor-kafka/src/main/java/org/kie/kogito/serverless/workflow/executor/KafkaEventReceiver.java b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/CloudEventReceiver.java
similarity index 92%
rename from kogito-serverless-workflow/kogito-serverless-workflow-executor-kafka/src/main/java/org/kie/kogito/serverless/workflow/executor/KafkaEventReceiver.java
rename to kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/CloudEventReceiver.java
index 42fe070290..87d81f4f25 100644
--- a/kogito-serverless-workflow/kogito-serverless-workflow-executor-kafka/src/main/java/org/kie/kogito/serverless/workflow/executor/KafkaEventReceiver.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/CloudEventReceiver.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.kie.kogito.serverless.workflow.executor;
+package org.kie.kogito.serverless.workflow.executor.events;
import java.io.IOException;
import java.util.ArrayList;
@@ -39,21 +39,10 @@ import org.slf4j.LoggerFactory;
import io.cloudevents.CloudEvent;
import io.cloudevents.CloudEventData;
-public class KafkaEventReceiver implements EventReceiver {
+public class CloudEventReceiver implements EventReceiver {
- private static final Logger logger = LoggerFactory.getLogger(KafkaEventReceiver.class);
- private final Collection<Subscription<?, CloudEvent>> subscriptions = new ArrayList<>();
-
- @SuppressWarnings({ "rawtypes", "unchecked" })
- public void onEvent(CloudEvent value) {
- for (Subscription subscription : subscriptions) {
- try {
- subscription.getConsumer().apply(subscription.getConverter().convert(value));
- } catch (IOException e) {
- logger.info("Problem deserializing event {}", value, e);
- }
- }
- }
+ private static final Logger logger = LoggerFactory.getLogger(CloudEventReceiver.class);
+ protected final Collection<Subscription<?, CloudEvent>> subscriptions = new ArrayList<>();
@Override
public <T> void subscribe(Function<DataEvent<T>, CompletionStage<?>> consumer, Class<T> dataClass) {
@@ -79,4 +68,15 @@ public class KafkaEventReceiver implements EventReceiver {
}
})));
}
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ public void onEvent(CloudEvent value) {
+ for (Subscription subscription : subscriptions) {
+ try {
+ subscription.getConsumer().apply(subscription.getConverter().convert(value));
+ } catch (IOException e) {
+ logger.info("Problem deserializing event {}", value, e);
+ }
+ }
+ }
}
diff --git a/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventEmitter.java
similarity index 54%
copy from kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
copy to kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventEmitter.java
index d2bf12f26d..7f5c90da3b 100644
--- a/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventEmitter.java
@@ -16,21 +16,29 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.kie.kogito.jackson.utils;
+package org.kie.kogito.serverless.workflow.executor.events;
-import java.util.function.Function;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+import org.kie.kogito.event.DataEvent;
+import org.kie.kogito.event.EventEmitter;
-import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
-public class StringConverter implements Function<JsonNode, String> {
+import io.cloudevents.jackson.JsonCloudEventData;
+
+public class InMemoryEventEmitter implements EventEmitter {
+
+ private CloudEventReceiver eventReceiver;
+
+ InMemoryEventEmitter(CloudEventReceiver eventReceiver) {
+ this.eventReceiver = eventReceiver;
+ }
@Override
- public String apply(JsonNode t) {
- try {
- return JsonObjectUtils.toString(t);
- } catch (JsonProcessingException e) {
- throw new IllegalArgumentException("Invalid value for json node " + t);
- }
+ public CompletionStage<Void> emit(DataEvent<?> dataEvent) {
+ eventReceiver.onEvent(dataEvent.asCloudEvent(o -> JsonCloudEventData.wrap((JsonNode) o)));
+ return CompletableFuture.completedStage(null);
}
}
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventEmitterFactory.java b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventEmitterFactory.java
new file mode 100644
index 0000000000..a74d60924e
--- /dev/null
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventEmitterFactory.java
@@ -0,0 +1,53 @@
+/*
+ * 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.serverless.workflow.executor.events;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.kie.kogito.event.EventEmitter;
+import org.kie.kogito.event.EventEmitterFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class InMemoryEventEmitterFactory implements EventEmitterFactory {
+
+ private static final Logger logger = LoggerFactory.getLogger(InMemoryEventEmitterFactory.class);
+
+ private Map<String, EventEmitter> emitters = new ConcurrentHashMap<>();
+
+ @Override
+ public EventEmitter apply(String trigger) {
+ return emitters.computeIfAbsent(trigger, t -> {
+ CloudEventReceiver eventReceiver = InMemoryEventShared.INSTANCE.receivers().get(trigger);
+ if (eventReceiver != null) {
+ return new InMemoryEventEmitter(eventReceiver);
+ } else {
+ logger.warn("Unregisterd event type {}", trigger);
+ return null;
+ }
+ });
+
+ }
+
+ // lower priority
+ public int ordinal() {
+ return Integer.MAX_VALUE;
+ }
+}
diff --git a/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventReceiverFactory.java
similarity index 62%
copy from kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
copy to kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventReceiverFactory.java
index d2bf12f26d..5864607dde 100644
--- a/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventReceiverFactory.java
@@ -16,21 +16,20 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.kie.kogito.jackson.utils;
+package org.kie.kogito.serverless.workflow.executor.events;
-import java.util.function.Function;
+import org.kie.kogito.event.EventReceiver;
+import org.kie.kogito.event.EventReceiverFactory;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
-
-public class StringConverter implements Function<JsonNode, String> {
+public class InMemoryEventReceiverFactory implements EventReceiverFactory {
@Override
- public String apply(JsonNode t) {
- try {
- return JsonObjectUtils.toString(t);
- } catch (JsonProcessingException e) {
- throw new IllegalArgumentException("Invalid value for json node " + t);
- }
+ public EventReceiver apply(String trigger) {
+ return InMemoryEventShared.INSTANCE.receivers().computeIfAbsent(trigger, t -> new CloudEventReceiver());
+ }
+
+ // lower priority
+ public int ordinal() {
+ return Integer.MAX_VALUE;
}
}
diff --git a/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventShared.java
similarity index 61%
copy from kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
copy to kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventShared.java
index d2bf12f26d..19f225b8ab 100644
--- a/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/java/org/kie/kogito/serverless/workflow/executor/events/InMemoryEventShared.java
@@ -16,21 +16,22 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.kie.kogito.jackson.utils;
+package org.kie.kogito.serverless.workflow.executor.events;
-import java.util.function.Function;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
+public class InMemoryEventShared {
-public class StringConverter implements Function<JsonNode, String> {
+ public static final InMemoryEventShared INSTANCE = new InMemoryEventShared();
+
+ private Map<String, CloudEventReceiver> receivers = new ConcurrentHashMap<>();
+
+ public Map<String, CloudEventReceiver> receivers() {
+ return receivers;
+ }
+
+ private InMemoryEventShared() {
- @Override
- public String apply(JsonNode t) {
- try {
- return JsonObjectUtils.toString(t);
- } catch (JsonProcessingException e) {
- throw new IllegalArgumentException("Invalid value for json node " + t);
- }
}
}
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/resources/META-INF/services/org.kie.kogito.event.EventEmitterFactory b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/resources/META-INF/services/org.kie.kogito.event.EventEmitterFactory
new file mode 100644
index 0000000000..b5fa9c6ee4
--- /dev/null
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/resources/META-INF/services/org.kie.kogito.event.EventEmitterFactory
@@ -0,0 +1,20 @@
+#
+# 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.
+#
+
+org.kie.kogito.serverless.workflow.executor.events.InMemoryEventEmitterFactory
\ No newline at end of file
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/resources/META-INF/services/org.kie.kogito.event.EventReceiverFactory b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/resources/META-INF/services/org.kie.kogito.event.EventReceiverFactory
new file mode 100644
index 0000000000..1ef9aa52fd
--- /dev/null
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/main/resources/META-INF/services/org.kie.kogito.event.EventReceiverFactory
@@ -0,0 +1,20 @@
+#
+# 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.
+#
+
+org.kie.kogito.serverless.workflow.executor.events.InMemoryEventReceiverFactory
\ No newline at end of file
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/test/java/org/kie/kogito/serverless/workflow/executor/StaticFluentWorkflowApplicationTest.java b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/test/java/org/kie/kogito/serverless/workflow/executor/StaticFluentWorkflowApplicationTest.java
index c66e6f0e6f..37eb91e4b9 100644
--- a/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/test/java/org/kie/kogito/serverless/workflow/executor/StaticFluentWorkflowApplicationTest.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-executor-core/src/test/java/org/kie/kogito/serverless/workflow/executor/StaticFluentWorkflowApplicationTest.java
@@ -62,9 +62,12 @@ import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.kie.kogito.serverless.workflow.fluent.ActionBuilder.call;
import static org.kie.kogito.serverless.workflow.fluent.ActionBuilder.log;
import static org.kie.kogito.serverless.workflow.fluent.ActionBuilder.subprocess;
+import static org.kie.kogito.serverless.workflow.fluent.ActionBuilder.trigger;
+import static org.kie.kogito.serverless.workflow.fluent.EventDefBuilder.eventDef;
import static org.kie.kogito.serverless.workflow.fluent.FunctionBuilder.expr;
import static org.kie.kogito.serverless.workflow.fluent.FunctionBuilder.java;
import static org.kie.kogito.serverless.workflow.fluent.FunctionBuilder.log;
+import static org.kie.kogito.serverless.workflow.fluent.StateBuilder.callback;
import static org.kie.kogito.serverless.workflow.fluent.StateBuilder.forEach;
import static org.kie.kogito.serverless.workflow.fluent.StateBuilder.inject;
import static org.kie.kogito.serverless.workflow.fluent.StateBuilder.operation;
@@ -141,6 +144,20 @@ public class StaticFluentWorkflowApplicationTest {
}
}
+ @Test
+ void testEventPubSub() throws InterruptedException, TimeoutException {
+ final String eventType = "eventType";
+ Workflow subscriber =
+ workflow("testCallback").start(callback(call(expr("prefix", "{slogan:.slogan+\"er Beti\"}")), eventDef(eventType)).outputFilter("{slogan:.slogan+.name}")).end().build();
+ Workflow publisher = workflow("testPublishEvent").start(operation().action(trigger(eventDef("eventType"), jsonObject().put("name", ".name"), ".id"))).end().build();
+ try (StaticWorkflowApplication application = StaticWorkflowApplication.create()) {
+ String id = application.execute(subscriber, jsonObject().put("slogan", "Viva ")).getId();
+ application.execute(publisher, jsonObject().put("name", " manque pierda").put("id", id));
+ assertThat(application.waitForFinish(id, Duration.ofSeconds(3)).orElseThrow().getWorkflowdata())
+ .isEqualTo(jsonObject().put("slogan", "Viva er Beti manque pierda"));
+ }
+ }
+
@Test
void testSwitchLoop() throws InterruptedException, TimeoutException {
OperationStateBuilder startTask = operation().action(call(expr("startTask", "{finish:true}")));
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-executor-kafka/src/main/java/org/kie/kogito/serverless/workflow/executor/KafkaEventReceiverFactory.java b/kogito-serverless-workflow/kogito-serverless-workflow-executor-kafka/src/main/java/org/kie/kogito/serverless/workflow/executor/KafkaEventReceiverFactory.java
index bd91f88065..465b6a1197 100644
--- a/kogito-serverless-workflow/kogito-serverless-workflow-executor-kafka/src/main/java/org/kie/kogito/serverless/workflow/executor/KafkaEventReceiverFactory.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-executor-kafka/src/main/java/org/kie/kogito/serverless/workflow/executor/KafkaEventReceiverFactory.java
@@ -30,6 +30,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.kie.kogito.event.EventReceiver;
import org.kie.kogito.event.EventReceiverFactory;
+import org.kie.kogito.serverless.workflow.executor.events.CloudEventReceiver;
import org.kie.kogito.serverless.workflow.utils.ConfigResolverHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -40,14 +41,14 @@ public class KafkaEventReceiverFactory implements EventReceiverFactory {
private static final Logger logger = LoggerFactory.getLogger(KafkaEventReceiverFactory.class);
private Map<String, String> trigger2Topic = KafkaPropertiesFactory.get().triggerToTopicMap("kogito.addon.messaging.incoming.trigger.");
- private Map<String, KafkaEventReceiver> receivers = new ConcurrentHashMap<>();
+ private Map<String, CloudEventReceiver> receivers = new ConcurrentHashMap<>();
private Consumer<byte[], CloudEvent> consumer;
private Lock consumerLock = new ReentrantLock();
private Thread consumerThread;
@Override
public EventReceiver apply(String trigger) {
- return receivers.computeIfAbsent(trigger2Topic.getOrDefault(trigger, trigger), k -> new KafkaEventReceiver());
+ return receivers.computeIfAbsent(trigger2Topic.getOrDefault(trigger, trigger), k -> new CloudEventReceiver());
}
@Override
@@ -111,7 +112,7 @@ public class KafkaEventReceiverFactory implements EventReceiverFactory {
}
for (ConsumerRecord<byte[], CloudEvent> record : records) {
String topic = record.topic();
- KafkaEventReceiver receiver = receivers.get(topic);
+ CloudEventReceiver receiver = receivers.get(topic);
if (receiver == null) {
logger.info("No subscription for topic {}", topic);
} else {
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-fluent/src/main/java/org/kie/kogito/serverless/workflow/fluent/ActionBuilder.java b/kogito-serverless-workflow/kogito-serverless-workflow-fluent/src/main/java/org/kie/kogito/serverless/workflow/fluent/ActionBuilder.java
index bd0f6b1942..55e48cf889 100644
--- a/kogito-serverless-workflow/kogito-serverless-workflow-fluent/src/main/java/org/kie/kogito/serverless/workflow/fluent/ActionBuilder.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-fluent/src/main/java/org/kie/kogito/serverless/workflow/fluent/ActionBuilder.java
@@ -19,8 +19,11 @@
package org.kie.kogito.serverless.workflow.fluent;
import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
import java.util.Optional;
+import org.kie.kogito.event.cloudevents.CloudEventExtensionConstants;
import org.kie.kogito.jackson.utils.JsonObjectUtils;
import org.kie.kogito.jackson.utils.ObjectMapperFactory;
import org.kie.kogito.process.Process;
@@ -76,7 +79,15 @@ public class ActionBuilder {
}
public static ActionBuilder trigger(EventDefBuilder builder, JsonNode data) {
- ActionBuilder actionBuilder = new ActionBuilder(new Action().withEventRef(new EventRef().withData(data).withTriggerEventRef(builder.getName())));
+ return trigger(builder, data, Collections.emptyMap());
+ }
+
+ public static ActionBuilder trigger(EventDefBuilder builder, JsonNode data, String procRefId) {
+ return trigger(builder, data, Map.of(CloudEventExtensionConstants.PROCESS_REFERENCE_ID, procRefId));
+ }
+
+ public static ActionBuilder trigger(EventDefBuilder builder, JsonNode data, Map<String, String> contextAttributes) {
+ ActionBuilder actionBuilder = new ActionBuilder(new Action().withEventRef(new EventRef().withContextAttributes(contextAttributes).withData(data).withTriggerEventRef(builder.getName())));
actionBuilder.eventDefinition = Optional.of(builder);
return actionBuilder;
}
diff --git a/kogito-serverless-workflow/kogito-serverless-workflow-runtime/src/main/java/org/kie/kogito/serverless/workflow/actions/SWFProduceEventAction.java b/kogito-serverless-workflow/kogito-serverless-workflow-runtime/src/main/java/org/kie/kogito/serverless/workflow/actions/SWFProduceEventAction.java
index c01033cf4c..e9589d3943 100644
--- a/kogito-serverless-workflow/kogito-serverless-workflow-runtime/src/main/java/org/kie/kogito/serverless/workflow/actions/SWFProduceEventAction.java
+++ b/kogito-serverless-workflow/kogito-serverless-workflow-runtime/src/main/java/org/kie/kogito/serverless/workflow/actions/SWFProduceEventAction.java
@@ -18,10 +18,11 @@
*/
package org.kie.kogito.serverless.workflow.actions;
+import java.util.Map;
import java.util.function.Supplier;
import org.jbpm.process.instance.impl.actions.ProduceEventAction;
-import org.kie.kogito.event.impl.MessageProducer;
+import org.kie.kogito.event.impl.MessageProducerWithContext;
import org.kie.kogito.internal.process.runtime.KogitoProcessContext;
import org.kie.kogito.jackson.utils.JsonNodeVisitor;
import org.kie.kogito.jackson.utils.JsonObjectUtils;
@@ -34,19 +35,23 @@ public class SWFProduceEventAction extends ProduceEventAction<JsonNode> {
protected final String exprLang;
protected final JsonNode data;
+ protected final JsonNode contextAttrs;
- public SWFProduceEventAction(String triggerName, String varName, Supplier<MessageProducer<JsonNode>> supplier, String exprLang, JsonNode data) {
+ public SWFProduceEventAction(String triggerName, String varName, Supplier<MessageProducerWithContext<JsonNode>> supplier, String exprLang, JsonNode data, JsonNode contextAttrs) {
super(triggerName, varName, supplier);
this.exprLang = exprLang;
this.data = data;
+ this.contextAttrs = contextAttrs;
}
@Override
protected JsonNode getObject(Object object, KogitoProcessContext context) {
- if (data != null) {
- return JsonNodeVisitor.transformTextNode(data, node -> ExpressionHandlerUtils.transform(node, object, context, exprLang));
- } else {
- return JsonObjectUtils.fromValue(object);
- }
+ return data != null ? JsonNodeVisitor.transformTextNode(data, node -> ExpressionHandlerUtils.transform(node, object, context, exprLang)) : JsonObjectUtils.fromValue(object);
+ }
+
+ @Override
+ protected Map<String, Object> getContextAttrs(Object object, KogitoProcessContext context) {
+ return (Map<String, Object>) JsonObjectUtils
+ .toJavaValue(JsonNodeVisitor.transformNode(JsonObjectUtils.fromValue(contextAttrs), node -> ExpressionHandlerUtils.transform(node, object, context, exprLang), JsonNode::isTextual));
}
}
diff --git a/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java b/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
index d2bf12f26d..21a2b36ab1 100644
--- a/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
+++ b/kogito-workitems/kogito-jackson-utils/src/main/java/org/kie/kogito/jackson/utils/StringConverter.java
@@ -28,7 +28,7 @@ public class StringConverter implements Function<JsonNode, String> {
@Override
public String apply(JsonNode t) {
try {
- return JsonObjectUtils.toString(t);
+ return t.isNull() ? null : JsonObjectUtils.toString(t);
} catch (JsonProcessingException e) {
throw new IllegalArgumentException("Invalid value for json node " + t);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@kie.apache.org
For additional commands, e-mail: commits-help@kie.apache.org