juplo.de Git - demos/kafka/chat/blob - src/main/java/de/juplo/kafka/chat/backend/implementation/StorageStrategy.java

   1 package de.juplo.kafka.chat.backend.implementation;
   2
   3 import de.juplo.kafka.chat.backend.domain.ChatHomeService;
   4 import de.juplo.kafka.chat.backend.domain.ChatRoomInfo;
   5 import de.juplo.kafka.chat.backend.domain.Message;
   6 import org.slf4j.Logger;
   7 import org.slf4j.LoggerFactory;
   8 import reactor.core.publisher.Flux;
   9
  10 import java.util.UUID;
  11
  12
  13 public interface StorageStrategy
  14 {
  15   Logger log = LoggerFactory.getLogger(StorageStrategy.class.getCanonicalName());
  16
  17   default Flux<ChatRoomInfo> write(ChatHomeService chatHomeService)
  18   {
  19     return writeChatRoomInfo(
  20         chatHomeService
  21             .getChatRoomInfo()
  22             .doOnComplete(() -> log.info("Stored {}", chatHomeService))
  23             .doOnError(throwable -> log.error("Could not store {}: {}", chatHomeService, throwable))
  24             .doOnNext(chatRoomInfo ->
  25                 writeChatRoomData(
  26                     chatRoomInfo.getId(),
  27                     chatHomeService
  28                         .getChatRoomData(chatRoomInfo.getId())
  29                         .flatMapMany(chatRoomData -> chatRoomData.getMessages())
  30                         .doOnComplete(() -> log.info("Stored {}", chatRoomInfo))
  31                         .doOnError(throwable -> log.error("Could not store {}: {}", chatRoomInfo, throwable)))
  32                     .subscribe()));
  33   }
  34
  35   Flux<ChatRoomInfo> writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux);
  36   Flux<ChatRoomInfo> readChatRoomInfo();
  37   Flux<Message> writeChatRoomData(UUID chatRoomId, Flux<Message> messageFlux);
  38   Flux<Message> readChatRoomData(UUID chatRoomId);
  39 }