5.5 Define Events
Define the events used.
In the previous section we have configured the tables for our event handling. Now it’s time to define our actual events for the application.
Event overview
We will build the following event flow. Keep in mind that messages are not directly written to Kafka.
Define Events
To define our events we have to implement ExportedEvent. This event implementation is responsible to fill up our outboxevent table. The definition of the values is as follows:
| Field | Description |
|---|
id | unique id. Can be used by consumers to detect duplicates. Should be generated when creating the event Object. |
aggregatetype | the origin of a given event. It could be the microservice name or for example in case of domain-driven design the origin domain. For example “order” or “stock”. |
aggregateid | contains the id affected by a given event. In case of the aggregatetype order this could be the order id. This id is used as the key for Kafka messages. Only messages with the same key are guaranteed to be processed in order as they are placed in the same partition of a kafka topic. |
type | the actual type of the event. For example “OrderCreated” or “StockComplete”. We will use this information to trigger specific event handlers on the consumer side. |
payload | in our example a JSON structure which contains the event body. For example the actual Order with Article id and Amount ordered. |
Task 5.5.1 - Implement OrderCreated Event
- Given the information above. Complete the
ch.puzzle.mm.debezium.event.entity.OrderCreatedEvent event implementation.- Set the required fields accordingly (
AGGREGATE_TYPE and EVENT_TYPE)
For help, here’s an excerpt of the order outboxevent table. For simplicity the timestamp is omitted.
1
2
3
4
5
6
7
| admin=# select id, aggregatetype, aggregateid, type, payload from outboxevent;
id | aggregatetype | aggregateid | type | payload
-----------+---------------+-------------+----------------+---------------------------------------------------------
bed8bd... | Order | 100000 | OrderCreated | {"orderId":100000,"items":[{"articleId":1,"amount":9}]}
28e777... | Order | 100002 | OrderCreated | {"orderId":100002,"items":[{"articleId":1,"amount":3}]}
e09acc... | Order | 100000 | OrderCancelled | {"orderId":100000,"items":[{"articleId":1,"amount":9}]}
3709f2... | Order | 100004 | OrderCreated | {"orderId":100004,"items":[{"articleId":1,"amount":3}]}
|
Task Hint
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
| public class OrderCreatedEvent implements ExportedEvent<String, JsonNode> {
private static ObjectMapper mapper = new ObjectMapper();
private static final String AGGREGATE_TYPE = "Order";
private static final String EVENT_TYPE = "OrderCreated";
private final UUID id;
private final Long orderId;
private final JsonNode jsonNode;
private final Instant timestamp;
public OrderCreatedEvent(Instant created, ShopOrder shopOrder) {
this.id = UUID.randomUUID();
this.orderId = shopOrder.id;
this.timestamp = created;
this.jsonNode = asJson(shopOrder);
}
@Override
public String getAggregateId() {
return String.valueOf(orderId);
}
@Override
public String getAggregateType() {
return AGGREGATE_TYPE;
}
@Override
public String getType() {
return EVENT_TYPE;
}
@Override
public Instant getTimestamp() {
return timestamp;
}
@Override
public JsonNode getPayload() {
return jsonNode;
}
public ObjectNode asJson(ShopOrder order) {
ObjectNode asJson = mapper.createObjectNode()
.put("orderId", order.id);
ArrayNode items = asJson.putArray("items");
for (ArticleOrder article : order.getArticleOrders()) {
items.add(mapper.createObjectNode()
.put("articleId", article.getArticleId())
.put("amount", article.getAmount())
);
}
return asJson;
}
}
|