juplo.de Git - demos/kafka/chat/blob - src/main/java/de/juplo/kafka/chat/backend/persistence/inmemory/SimpleChatHome.java

   1 package de.juplo.kafka.chat.backend.persistence.inmemory;
   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.domain.UnknownChatroomException;
   6 import lombok.extern.slf4j.Slf4j;
   7 import reactor.core.publisher.Flux;
   8 import reactor.core.publisher.Mono;
   9
  10 import java.util.*;
  11
  12
  13 @Slf4j
  14 public class SimpleChatHome implements ChatHome
  15 {
  16   private final int shard;
  17   private final Map<UUID, ChatRoom> chatrooms;
  18
  19
  20
  21   public SimpleChatHome(
  22       int shard,
  23       Flux<ChatRoom> chatroomFlux)
  24   {
  25     log.info("Created SimpleChatHome for shard {}", shard);
  26
  27     this.chatrooms = new HashMap<>();
  28     chatroomFlux
  29         .filter(chatRoom ->
  30         {
  31           if (shard > -1 && chatRoom.getShard() == shard)
  32           {
  33             return true;
  34           }
  35           else
  36           {
  37             log.info(
  38                 "SimpleChatHome for shard {} ignores not owned chat-room {}",
  39                 shard,
  40                 chatRoom);
  41             return false;
  42           }
  43         })
  44         .toStream()
  45         .forEach(chatroom -> chatrooms.put(chatroom.getId(), chatroom));
  46     this.shard = shard;
  47   }
  48
  49
  50   @Override
  51   public Mono<ChatRoom> getChatRoom(UUID id)
  52   {
  53     return Mono
  54         .justOrEmpty(chatrooms.get(id))
  55         .switchIfEmpty(Mono.error(() -> new UnknownChatroomException(id)));
  56   }
  57
  58   @Override
  59   public Flux<ChatRoom> getChatRooms()
  60   {
  61     return Flux.fromIterable(chatrooms.values());
  62   }
  63 }