This is an automated email from the ASF dual-hosted git repository.
porcelli pushed a commit to branch main
in repository
https://gitbox.apache.org/repos/asf/incubator-kie-kogito-examples.git
The following commit(s) were added to refs/heads/main by this push:
new 59a2aabb0 [kie-issues-249] Data Index Events changes for allowing
historical data (#1815)
59a2aabb0 is described below
commit 59a2aabb049736fed99b3dd1cbb5aa248dfcdc40
Author: Enrique <[email protected]>
AuthorDate: Wed Oct 18 19:28:06 2023 +0200
[kie-issues-249] Data Index Events changes for allowing historical data
(#1815)
* [kie-issues-249] Data Index Events changes for allowing historical data
* fix tests
* add some logging for messages
* test loggers
* fix for input1 in user tasks
---
.../org/kie/kogito/quarkus/outbox/OutboxIT.java | 33 +++++++++++++---------
.../kogito/performance/client/KafkaDispatcher.java | 4 +--
.../usertasks/CustomHumanTaskLifeCycle.java | 7 ++---
.../org/kie/kogito/springboot/outbox/OutboxIT.java | 32 ++++++++++++---------
.../usertasks/CustomHumanTaskLifeCycle.java | 7 ++---
5 files changed, 47 insertions(+), 36 deletions(-)
diff --git
a/kogito-quarkus-examples/process-outbox-mongodb-quarkus/src/test/java/org/kie/kogito/quarkus/outbox/OutboxIT.java
b/kogito-quarkus-examples/process-outbox-mongodb-quarkus/src/test/java/org/kie/kogito/quarkus/outbox/OutboxIT.java
index 35b77ce25..d1557806d 100644
---
a/kogito-quarkus-examples/process-outbox-mongodb-quarkus/src/test/java/org/kie/kogito/quarkus/outbox/OutboxIT.java
+++
b/kogito-quarkus-examples/process-outbox-mongodb-quarkus/src/test/java/org/kie/kogito/quarkus/outbox/OutboxIT.java
@@ -57,7 +57,6 @@ public class OutboxIT {
private static final String PROCESS_EVENTS_TOPIC =
"kogito-processinstances-events";
private static final String USERTASK_EVENTS_TOPIC =
"kogito-usertaskinstances-events";
- private static final String VARIABLE_EVENTS_TOPIC =
"kogito-variables-events";
private static final int KOGITO_PORT = 8080;
private static final int KAFKA_PORT = 9092;
private static final int DEBEZIUM_PORT = 8083;
@@ -156,18 +155,25 @@ public class OutboxIT {
CountDownLatch processEventCounter = new CountDownLatch(2);
CountDownLatch userTaskEventCounter = new CountDownLatch(1);
kafkaClient.consume(Set.of(PROCESS_EVENTS_TOPIC,
USERTASK_EVENTS_TOPIC), message -> {
+ LOGGER.info("ProcessInstanceVariableDataEvent: {}", message);
String type = JsonPath.read(message, "$.type");
- if ("ProcessInstanceEvent".equals(type)) {
- String orderNumber = JsonPath.read(message,
"$.data.variables.order.orderNumber");
- boolean shipped = JsonPath.read(message,
"$.data.variables.order.shipped");
- if ("23570".equals(orderNumber) && !shipped) {
- processEventCounter.countDown();
+ if ("ProcessInstanceVariableDataEvent".equals(type)) {
+ String varName = JsonPath.read(message, "$.data.variableName");
+ if ("order".equals(varName)) {
+ String orderNumber = JsonPath.read(message,
"$.data.variableValue.orderNumber");
+ boolean shipped = JsonPath.read(message,
"$.data.variableValue.shipped");
+ if ("23570".equals(orderNumber) && !shipped) {
+ processEventCounter.countDown();
+ }
}
- } else if ("UserTaskInstanceEvent".equals(type)) {
- String orderNumber = JsonPath.read(message,
"$.data.inputs.input1.orderNumber");
- boolean shipped = JsonPath.read(message,
"$.data.inputs.input1.shipped");
- if ("23570".equals(orderNumber) && !shipped) {
- userTaskEventCounter.countDown();
+ } else if ("UserTaskInstanceVariableDataEvent".equals(type)) {
+ String varName = JsonPath.read(message, "$.data.variableName");
+ if ("input1".equals(varName)) {
+ String orderNumber = JsonPath.read(message,
"$.data.variableValue.orderNumber");
+ boolean shipped = JsonPath.read(message,
"$.data.variableValue.shipped");
+ if ("23570".equals(orderNumber) && !shipped) {
+ userTaskEventCounter.countDown();
+ }
}
}
});
@@ -195,11 +201,12 @@ public class OutboxIT {
.get("/connectors/{connector}/topics")
.then()
.statusCode(200)
- .body("kogito-connector.topics", hasSize(3))
- .body("kogito-connector.topics",
hasItems(PROCESS_EVENTS_TOPIC, USERTASK_EVENTS_TOPIC, VARIABLE_EVENTS_TOPIC)));
+ .body("kogito-connector.topics", hasSize(2))
+ .body("kogito-connector.topics",
hasItems(PROCESS_EVENTS_TOPIC, USERTASK_EVENTS_TOPIC)));
// Check process events pushed
assertTrue(processEventCounter.await(TIMEOUT.getSeconds(),
TimeUnit.SECONDS));
+
assertTrue(userTaskEventCounter.await(TIMEOUT.getSeconds(),
TimeUnit.SECONDS));
}
}
diff --git
a/kogito-quarkus-examples/process-performance-client/src/main/java/org/kie/kogito/performance/client/KafkaDispatcher.java
b/kogito-quarkus-examples/process-performance-client/src/main/java/org/kie/kogito/performance/client/KafkaDispatcher.java
index 6355ffba1..119cdc3f4 100644
---
a/kogito-quarkus-examples/process-performance-client/src/main/java/org/kie/kogito/performance/client/KafkaDispatcher.java
+++
b/kogito-quarkus-examples/process-performance-client/src/main/java/org/kie/kogito/performance/client/KafkaDispatcher.java
@@ -26,7 +26,7 @@ import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.kie.kogito.event.process.ProcessDataEvent;
+import org.kie.kogito.event.process.ProcessInstanceDataEvent;
import org.kie.kogito.jackson.utils.ObjectMapperFactory;
import com.fasterxml.jackson.core.JsonProcessingException;
@@ -34,7 +34,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
public class KafkaDispatcher implements RequestDispatcher {
- private class ObjectCloudEvent extends ProcessDataEvent<Object> {
+ private class ObjectCloudEvent extends ProcessInstanceDataEvent<Object> {
public ObjectCloudEvent(String trigger, Object data) {
super(trigger, "java_client", data, null, null, null, null, null,
null, null, null, null, null, null);
}
diff --git
a/kogito-quarkus-examples/process-usertasks-custom-lifecycle-quarkus/src/main/java/org/acme/travels/usertasks/CustomHumanTaskLifeCycle.java
b/kogito-quarkus-examples/process-usertasks-custom-lifecycle-quarkus/src/main/java/org/acme/travels/usertasks/CustomHumanTaskLifeCycle.java
index 0265dcea7..c83b80302 100644
---
a/kogito-quarkus-examples/process-usertasks-custom-lifecycle-quarkus/src/main/java/org/acme/travels/usertasks/CustomHumanTaskLifeCycle.java
+++
b/kogito-quarkus-examples/process-usertasks-custom-lifecycle-quarkus/src/main/java/org/acme/travels/usertasks/CustomHumanTaskLifeCycle.java
@@ -23,7 +23,7 @@ import java.util.LinkedHashMap;
import java.util.Map;
import org.jbpm.process.instance.impl.humantask.BaseHumanTaskLifeCycle;
-import org.jbpm.process.instance.impl.humantask.HumanTaskWorkItemImpl;
+import org.jbpm.process.instance.impl.humantask.InternalHumanTaskWorkItem;
import org.jbpm.process.instance.impl.humantask.phases.Claim;
import org.jbpm.process.instance.impl.humantask.phases.Release;
import org.jbpm.process.instance.impl.humantask.phases.Skip;
@@ -104,7 +104,7 @@ public class CustomHumanTaskLifeCycle implements
LifeCycle<Map<String, Object>>
logger.debug("Transition method invoked for work item {} to transition
to {}, currently in phase {} and status {}", workItem.getStringId(),
transition.phase(), workItem.getPhaseId(),
workItem.getPhaseStatus());
- HumanTaskWorkItemImpl humanTaskWorkItem = (HumanTaskWorkItemImpl)
workItem;
+ InternalHumanTaskWorkItem humanTaskWorkItem =
(InternalHumanTaskWorkItem) workItem;
LifeCyclePhase targetPhase = phases.get(transition.phase());
if (targetPhase == null) {
@@ -144,7 +144,6 @@ public class CustomHumanTaskLifeCycle implements
LifeCycle<Map<String, Object>>
@Override
public Map<String, Object> data(KogitoWorkItem workItem) {
-
- return ((HumanTaskWorkItemImpl) workItem).getResults();
+ return ((InternalHumanTaskWorkItem) workItem).getResults();
}
}
diff --git
a/kogito-springboot-examples/process-outbox-mongodb-springboot/src/test/java/org/kie/kogito/springboot/outbox/OutboxIT.java
b/kogito-springboot-examples/process-outbox-mongodb-springboot/src/test/java/org/kie/kogito/springboot/outbox/OutboxIT.java
index 0aea63a4b..e11f31ef3 100644
---
a/kogito-springboot-examples/process-outbox-mongodb-springboot/src/test/java/org/kie/kogito/springboot/outbox/OutboxIT.java
+++
b/kogito-springboot-examples/process-outbox-mongodb-springboot/src/test/java/org/kie/kogito/springboot/outbox/OutboxIT.java
@@ -57,7 +57,6 @@ public class OutboxIT {
private static final String PROCESS_EVENTS_TOPIC =
"kogito-processinstances-events";
private static final String USERTASK_EVENTS_TOPIC =
"kogito-usertaskinstances-events";
- private static final String VARIABLE_EVENTS_TOPIC =
"kogito-variables-events";
private static final int KOGITO_PORT = 8080;
private static final int KAFKA_PORT = 9092;
private static final int DEBEZIUM_PORT = 8083;
@@ -156,18 +155,25 @@ public class OutboxIT {
CountDownLatch processEventCounter = new CountDownLatch(2);
CountDownLatch userTaskEventCounter = new CountDownLatch(1);
kafkaClient.consume(Set.of(PROCESS_EVENTS_TOPIC,
USERTASK_EVENTS_TOPIC), message -> {
+ LOGGER.info("ProcessInstanceVariableDataEvent: {}", message);
String type = JsonPath.read(message, "$.type");
- if ("ProcessInstanceEvent".equals(type)) {
- String orderNumber = JsonPath.read(message,
"$.data.variables.order.orderNumber");
- boolean shipped = JsonPath.read(message,
"$.data.variables.order.shipped");
- if ("23570".equals(orderNumber) && !shipped) {
- processEventCounter.countDown();
+ if ("ProcessInstanceVariableDataEvent".equals(type)) {
+ String varName = JsonPath.read(message, "$.data.variableName");
+ if ("order".equals(varName)) {
+ String orderNumber = JsonPath.read(message,
"$.data.variableValue.orderNumber");
+ boolean shipped = JsonPath.read(message,
"$.data.variableValue.shipped");
+ if ("23570".equals(orderNumber) && !shipped) {
+ processEventCounter.countDown();
+ }
}
- } else if ("UserTaskInstanceEvent".equals(type)) {
- String orderNumber = JsonPath.read(message,
"$.data.inputs.input1.orderNumber");
- boolean shipped = JsonPath.read(message,
"$.data.inputs.input1.shipped");
- if ("23570".equals(orderNumber) && !shipped) {
- userTaskEventCounter.countDown();
+ } else if ("UserTaskInstanceVariableDataEvent".equals(type)) {
+ String varName = JsonPath.read(message, "$.data.variableName");
+ if ("input1".equals(varName)) {
+ String orderNumber = JsonPath.read(message,
"$.data.variableValue.orderNumber");
+ boolean shipped = JsonPath.read(message,
"$.data.variableValue.shipped");
+ if ("23570".equals(orderNumber) && !shipped) {
+ userTaskEventCounter.countDown();
+ }
}
}
});
@@ -195,8 +201,8 @@ public class OutboxIT {
.get("/connectors/{connector}/topics")
.then()
.statusCode(200)
- .body("kogito-connector.topics", hasSize(3))
- .body("kogito-connector.topics",
hasItems(PROCESS_EVENTS_TOPIC, USERTASK_EVENTS_TOPIC, VARIABLE_EVENTS_TOPIC)));
+ .body("kogito-connector.topics", hasSize(2))
+ .body("kogito-connector.topics",
hasItems(PROCESS_EVENTS_TOPIC, USERTASK_EVENTS_TOPIC)));
// Check process events pushed
assertTrue(processEventCounter.await(TIMEOUT.getSeconds(),
TimeUnit.SECONDS));
diff --git
a/kogito-springboot-examples/process-usertasks-custom-lifecycle-springboot/src/main/java/org/acme/travels/usertasks/CustomHumanTaskLifeCycle.java
b/kogito-springboot-examples/process-usertasks-custom-lifecycle-springboot/src/main/java/org/acme/travels/usertasks/CustomHumanTaskLifeCycle.java
index 49c51e05e..d5c071561 100644
---
a/kogito-springboot-examples/process-usertasks-custom-lifecycle-springboot/src/main/java/org/acme/travels/usertasks/CustomHumanTaskLifeCycle.java
+++
b/kogito-springboot-examples/process-usertasks-custom-lifecycle-springboot/src/main/java/org/acme/travels/usertasks/CustomHumanTaskLifeCycle.java
@@ -23,7 +23,7 @@ import java.util.LinkedHashMap;
import java.util.Map;
import org.jbpm.process.instance.impl.humantask.BaseHumanTaskLifeCycle;
-import org.jbpm.process.instance.impl.humantask.HumanTaskWorkItemImpl;
+import org.jbpm.process.instance.impl.humantask.InternalHumanTaskWorkItem;
import org.jbpm.process.instance.impl.humantask.phases.Claim;
import org.jbpm.process.instance.impl.humantask.phases.Release;
import org.jbpm.process.instance.impl.humantask.phases.Skip;
@@ -104,7 +104,7 @@ public class CustomHumanTaskLifeCycle implements
LifeCycle<Map<String, Object>>
logger.debug("Transition method invoked for work item {} to transition
to {}, currently in phase {} and status {}", workItem.getStringId(),
transition.phase(), workItem.getPhaseId(),
workItem.getPhaseStatus());
- HumanTaskWorkItemImpl humanTaskWorkItem = (HumanTaskWorkItemImpl)
workItem;
+ InternalHumanTaskWorkItem humanTaskWorkItem =
(InternalHumanTaskWorkItem) workItem;
LifeCyclePhase targetPhase = phases.get(transition.phase());
if (targetPhase == null) {
@@ -144,7 +144,6 @@ public class CustomHumanTaskLifeCycle implements
LifeCycle<Map<String, Object>>
@Override
public Map<String, Object> data(KogitoWorkItem workItem) {
-
- return ((HumanTaskWorkItemImpl) workItem).getResults();
+ return ((InternalHumanTaskWorkItem) workItem).getResults();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
