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 import reactor.core.publisher.Mono;
  10
  11 import java.util.UUID;
  12
  13
  14 public interface StorageStrategy
  15 {
  16   Logger log = LoggerFactory.getLogger(StorageStrategy.class.getCanonicalName());
  17
  18   default Mono<Void> write(ChatHomeService chatHomeService)
  19   {
  20     return writeChatRoomInfo(chatHomeService.getChatRoomInfo())
  21         .flatMap(chatRoomInfo -> writeChatRoomData(
  22             chatRoomInfo.getId(),
  23             chatHomeService
  24                 .getChatRoomData(chatRoomInfo.getId())
  25                 .flatMapMany(chatRoomData -> chatRoomData.getMessages())
  26             )
  27             .count()
  28             .doOnSuccess(count -> log.info("Stored {} messages for {}", count, chatRoomInfo))
  29             .doOnError(throwable -> log.error("Could not store {}: {}", chatRoomInfo, throwable)))
  30         .count()
  31         .doOnSuccess(count -> log.info("Stored {} chat-rooms for {}", count, chatHomeService))
  32         .doOnError(throwable -> log.error("Could not store {}: {}", chatHomeService, throwable))
  33         .then();
  34   }
  35
  36   Flux<ChatRoomInfo> writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux);
  37   Flux<ChatRoomInfo> readChatRoomInfo();
  38   Flux<Message> writeChatRoomData(UUID chatRoomId, Flux<Message> messageFlux);
  39   Flux<Message> readChatRoomData(UUID chatRoomId);
  40 }