X-Git-Url: https://juplo.de/gitweb/?p=demos%2Fkafka%2Fdemos-kafka-payment-system-transfer;a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fpayment%2Ftransfer%2Fadapter%2FTransferController.java;h=e60ba949da5bf7e3be4f6ae83ffffebfbfa8ba64;hp=e20f9bf9caf5bc94969ff26b7f9f6ef2a727758c;hb=19aec49a7f3a46f55e696a5a930c48883c4f1cd2;hpb=6191849fee717b080118717c86df79fad12bafc8

diff --git a/src/main/java/de/juplo/kafka/payment/transfer/adapter/TransferController.java b/src/main/java/de/juplo/kafka/payment/transfer/adapter/TransferController.java
index e20f9bf..e60ba94 100644
--- a/src/main/java/de/juplo/kafka/payment/transfer/adapter/TransferController.java
+++ b/src/main/java/de/juplo/kafka/payment/transfer/adapter/TransferController.java
@@ -3,7 +3,7 @@ package de.juplo.kafka.payment.transfer.adapter;
 
 import de.juplo.kafka.payment.transfer.domain.Transfer;
 import de.juplo.kafka.payment.transfer.ports.GetTransferUseCase;
-import de.juplo.kafka.payment.transfer.ports.InitiateTransferUseCase;
+import de.juplo.kafka.payment.transfer.ports.MessagingService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.HttpStatus;
@@ -12,6 +12,9 @@ import org.springframework.http.ResponseEntity;
 import org.springframework.validation.FieldError;
 import org.springframework.web.bind.MethodArgumentNotValidException;
 import org.springframework.web.bind.annotation.*;
+import org.springframework.web.context.request.async.DeferredResult;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.validation.Valid;
@@ -19,49 +22,109 @@ import java.net.URI;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 
 
-@RestController
+@RequestMapping(TransferController.PATH)
+@ResponseBody
 @RequiredArgsConstructor
 @Slf4j
  public class TransferController
 {
   public final static String PATH = "/transfers";
 
-  private final InitiateTransferUseCase initiateTransferUseCase;
   private final GetTransferUseCase getTransferUseCase;
+  private final MessagingService messagingService;
+  private final TransferConsumer consumer;
+  private final WebClient webClient;
 
 
   @PostMapping(
-      path = PATH,
+      path = "",
       consumes = MediaType.APPLICATION_JSON_VALUE,
       produces = MediaType.APPLICATION_JSON_VALUE)
-  public ResponseEntity<?> transfer(@Valid @RequestBody TransferDTO transferDTO)
+  public DeferredResult<ResponseEntity<?>> transfer(@Valid @RequestBody TransferDTO transferDTO)
   {
-    Transfer transfer =
-        Transfer
-            .builder()
-            .id(transferDTO.getId())
-            .payer(transferDTO.getPayer())
-            .payee(transferDTO.getPayee())
-            .amount(transferDTO.getAmount())
-            .build();
+    DeferredResult<ResponseEntity<?>> result = new DeferredResult<>();
 
-    initiateTransferUseCase.initiate(transfer);
+    Long id = transferDTO.getId();
 
-    return ResponseEntity.created(URI.create(PATH + "/" + transferDTO.getId())).build();
+    consumer
+        .uriForKey(id.toString())
+        .map(uri ->
+            webClient.get()
+                .uri(uri + PATH + "/" + id)
+                .accept(MediaType.APPLICATION_JSON)
+                .retrieve()
+                .onStatus(status -> true, bar -> Mono.empty())
+                .toBodilessEntity()
+                .blockOptional()
+                .flatMap(resp ->
+                    resp.getStatusCode().is2xxSuccessful()
+                        ? Optional.of(Boolean.TRUE)
+                        : Optional.<Boolean>empty()))
+        .or(() -> Optional.of(getTransferUseCase.get(transferDTO.getId()).map(transfer -> Boolean.TRUE)))
+        .flatMap(optional -> optional)
+        .map($ ->
+            CompletableFuture.completedFuture(
+                ResponseEntity
+                    .ok()
+                    .location(location(transferDTO))
+                    .build()))
+        .or(() ->
+            Optional.of(
+                messagingService
+                    .send(
+                        Transfer
+                            .builder()
+                            .id(transferDTO.getId())
+                            .payer(transferDTO.getPayer())
+                            .payee(transferDTO.getPayee())
+                            .amount(transferDTO.getAmount())
+                            .build())
+                    .thenApply($ ->
+                        ResponseEntity
+                            .created(location(transferDTO))
+                            .build())))
+        .get()
+        .thenAccept(responseEntity -> result.setResult(responseEntity))
+        .exceptionally(e ->
+        {
+          result.setErrorResult(e);
+          return null;
+        });
+
+    return result;
+  }
+
+  private URI location(TransferDTO transferDTO)
+  {
+    return URI.create(PATH + "/" + transferDTO.getId());
   }
 
   @GetMapping(
-      path = PATH + "/{id}",
+      path = "/{id}",
       produces = MediaType.APPLICATION_JSON_VALUE)
   public ResponseEntity<TransferDTO> get(@PathVariable Long id)
   {
     return
-        getTransferUseCase
-            .get(id)
-            .map(transfer -> ResponseEntity.ok(TransferDTO.of(transfer)))
-            .orElse(ResponseEntity.notFound().build());
+        consumer
+            .uriForKey(Long.toString(id))
+            .map(uri ->
+            {
+              ResponseEntity<TransferDTO> response =
+                  ResponseEntity
+                      .status(HttpStatus.TEMPORARY_REDIRECT)
+                      .location(URI.create(uri + PATH + "/" + id))
+                      .build();
+              return response;
+            })
+            .orElseGet(() ->
+                getTransferUseCase
+                    .get(id)
+                    .map(transfer -> ResponseEntity.ok(TransferDTO.of(transfer)))
+                    .orElse(ResponseEntity.notFound().build()));
   }
 
   @ResponseStatus(HttpStatus.BAD_REQUEST)