-package de.juplo.kafka.chat.backend.persistence;
+package de.juplo.kafka.chat.backend.implementation;
import de.juplo.kafka.chat.backend.domain.ChatHomeService;
import de.juplo.kafka.chat.backend.domain.ChatRoomInfo;
import de.juplo.kafka.chat.backend.domain.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import java.util.UUID;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
public interface StorageStrategy
{
+ Logger log = LoggerFactory.getLogger(StorageStrategy.class.getCanonicalName());
+
default void write(ChatHomeService chatHomeService)
{
writeChatRoomInfo(
.flatMapMany(chatRoomData -> chatRoomData.getMessages()))));
}
+ default void writeChatRoomData(UUID chatRoomId, Flux<Message> messageFlux)
+ {
+ writeChatRoomData(
+ chatRoomId,
+ messageFlux,
+ (id) -> logSuccess(id),
+ (id, throwable) -> logFailure(id, throwable));
+ }
+
void writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux);
Flux<ChatRoomInfo> readChatRoomInfo();
- void writeChatRoomData(UUID chatRoomId, Flux<Message> messageFlux);
+ void writeChatRoomData(
+ UUID chatRoomId,
+ Flux<Message> messageFlux,
+ SuccessCallback callback,
+ FailureCallback failureCallback);
Flux<Message> readChatRoomData(UUID chatRoomId);
+
+ interface SuccessCallback extends Consumer<UUID> {}
+ interface FailureCallback extends BiConsumer<UUID, Throwable> {}
+
+ default void logSuccess(UUID chatRoomId)
+ {
+ log.info("Successfully stored chat-room {}", chatRoomId);
+ }
+
+ default void logFailure(UUID chatRoomId, Throwable throwable)
+ {
+ log.error("Could not store chat-room {}: {}", chatRoomId, throwable);
+ }
}