ROT: Verbesserungen aus 'deserialization' in 'sumup-adder' gemerged
[demos/kafka/training] / 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 AdderRecordHandler adderRecordHandler;
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, Long>> state()
37   {
38     return
39         adderRecordHandler
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 adderBusinessLogic : adderRecordHandler.getState().values())
52     {
53       Optional<Long> sum = adderBusinessLogic.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 }