Skip to content
This repository has been archived by the owner on Jul 26, 2024. It is now read-only.

Commit

Permalink
Introduce mapper for CustomerEventRecord, extend producer to also sen…
Browse files Browse the repository at this point in the history
…d REPLACE- and DELETE-Events.
  • Loading branch information
Ralf Ueberfuhr committed Jun 27, 2024
1 parent 6fe01a5 commit 5fa738f
Show file tree
Hide file tree
Showing 2 changed files with 82 additions and 13 deletions.
Original file line number Diff line number Diff line change
@@ -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
);
}

}
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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,
Expand All @@ -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
);
}

}

0 comments on commit 5fa738f

Please sign in to comment.