From 5fa738f0e39ed58be9879c7f607de43f478f799d Mon Sep 17 00:00:00 2001
From: Ralf Ueberfuhr <ralf.ueberfuhr@ars.de>
Date: Thu, 27 Jun 2024 16:53:54 +0200
Subject: [PATCH] Introduce mapper for CustomerEventRecord, extend producer to
 also send REPLACE- and DELETE-Events.

---
 .../kafka/CustomerEventRecordMapper.java      | 55 +++++++++++++++++++
 .../kafka/CustomerEventsProducer.java         | 40 +++++++++-----
 2 files changed, 82 insertions(+), 13 deletions(-)
 create mode 100644 account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/CustomerEventRecordMapper.java

diff --git a/account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/CustomerEventRecordMapper.java b/account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/CustomerEventRecordMapper.java
new file mode 100644
index 0000000..043ec8c
--- /dev/null
+++ b/account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/CustomerEventRecordMapper.java
@@ -0,0 +1,55 @@
+package de.sample.schulung.accounts.kafka;
+
+import de.sample.schulung.accounts.domain.Customer;
+import de.sample.schulung.accounts.domain.Customer.CustomerState;
+import de.sample.schulung.accounts.domain.events.CustomerCreatedEvent;
+import de.sample.schulung.accounts.domain.events.CustomerDeletedEvent;
+import de.sample.schulung.accounts.domain.events.CustomerReplacedEvent;
+import org.springframework.stereotype.Component;
+
+@Component
+public class CustomerEventRecordMapper {
+
+  public String map(CustomerState state) {
+    return switch (state) {
+      case ACTIVE -> "active";
+      case LOCKED -> "locked";
+      case DISABLED -> "disabled";
+    };
+  }
+
+  public CustomerRecord map(Customer customer) {
+    return new CustomerRecord(
+      customer.getName(),
+      customer.getDateOfBirth(),
+      this.map(customer.getState())
+    );
+  }
+
+  public CustomerEventRecord map(CustomerCreatedEvent event) {
+    var customer = event.customer();
+    return new CustomerEventRecord(
+      "created",
+      customer.getUuid(),
+      this.map(customer)
+    );
+  }
+
+  public CustomerEventRecord map(CustomerReplacedEvent event) {
+    var customer = event.customer();
+    return new CustomerEventRecord(
+      "replaced",
+      customer.getUuid(),
+      this.map(customer)
+    );
+  }
+
+  public CustomerEventRecord map(CustomerDeletedEvent event) {
+    return new CustomerEventRecord(
+      "deleted",
+      event.uuid(),
+      null
+    );
+  }
+
+}
diff --git a/account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/CustomerEventsProducer.java b/account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/CustomerEventsProducer.java
index 569f64f..44664e7 100644
--- a/account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/CustomerEventsProducer.java
+++ b/account-service-provider/src/main/java/de/sample/schulung/accounts/kafka/CustomerEventsProducer.java
@@ -1,6 +1,8 @@
 package de.sample.schulung.accounts.kafka;
 
 import de.sample.schulung.accounts.domain.events.CustomerCreatedEvent;
+import de.sample.schulung.accounts.domain.events.CustomerDeletedEvent;
+import de.sample.schulung.accounts.domain.events.CustomerReplacedEvent;
 import lombok.RequiredArgsConstructor;
 import org.springframework.context.event.EventListener;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -13,24 +15,24 @@
 public class CustomerEventsProducer {
 
   private final KafkaTemplate<UUID, Object> kafkaTemplate;
+  private final CustomerEventRecordMapper mapper;
 
   @EventListener
   public void handleCustomerCreatedEvent(CustomerCreatedEvent event) {
     // map event to record
-    var customer = event.customer();
-    var payload = new CustomerEventRecord(
-      "created",
-      customer.getUuid(),
-      new CustomerRecord(
-        customer.getName(),
-        customer.getDateOfBirth(),
-        switch(customer.getState()) {
-          case ACTIVE -> "active";
-          case LOCKED -> "locked";
-          case DISABLED -> "disabled";
-        }
-      )
+    var payload = this.mapper.map(event);
+    // send message
+    kafkaTemplate.send(
+      KafkaConstants.CUSTOMER_EVENTS_TOPIC,
+      event.customer().getUuid(),
+      payload
     );
+  }
+
+  @EventListener
+  public void handleCustomerReplacedEvent(CustomerReplacedEvent event) {
+    // map event to record
+    var payload = this.mapper.map(event);
     // send message
     kafkaTemplate.send(
       KafkaConstants.CUSTOMER_EVENTS_TOPIC,
@@ -39,4 +41,16 @@ public void handleCustomerCreatedEvent(CustomerCreatedEvent event) {
     );
   }
 
+  @EventListener
+  public void handleCustomerDeletedEvent(CustomerDeletedEvent event) {
+    // map event to record
+    var payload = this.mapper.map(event);
+    // send message
+    kafkaTemplate.send(
+      KafkaConstants.CUSTOMER_EVENTS_TOPIC,
+      event.uuid(),
+      payload
+    );
+  }
+
 }