Moved all business logic into TransferService

[demos/kafka/demos-kafka-payment-system-transfer] / src / main / java / de / juplo / kafka / payment / transfer / domain / TransferService.java

diff --git a/src/main/java/de/juplo/kafka/payment/transfer/domain/TransferService.java b/src/main/java/de/juplo/kafka/payment/transfer/domain/TransferService.java
index 0cbcd2c..ed1a98e 100644 (file)
--- a/src/main/java/de/juplo/kafka/payment/transfer/domain/TransferService.java
+++ b/src/main/java/de/juplo/kafka/payment/transfer/domain/TransferService.java
@@ -4,66 +4,84 @@ package de.juplo.kafka.payment.transfer.domain;
 import de.juplo.kafka.payment.transfer.ports.*;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.common.TopicPartition;
 
 import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
 
 import static de.juplo.kafka.payment.transfer.domain.Transfer.State.CHECKED;
-import static de.juplo.kafka.payment.transfer.domain.Transfer.State.RECEIVED;
+import static de.juplo.kafka.payment.transfer.domain.Transfer.State.CREATED;
 
 
 @Slf4j
 @RequiredArgsConstructor
-public class TransferService implements ReceiveTransferUseCase, HandleTransferUseCase, GetTransferUseCase
+public class TransferService implements CreateTransferUseCase, HandleStateChangeUseCase, GetTransferUseCase
 {
   private final TransferRepository repository;
   private final MessagingService messagingService;
 
-  public CompletableFuture<TopicPartition> receive(Transfer transfer)
+  @Override
+  public void create(Long id, Long payer, Long payee, Integer amount)
   {
-    transfer.setState(RECEIVED);
-    return messagingService.send(transfer);
+    repository
+        .get(id)
+        .ifPresentOrElse(
+            stored -> log.info(
+                "transfer already exisits: {}, ignoring: id={}, payer={}, payee={}, amount={}",
+                stored,
+                payer,
+                payee,
+                amount),
+            () ->
+            {
+              Transfer transfer =
+                  Transfer
+                      .builder()
+                      .id(id)
+                      .payer(payer)
+                      .payee(payee)
+                      .amount(amount)
+                      .build();
+
+              log.info("creating transfer: {}", transfer);
+              repository.store(transfer);
+              messagingService.send(transfer.getId(), CREATED);
+            });
   }
 
   @Override
-  public void handle(Transfer transfer)
+  public void handleStateChange(Long id, Transfer.State state)
   {
-    Transfer.State state = transfer.getState();
-    switch (state)
-    {
-      case RECEIVED:
-        repository.store(transfer);
-        check(transfer);
-        break;
-
-      case CHECKED:
-        repository.store(transfer);
-        // TODO: What's next...?
-        break;
-
-      default:
-        log.warn("TODO: handle {} state {}", state.foreign ? "foreign" : "domain", state);
-    }
+    get(id)
+        .ifPresentOrElse(
+            transfer ->
+            {
+              switch (state)
+              {
+                case CREATED:
+
+                  transfer.setState(CREATED);
+                  repository.store(transfer);
+                  check(transfer);
+                  break;
+
+                case CHECKED:
+
+                  transfer.setState(CHECKED);
+                  repository.store(transfer);
+                  // TODO: What's next...?
+                  break;
+
+                default:
+
+                  log.warn("TODO: handle {} state {}", state.foreign ? "foreign" : "domain", state);
+              }
+            },
+            () -> log.error("unknown transfer: {}", id));
   }
 
   private void check(Transfer transfer)
   {
-    repository
-        .get(transfer.getId())
-        .ifPresentOrElse(
-            stored ->
-            {
-              if (!transfer.equals(stored))
-                log.error("ignoring already received transfer with differing data: old={}, new={}", stored, transfer);
-            },
-            () ->
-            {
-              repository.store(transfer);
-              // TODO: Do some time consuming checks...
-              transfer.setState(CHECKED);
-              messagingService.send(transfer);
-            });
+    // TODO: Do some time consuming checks...
+    messagingService.send(transfer.getId(), CHECKED);
   }
 
   public Optional<Transfer> get(Long id)