getEvent(String id) {
+ return eventRepository.findById(id).map(Mapper::map);
+ }
+
+ private static EventEntity entity(StateName state) {
+ EventEntity entity = new EventEntity();
+ entity.setState(map(state));
+ return entity;
+ }
+
+ public Event updateEventStateToPublishing(String id) {
+ EventEntity eventEntity = eventRepository.findById(id).orElseThrow();
+ eventEntity.setPublishingAt(OffsetDateTime.now());
+
+ eventEntity.setState(State.PUBLISHING);
+ eventRepository.save(eventEntity);
+ return map(eventEntity);
+ }
+
+ public Event updateEventStateToPublished(String id) {
+ EventEntity eventEntity = eventRepository.findById(id).orElseThrow();
+ eventEntity.setPublishedAt(OffsetDateTime.now());
+
+ eventEntity.setState(State.PUBLISHED);
+ eventRepository.save(eventEntity);
+ return map(eventEntity);
+ }
+
+ public record UnsavedEvent(String id, String name, ObjectNode content) {}
+
+ public record Event(String id, String name, ObjectNode content, State state) {
+ public record State(
+ StateName name,
+ OffsetDateTime createdAt,
+ OffsetDateTime publishingAt,
+ OffsetDateTime publishedAt) {
+ public enum StateName {
+ CREATED,
+ PUBLISHING,
+ PUBLISHED
+ }
+ }
+ }
+}
diff --git a/event-processing/src/main/java/com/camunda/consulting/eventprocessing/ExampleServiceCalls.java b/event-processing/src/main/java/com/camunda/consulting/eventprocessing/ExampleServiceCalls.java
new file mode 100644
index 00000000..b442fa9b
--- /dev/null
+++ b/event-processing/src/main/java/com/camunda/consulting/eventprocessing/ExampleServiceCalls.java
@@ -0,0 +1,26 @@
+package com.camunda.consulting.eventprocessing;
+
+import io.camunda.zeebe.spring.client.annotation.JobWorker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ExampleServiceCalls {
+ private static final Logger LOG = LoggerFactory.getLogger(ExampleServiceCalls.class);
+
+ @JobWorker
+ public void service1() {
+ LOG.info("Service 1 called");
+ }
+
+ @JobWorker
+ public void service2() {
+ LOG.info("Service 2 called");
+ }
+
+ @JobWorker
+ public void service3() {
+ LOG.info("Service 3 called");
+ }
+}
diff --git a/event-processing/src/main/java/com/camunda/consulting/eventprocessing/Mapper.java b/event-processing/src/main/java/com/camunda/consulting/eventprocessing/Mapper.java
new file mode 100644
index 00000000..53e1df7a
--- /dev/null
+++ b/event-processing/src/main/java/com/camunda/consulting/eventprocessing/Mapper.java
@@ -0,0 +1,188 @@
+package com.camunda.consulting.eventprocessing;
+
+import com.camunda.consulting.eventprocessing.EventController.CreateEventRequest;
+import com.camunda.consulting.eventprocessing.EventController.CreateEventResponse;
+import com.camunda.consulting.eventprocessing.EventController.GetEventResponse;
+import com.camunda.consulting.eventprocessing.EventRepository.EventEntity;
+import com.camunda.consulting.eventprocessing.EventRepository.EventEntity.State;
+import com.camunda.consulting.eventprocessing.EventService.Event;
+import com.camunda.consulting.eventprocessing.EventService.UnsavedEvent;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+import java.time.OffsetDateTime;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+
+/**
+ * Please ignore this class for the example.
+ *
+ * This is just a generic mapper to map dto - pojo - entity as they are separated
+ */
+public class Mapper {
+ private static final Map, Map, Function>> MAPPER_REGISTRY =
+ new HashMap<>();
+ private static final ObjectMapper objectMapper = new ObjectMapper();
+
+ static {
+ Arrays.stream(Mapper.class.getDeclaredMethods())
+ .filter(m -> Modifier.isStatic(m.getModifiers()))
+ .filter(m -> Modifier.isPrivate(m.getModifiers()))
+ .filter(m -> Objects.nonNull(m.getAnnotation(IsMapper.class)))
+ .forEach(Mapper::process);
+ }
+
+ private static void process(Method method) {
+ // only 1 parameter allowed
+ if (method.getParameterCount() != 1) {
+ throw new RuntimeException("Method must have exactly one parameter");
+ }
+ Class sourceType = method.getParameterTypes()[0];
+ Class targetType = method.getReturnType();
+ Map, Function> targetMap =
+ MAPPER_REGISTRY.computeIfAbsent(sourceType, (c) -> new HashMap<>());
+ if (targetMap.containsKey(targetType)) {
+ throw new RuntimeException(
+ "There are 2 functions defined for mapping " + sourceType + " to " + targetType);
+ }
+ targetMap.put(
+ targetType,
+ (s) -> {
+ try {
+ return method.invoke(null, s);
+ } catch (IllegalAccessException | InvocationTargetException e) {
+ throw new RuntimeException("Error while invoking mapping method", e);
+ }
+ });
+ }
+
+ @SafeVarargs
+ public static T map(S event, T... omit) {
+ Class sourceType = event.getClass();
+ Class targetType = omit.getClass().getComponentType();
+ if (MAPPER_REGISTRY.containsKey(sourceType)) {
+ Map, Function> targetMap = MAPPER_REGISTRY.get(sourceType);
+ if (targetMap.containsKey(targetType)) {
+ Function function = (Function) targetMap.get(targetType);
+ return function.apply(event);
+ } else {
+ throw new RuntimeException(
+ "There are no functions defined for mapping " + sourceType + " to " + targetType);
+ }
+ } else {
+ throw new RuntimeException(
+ "There are no functions defined for mapping " + sourceType + " to " + targetType);
+ }
+ }
+
+ @IsMapper
+ private static EventEntity mapToEventEntity(UnsavedEvent event) {
+ EventEntity eventEntity = new EventEntity();
+ eventEntity.setId(event.id());
+ eventEntity.setName(event.name());
+ eventEntity.setContent(mapToString(event.content()));
+ eventEntity.setState(State.CREATED);
+ eventEntity.setCreatedAt(OffsetDateTime.now());
+ return eventEntity;
+ }
+
+ @IsMapper
+ private static String mapToString(JsonNode node) {
+ try {
+ return objectMapper.writeValueAsString(node);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Error while writing json node to string", e);
+ }
+ }
+
+ @IsMapper
+ private static ObjectNode mapToJsonNode(String json) {
+ try {
+ return (ObjectNode) objectMapper.readTree(json);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException("Error while reading string to json node", e);
+ }
+ }
+
+ @IsMapper
+ private static Event mapToEvent(EventEntity eventEntity) {
+ return new Event(
+ eventEntity.getId(),
+ eventEntity.getName(),
+ mapToJsonNode(eventEntity.getContent()),
+ mapToState(eventEntity));
+ }
+
+ @IsMapper
+ private static Event.State mapToState(EventEntity eventEntity) {
+ return new Event.State(
+ mapToStateName(eventEntity.getState()),
+ eventEntity.getCreatedAt(),
+ eventEntity.getPublishingAt(),
+ eventEntity.getPublishedAt());
+ }
+
+ @IsMapper
+ private static Event.State.StateName mapToStateName(EventEntity.State state) {
+ return switch (state) {
+ case CREATED -> Event.State.StateName.CREATED;
+ case PUBLISHED -> Event.State.StateName.PUBLISHED;
+ case PUBLISHING -> Event.State.StateName.PUBLISHING;
+ case null -> null;
+ };
+ }
+
+ @IsMapper
+ private static EventEntity.State mapToState(Event.State.StateName stateName) {
+ return switch (stateName) {
+ case CREATED -> EventEntity.State.CREATED;
+ case PUBLISHED -> EventEntity.State.PUBLISHED;
+ case PUBLISHING -> EventEntity.State.PUBLISHING;
+ case null -> null;
+ };
+ }
+
+ @IsMapper
+ private static UnsavedEvent mapToUnsavedEvent(CreateEventRequest event) {
+ return new UnsavedEvent(event.id(), event.name(), event.content());
+ }
+
+ @IsMapper
+ private static CreateEventResponse mapToCreatedEventResponse(Event event) {
+ return new CreateEventResponse(event.id());
+ }
+
+ @IsMapper
+ private static GetEventResponse mapToGetEventResponse(Event event) {
+ return new GetEventResponse(
+ event.id(), event.name(), event.content(), mapToState(event.state()));
+ }
+
+ @IsMapper
+ private static GetEventResponse.State mapToState(Event.State event) {
+ return new GetEventResponse.State(
+ mapToStateName(event.name()), event.createdAt(), event.publishingAt(), event.publishedAt());
+ }
+
+ @IsMapper
+ private static GetEventResponse.State.StateName mapToStateName(Event.State.StateName state) {
+ return switch (state) {
+ case CREATED -> GetEventResponse.State.StateName.CREATED;
+ case PUBLISHED -> GetEventResponse.State.StateName.PUBLISHED;
+ case PUBLISHING -> GetEventResponse.State.StateName.PUBLISHING;
+ case null -> null;
+ };
+ }
+
+ @Retention(RetentionPolicy.RUNTIME)
+ private @interface IsMapper {}
+}
diff --git a/event-processing/src/main/resources/application.yaml b/event-processing/src/main/resources/application.yaml
new file mode 100644
index 00000000..8b7232a0
--- /dev/null
+++ b/event-processing/src/main/resources/application.yaml
@@ -0,0 +1,8 @@
+spring:
+ datasource:
+ url: jdbc:h2:file:./events
+ jpa:
+ generate-ddl: true
+camunda:
+ client:
+ mode: selfmanaged
\ No newline at end of file
diff --git a/event-processing/src/test/java/com/camunda/consulting/eventprocessing/ProcessTest.java b/event-processing/src/test/java/com/camunda/consulting/eventprocessing/ProcessTest.java
new file mode 100644
index 00000000..7247f529
--- /dev/null
+++ b/event-processing/src/test/java/com/camunda/consulting/eventprocessing/ProcessTest.java
@@ -0,0 +1,60 @@
+package com.camunda.consulting.eventprocessing;
+
+import static org.assertj.core.api.Assertions.*;
+
+import com.camunda.consulting.eventprocessing.EventController.CreateEventRequest;
+import com.camunda.consulting.eventprocessing.EventController.CreateEventResponse;
+import com.camunda.consulting.eventprocessing.EventService.Event.State.StateName;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.camunda.process.test.api.CamundaSpringProcessTest;
+import io.camunda.zeebe.client.ZeebeClient;
+import java.io.IOException;
+import java.io.InputStream;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+@CamundaSpringProcessTest
+public class ProcessTest {
+ @Autowired ZeebeClient zeebeClient;
+
+ @Autowired EventService eventService;
+
+ @Autowired EventController eventController;
+
+ private static CreateEventRequest loadCreateEventRequest() {
+ try (InputStream stream =
+ ProcessTest.class.getClassLoader().getResourceAsStream("createEventRequest.json")) {
+
+ return new ObjectMapper().readValue(stream, CreateEventRequest.class);
+ } catch (IOException e) {
+ throw new RuntimeException("Error while loading unsaved event", e);
+ }
+ }
+
+ @BeforeEach
+ void deployProcess() {
+ zeebeClient
+ .newDeployResourceCommand()
+ .addResourceFromClasspath("eventHandler.bpmn")
+ .send()
+ .join();
+ }
+
+ @Test
+ void shouldPublishEvent() {
+ // when
+ CreateEventResponse event = eventController.createEvent(loadCreateEventRequest()).getBody();
+ // then
+ Awaitility.await()
+ .untilAsserted(
+ () ->
+ assertThat(eventService.getEvent(event.id()))
+ .isPresent()
+ .get()
+ .matches(e -> e.state().name().equals(StateName.PUBLISHED)));
+ }
+}
diff --git a/event-processing/src/test/resources/application.yaml b/event-processing/src/test/resources/application.yaml
new file mode 100644
index 00000000..bbe10c53
--- /dev/null
+++ b/event-processing/src/test/resources/application.yaml
@@ -0,0 +1,5 @@
+spring:
+ datasource:
+ url: jdbc:h2:mem:test
+ jpa:
+ generate-ddl: true
\ No newline at end of file
diff --git a/event-processing/src/test/resources/createEventRequest.json b/event-processing/src/test/resources/createEventRequest.json
new file mode 100644
index 00000000..410b9ba3
--- /dev/null
+++ b/event-processing/src/test/resources/createEventRequest.json
@@ -0,0 +1,7 @@
+{
+ "id": "123",
+ "name": "random",
+ "content": {
+ "foo": "baz"
+ }
+}
\ No newline at end of file
diff --git a/event-processing/src/test/resources/eventHandler.bpmn b/event-processing/src/test/resources/eventHandler.bpmn
new file mode 100644
index 00000000..b8c9695c
--- /dev/null
+++ b/event-processing/src/test/resources/eventHandler.bpmn
@@ -0,0 +1,98 @@
+
+
+
+
+
+
+
+
+
+ Flow_07icx7o
+
+
+
+ Flow_0zg3kkg
+
+
+
+
+
+
+
+ Flow_07icx7o
+ Flow_0xz040e
+
+
+
+
+
+ Flow_0xz040e
+ Flow_01qhw0g
+
+
+
+
+
+
+ Flow_01qhw0g
+ Flow_0zg3kkg
+
+
+
+ The published ack happens as end listener, can also be a separate service task
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+