juplo.de Git - demos/kafka/training/blob - src/main/java/de/juplo/kafka/DriverController.java

   1 package de.juplo.kafka;
   2
   3 import lombok.RequiredArgsConstructor;
   4 import org.springframework.http.HttpStatus;
   5 import org.springframework.http.ResponseEntity;
   6 import org.springframework.web.bind.annotation.*;
   7
   8 import java.util.Map;
   9 import java.util.Optional;
  10 import java.util.concurrent.ExecutionException;
  11 import java.util.stream.Collectors;
  12
  13
  14 @RestController
  15 @RequiredArgsConstructor
  16 public class DriverController
  17 {
  18   private final EndlessConsumer consumer;
  19   private final ApplicationRecordHandler recordHandler;
  20
  21
  22   @PostMapping("start")
  23   public void start()
  24   {
  25     consumer.start();
  26   }
  27
  28   @PostMapping("stop")
  29   public void stop() throws ExecutionException, InterruptedException
  30   {
  31     consumer.stop();
  32   }
  33
  34
  35   @GetMapping("state")
  36   public Map<Integer, Map<String, AdderResult>> state()
  37   {
  38     return
  39         recordHandler
  40             .getState()
  41             .entrySet()
  42             .stream()
  43             .collect(Collectors.toMap(
  44                 entry -> entry.getKey(),
  45                 entry -> entry.getValue().getState()));
  46   }
  47
  48   @GetMapping("state/{user}")
  49   public ResponseEntity<Long> seen(@PathVariable String user)
  50   {
  51     for (AdderBusinessLogic adder : recordHandler.getState().values())
  52     {
  53       Optional<Long> sum = adder.getSum(user);
  54       if (sum.isPresent())
  55         return ResponseEntity.ok(sum.get());
  56     }
  57
  58     return ResponseEntity.notFound().build();
  59   }
  60
  61
  62   @ExceptionHandler
  63   @ResponseStatus(HttpStatus.BAD_REQUEST)
  64   public ErrorResponse illegalStateException(IllegalStateException e)
  65   {
  66     return new ErrorResponse(e.getMessage(), HttpStatus.BAD_REQUEST.value());
  67   }
  68 }