1 package de.juplo.kafka.chat.backend.implementation;
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;
11 import java.util.UUID;
14 public interface StorageStrategy
16 Logger log = LoggerFactory.getLogger(StorageStrategy.class.getCanonicalName());
18 default Mono<Void> write(ChatHomeService chatHomeService)
20 return writeChatRoomInfo(chatHomeService.getChatRoomInfo())
21 .flatMap(chatRoomInfo -> writeChatRoomData(
24 .getChatRoomData(chatRoomInfo.getId())
25 .flatMapMany(chatRoomData -> chatRoomData.getMessages())
28 .doOnSuccess(count -> log.info("Stored {} messages for {}", count, chatRoomInfo))
29 .doOnError(throwable -> log.error("Could not store {}: {}", chatRoomInfo, throwable)))
31 .doOnSuccess(count -> log.info("Stored {} chat-rooms for {}", count, chatHomeService))
32 .doOnError(throwable -> log.error("Could not store {}: {}", chatHomeService, throwable))
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);