From 6d64dff7c410dd644e42a25b0a67513c3b30cfd3 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Mon, 15 Nov 2021 21:41:03 +0100 Subject: [PATCH] WIP --- .../transfer/TransferServiceApplication.java | 18 ++++++----- .../transfer/adapter/TransferConsumer.java | 2 +- .../transfer/ports/CreateTransferUseCase.java | 6 ++-- .../ports/HandleStateChangeUseCase.java | 4 +-- .../transfer/ports/TransferService.java | 32 +++++++++---------- 5 files changed, 33 insertions(+), 29 deletions(-) diff --git a/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java b/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java index a5c1faf..e94f703 100644 --- a/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java +++ b/src/main/java/de/juplo/kafka/payment/transfer/TransferServiceApplication.java @@ -113,9 +113,9 @@ public class TransferServiceApplication mapper, new TransferConsumer.ConsumerUseCases() { @Override - public void create(Long id, Long payer, Long payee, Integer amount) + public TransferStateChangedEvent create(Long id, Long payer, Long payee, Integer amount) { - productionTransferService.create(id, payer, payee, amount); + return productionTransferService.create(id, payer, payee, amount); } @Override @@ -125,16 +125,17 @@ public class TransferServiceApplication } @Override - public void handleStateChange(Long id, Transfer.State state) + public TransferStateChangedEvent handleStateChange( + TransferStateChangedEvent stateChangedEvent) { - productionTransferService.handleStateChange(id, state); + return productionTransferService.handleStateChange(stateChangedEvent); } }, new TransferConsumer.ConsumerUseCases() { @Override - public void create(Long id, Long payer, Long payee, Integer amount) + public TransferStateChangedEvent create(Long id, Long payer, Long payee, Integer amount) { - restoreTransferService.create(id, payer, payee, amount); + return restoreTransferService.create(id, payer, payee, amount); } @Override @@ -144,9 +145,10 @@ public class TransferServiceApplication } @Override - public void handleStateChange(Long id, Transfer.State state) + public TransferStateChangedEvent handleStateChange( + TransferStateChangedEvent stateChangedEvent) { - restoreTransferService.handleStateChange(id, state); + return restoreTransferService.handleStateChange(stateChangedEvent); } }); } diff --git a/src/main/java/de/juplo/kafka/payment/transfer/adapter/TransferConsumer.java b/src/main/java/de/juplo/kafka/payment/transfer/adapter/TransferConsumer.java index 1cae540..c66c534 100644 --- a/src/main/java/de/juplo/kafka/payment/transfer/adapter/TransferConsumer.java +++ b/src/main/java/de/juplo/kafka/payment/transfer/adapter/TransferConsumer.java @@ -172,7 +172,7 @@ public class TransferConsumer implements Runnable, ConsumerRebalanceListener TransferStateChangedEvent stateChangedEvent = mapper.readValue(record.value(), TransferStateChangedEvent.class); - useCases.handleStateChange(stateChangedEvent.getId(), stateChangedEvent.getState()); + useCases.handleStateChange(stateChangedEvent); break; } } diff --git a/src/main/java/de/juplo/kafka/payment/transfer/ports/CreateTransferUseCase.java b/src/main/java/de/juplo/kafka/payment/transfer/ports/CreateTransferUseCase.java index bfe6156..8591a0d 100644 --- a/src/main/java/de/juplo/kafka/payment/transfer/ports/CreateTransferUseCase.java +++ b/src/main/java/de/juplo/kafka/payment/transfer/ports/CreateTransferUseCase.java @@ -1,9 +1,11 @@ package de.juplo.kafka.payment.transfer.ports; -import de.juplo.kafka.payment.transfer.domain.Transfer; +import de.juplo.kafka.payment.transfer.adapter.TransferStateChangedEvent; + +import java.util.Optional; public interface CreateTransferUseCase { - void create(Long id, Long payer, Long payee, Integer amount); + Optional create(Long id, Long payer, Long payee, Integer amount); } diff --git a/src/main/java/de/juplo/kafka/payment/transfer/ports/HandleStateChangeUseCase.java b/src/main/java/de/juplo/kafka/payment/transfer/ports/HandleStateChangeUseCase.java index 9b3e270..7d9fe83 100644 --- a/src/main/java/de/juplo/kafka/payment/transfer/ports/HandleStateChangeUseCase.java +++ b/src/main/java/de/juplo/kafka/payment/transfer/ports/HandleStateChangeUseCase.java @@ -1,9 +1,9 @@ package de.juplo.kafka.payment.transfer.ports; -import de.juplo.kafka.payment.transfer.domain.Transfer; +import de.juplo.kafka.payment.transfer.adapter.TransferStateChangedEvent; public interface HandleStateChangeUseCase { - void handleStateChange(Long id, Transfer.State state); + TransferStateChangedEvent handleStateChange(TransferStateChangedEvent stateChangedEvent); } diff --git a/src/main/java/de/juplo/kafka/payment/transfer/ports/TransferService.java b/src/main/java/de/juplo/kafka/payment/transfer/ports/TransferService.java index d4eb3ef..7ee0432 100644 --- a/src/main/java/de/juplo/kafka/payment/transfer/ports/TransferService.java +++ b/src/main/java/de/juplo/kafka/payment/transfer/ports/TransferService.java @@ -1,6 +1,7 @@ package de.juplo.kafka.payment.transfer.ports; +import de.juplo.kafka.payment.transfer.adapter.TransferStateChangedEvent; import de.juplo.kafka.payment.transfer.domain.Transfer; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -19,36 +20,35 @@ public class TransferService implements CreateTransferUseCase, HandleStateChange private final MessagingService messagingService; @Override - public void create(Long id, Long payer, Long payee, Integer amount) + public Optional create(Long id, Long payer, Long payee, Integer amount) { + return repository .get(id) - .ifPresentOrElse( - stored -> log.info( + .flatMap( + stored -> + {log.info( "transfer already exisits: {}, ignoring: id={}, payer={}, payee={}, amount={}", stored, payer, payee, - amount), - () -> + amount); + return Optional.empty(); + }) + .or(() -> { - Transfer transfer = - Transfer + log.info("creating transfer: {}", transfer); + return + TransferStateChangedEvent .builder() - .id(id) - .payer(payer) - .payee(payee) - .amount(amount) + .id(transfer.getId()) + .state(CREATED) .build(); - - log.info("creating transfer: {}", transfer); - repository.store(transfer); - messagingService.send(transfer.getId(), CREATED); }); } @Override - public void handleStateChange(Long id, Transfer.State state) + public TransferStateChangedEvent handleStateChange(Long id, Transfer.State state) { get(id) .ifPresentOrElse( -- 2.20.1