juplo.de Git - demos/kafka/chat/blob - src/main/java/de/juplo/kafka/chat/backend/api/ChatBackendController.java

   1 package de.juplo.kafka.chat.backend.api;
   2
   3 import de.juplo.kafka.chat.backend.domain.ChatHome;
   4 import de.juplo.kafka.chat.backend.domain.ChatRoom;
   5 import de.juplo.kafka.chat.backend.persistence.StorageStrategy;
   6 import lombok.RequiredArgsConstructor;
   7 import org.springframework.http.codec.ServerSentEvent;
   8 import org.springframework.web.bind.annotation.*;
   9 import reactor.core.publisher.Flux;
  10 import reactor.core.publisher.Mono;
  11
  12 import java.time.Clock;
  13 import java.time.LocalDateTime;
  14 import java.util.Optional;
  15 import java.util.UUID;
  16 import java.util.stream.Stream;
  17
  18
  19 @RestController
  20 @RequiredArgsConstructor
  21 public class ChatBackendController
  22 {
  23   private final ChatHome chatHome;
  24   private final Clock clock;
  25   private final StorageStrategy storageStrategy;
  26
  27
  28   @PostMapping("create")
  29   public ChatroomTo create(@RequestBody String name)
  30   {
  31     return ChatroomTo.from(chatHome.createChatroom(name));
  32   }
  33
  34   @GetMapping("list")
  35   public Stream<ChatroomTo> list()
  36   {
  37     return chatHome.list().map(chatroom -> ChatroomTo.from(chatroom));
  38   }
  39
  40   @GetMapping("list/{chatroomId}")
  41   public Flux<MessageTo> list(@PathVariable UUID chatroomId)
  42   {
  43     return chatHome
  44         .getChatroom(chatroomId)
  45         .map(chatroom -> chatroom
  46             .getMessages()
  47             .map(MessageTo::from))
  48         .get();
  49   }
  50
  51   @GetMapping("get/{chatroomId}")
  52   public Optional<ChatroomTo> get(@PathVariable UUID chatroomId)
  53   {
  54     return chatHome.getChatroom(chatroomId).map(chatroom -> ChatroomTo.from(chatroom));
  55   }
  56
  57   @PutMapping("put/{chatroomId}/{username}/{messageId}")
  58   public Mono<MessageTo> put(
  59       @PathVariable UUID chatroomId,
  60       @PathVariable String username,
  61       @PathVariable Long messageId,
  62       @RequestBody String text)
  63   {
  64     return
  65         chatHome
  66             .getChatroom(chatroomId)
  67             .map(chatroom -> put(chatroom, username, messageId, text))
  68             .orElseThrow(() -> new UnknownChatroomException(chatroomId));
  69   }
  70
  71   public Mono<MessageTo> put(
  72       ChatRoom chatroom,
  73       String username,
  74       Long messageId,
  75       String text)
  76   {
  77     return
  78         chatroom
  79             .addMessage(
  80                 messageId,
  81                 LocalDateTime.now(clock),
  82                 username,
  83                 text)
  84             .switchIfEmpty(chatroom.getMessage(username, messageId))
  85             .map(message -> MessageTo.from(message));
  86   }
  87
  88   @GetMapping("get/{chatroomId}/{username}/{messageId}")
  89   public Mono<MessageTo> get(
  90       @PathVariable UUID chatroomId,
  91       @PathVariable String username,
  92       @PathVariable Long messageId)
  93   {
  94     return
  95         chatHome
  96             .getChatroom(chatroomId)
  97             .map(chatroom -> get(chatroom, username, messageId))
  98             .orElseThrow(() -> new UnknownChatroomException(chatroomId));
  99   }
 100
 101   private Mono<MessageTo> get(
 102       ChatRoom chatroom,
 103       String username,
 104       Long messageId)
 105   {
 106     return
 107         chatroom
 108             .getMessage(username, messageId)
 109             .map(message -> MessageTo.from(message));
 110   }
 111
 112   @GetMapping(path = "listen/{chatroomId}")
 113   public Flux<ServerSentEvent<MessageTo>> listen(@PathVariable UUID chatroomId)
 114   {
 115     return chatHome
 116         .getChatroom(chatroomId)
 117         .map(chatroom -> listen(chatroom))
 118         .orElseThrow(() -> new UnknownChatroomException(chatroomId));
 119   }
 120
 121   private Flux<ServerSentEvent<MessageTo>> listen(ChatRoom chatroom)
 122   {
 123     return chatroom
 124         .listen()
 125         .log()
 126         .map(message -> MessageTo.from(message))
 127         .map(messageTo ->
 128             ServerSentEvent
 129                 .builder(messageTo)
 130                 .id(messageTo.getSerial().toString())
 131                 .event("message")
 132                 .build());
 133   }
 134
 135   @PostMapping("/store")
 136   public void store()
 137   {
 138     storageStrategy.writeChatrooms(Flux.fromStream(chatHome.list()));
 139   }
 140 }