This is an automated email from the ASF dual-hosted git repository.
pefernan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-kie.git
The following commit(s) were added to refs/heads/main by this push:
new d61ec52cae6 [incubator-kie#7189] Message-correlated signals sent
through `ProcessEventDispatcher` don't publish DataEvents (#7190)
d61ec52cae6 is described below
commit d61ec52cae68fc2bfc112e9cd4da5d66c5ff4f27
Author: Pere Fernández <[email protected]>
AuthorDate: Thu Oct 15 13:07:20 2026 +0200
[incubator-kie#7189] Message-correlated signals sent through
`ProcessEventDispatcher` don't publish DataEvents (#7190)
---
.../org/kie/kogito/process/ProcessService.java | 5 +++
.../kogito/event/impl/ProcessEventDispatcher.java | 3 +-
.../event/impl/ProcessEventDispatcherTest.java | 10 ++++++
.../kogito/process/impl/ProcessServiceImpl.java | 11 ++++++
.../process/impl/ProcessServiceImplSignalTest.java | 42 +++++++++++++++++++++-
5 files changed, 68 insertions(+), 3 deletions(-)
diff --git
a/kogito-api/kogito-api/src/main/java/org/kie/kogito/process/ProcessService.java
b/kogito-api/kogito-api/src/main/java/org/kie/kogito/process/ProcessService.java
index 42585a783bc..6cf803966a3 100644
---
a/kogito-api/kogito-api/src/main/java/org/kie/kogito/process/ProcessService.java
+++
b/kogito-api/kogito-api/src/main/java/org/kie/kogito/process/ProcessService.java
@@ -63,6 +63,11 @@ public interface ProcessService {
<T extends MappableToModel<R>, R> Optional<R>
signalProcessInstance(Process<T> process, String id, Object data, String
signalName);
+ /**
+ * Sends a signal to all the instances of the given process waiting for it.
+ */
+ <T extends Model> void signalProcess(Process<T> process, Object data,
String signalName);
+
<T> void migrateProcessInstances(Process<T> process, String
targetProcessId, String targetProcessVersion, String... id) throws
UnsupportedOperationException;
<T> long migrateAll(Process<T> process, String targetProcessId, String
targetProcessVersion) throws UnsupportedOperationException;
diff --git
a/kogito-api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/ProcessEventDispatcher.java
b/kogito-api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/ProcessEventDispatcher.java
index 349de3a9b04..50f1e052140 100644
---
a/kogito-api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/ProcessEventDispatcher.java
+++
b/kogito-api/kogito-events-core/src/main/java/org/kie/kogito/event/impl/ProcessEventDispatcher.java
@@ -38,7 +38,6 @@ import org.kie.kogito.event.EventDispatcher;
import org.kie.kogito.process.Process;
import org.kie.kogito.process.ProcessInstance;
import org.kie.kogito.process.ProcessService;
-import org.kie.kogito.process.SignalFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -94,7 +93,7 @@ public class ProcessEventDispatcher<M extends Model, D>
implements EventDispatch
ProcessInstance<M> processInstance = startNewInstance(trigger, event);
// we signal all the processes waiting for trigger (this covers
intermediate catch events)
LOGGER.debug("sending event to process {} with trigger {} and payload
{}", process.id(), trigger, data);
- process.send(SignalFactory.of("Message-" + trigger, data));
+ processService.signalProcess(process, data, "Message-" + trigger);
return processInstance;
}
diff --git
a/kogito-api/kogito-events-core/src/test/java/org/kie/kogito/event/impl/ProcessEventDispatcherTest.java
b/kogito-api/kogito-events-core/src/test/java/org/kie/kogito/event/impl/ProcessEventDispatcherTest.java
index e6326dbc9a8..fe4fcc590cf 100644
---
a/kogito-api/kogito-events-core/src/test/java/org/kie/kogito/event/impl/ProcessEventDispatcherTest.java
+++
b/kogito-api/kogito-events-core/src/test/java/org/kie/kogito/event/impl/ProcessEventDispatcherTest.java
@@ -169,6 +169,16 @@ class ProcessEventDispatcherTest {
assertThat(processInstance).isEqualTo(instance);
}
+ @Test
+ void testDataEventWithoutTargetSignalsProcessThroughService() throws
Exception {
+ EventDispatcher<DummyModel, TestEvent> dispatcher = new
ProcessEventDispatcher<>(process, Optional.empty(), processService, null, o ->
o.getData());
+ TestEvent data = new TestEvent("pepe");
+ dispatcher.dispatch(DUMMY_TOPIC, DataEventFactory.from(data));
+
+ verify(processService, times(1)).signalProcess(process, data,
"Message-" + DUMMY_TOPIC);
+ verify(process, never()).send(any());
+ }
+
@Test
void testIgnoredCloudEvent() throws Exception {
EventDispatcher<DummyModel, TestEvent> dispatcher = new
ProcessEventDispatcher<>(process, modelConverter(), processService, null, o ->
o.getData());
diff --git
a/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/ProcessServiceImpl.java
b/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/ProcessServiceImpl.java
index abf3f87bb36..c189f547dc1 100644
---
a/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/ProcessServiceImpl.java
+++
b/kogito-jbpm/jbpm-flow/src/main/java/org/kie/kogito/process/impl/ProcessServiceImpl.java
@@ -180,6 +180,17 @@ public class ProcessServiceImpl implements ProcessService {
}));
}
+ @Override
+ public <T extends Model> void signalProcess(Process<T> process, Object
data, String signalName) {
+ Objects.requireNonNull(signalName, "signalName must not be null");
+ UnitOfWorkExecutor.executeInUnitOfWork(
+ application.unitOfWorkManager(),
+ () -> {
+ process.send(SignalFactory.of(signalName, data));
+ return null;
+ });
+ }
+
@Override
public <T extends Model> Optional<List<WorkItem>> getWorkItems(Process<T>
process, String id, Policy... policy) {
return process.instances()
diff --git
a/kogito-jbpm/jbpm-flow/src/test/java/org/kie/kogito/process/impl/ProcessServiceImplSignalTest.java
b/kogito-jbpm/jbpm-flow/src/test/java/org/kie/kogito/process/impl/ProcessServiceImplSignalTest.java
index 3b7c6e735d2..6e85f9b6c42 100644
---
a/kogito-jbpm/jbpm-flow/src/test/java/org/kie/kogito/process/impl/ProcessServiceImplSignalTest.java
+++
b/kogito-jbpm/jbpm-flow/src/test/java/org/kie/kogito/process/impl/ProcessServiceImplSignalTest.java
@@ -34,14 +34,20 @@ import org.kie.kogito.process.EventDescription;
import org.kie.kogito.process.Process;
import org.kie.kogito.process.ProcessInstance;
import org.kie.kogito.process.ProcessInstances;
+import org.kie.kogito.process.Signal;
import org.kie.kogito.process.flexible.AdHocFragment;
import org.kie.kogito.uow.UnitOfWorkManager;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -53,6 +59,7 @@ import static org.mockito.Mockito.when;
class ProcessServiceImplSignalTest {
private ProcessServiceImpl processService;
+ private org.kie.kogito.uow.UnitOfWork unitOfWork;
private Process<TestModel> process;
private ProcessInstances<TestModel> processInstances;
private ProcessInstance<TestModel> processInstance;
@@ -74,7 +81,7 @@ class ProcessServiceImplSignalTest {
when(process.instances()).thenReturn(processInstances);
// Make the UoW execute the supplied callable immediately
- org.kie.kogito.uow.UnitOfWork unitOfWork =
mock(org.kie.kogito.uow.UnitOfWork.class);
+ unitOfWork = mock(org.kie.kogito.uow.UnitOfWork.class);
when(unitOfWorkManager.newUnitOfWork()).thenReturn(unitOfWork);
when(unitOfWorkManager.currentUnitOfWork()).thenReturn(unitOfWork);
doAnswer(invocation -> {
@@ -209,6 +216,39 @@ class ProcessServiceImplSignalTest {
assertThat(result).isEmpty();
}
+ // --- process level signal tests ---
+
+ @Test
+ void signalProcess_sendsSignalWithinUnitOfWork() {
+ processService.signalProcess(process, "data", "Message-test");
+
+ ArgumentCaptor<Signal> signal = ArgumentCaptor.forClass(Signal.class);
+ InOrder order = inOrder(unitOfWork, process);
+ order.verify(unitOfWork).start();
+ order.verify(process).send(signal.capture());
+ order.verify(unitOfWork).end();
+ assertThat(signal.getValue().channel()).isEqualTo("Message-test");
+ assertThat(signal.getValue().payload()).isEqualTo("data");
+ }
+
+ @Test
+ void signalProcess_abortsUnitOfWorkWhenSendFails() {
+ doThrow(new IllegalStateException("boom")).when(process).send(any());
+
+ assertThatThrownBy(() -> processService.signalProcess(process, "data",
"Message-test"))
+ .isInstanceOf(IllegalStateException.class);
+
+ verify(unitOfWork).abort();
+ verify(unitOfWork, never()).end();
+ }
+
+ @Test
+ void signalProcess_throwsWhenSignalNameIsNull() {
+ assertThatThrownBy(() -> processService.signalProcess(process, "data",
null))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("signalName");
+ }
+
// --- test model ---
static class TestModel implements MappableToModel<TestModel>, Model {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
