From: Kai Moritz Date: Sat, 3 Feb 2024 23:12:06 +0000 (+0100) Subject: WIP:mongodb X-Git-Tag: rebase--2024-02-18--17-42~5 X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=9b5bff532bfff25bac6a714a41f2d4447f5007e0;p=demos%2Fkafka%2Fchat WIP:mongodb --- diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/StorageStrategy.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/StorageStrategy.java index ba8bc23a..a7fd9d2e 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/implementation/StorageStrategy.java +++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/StorageStrategy.java @@ -8,6 +8,8 @@ import org.slf4j.LoggerFactory; import reactor.core.publisher.Flux; import java.util.UUID; +import java.util.concurrent.Phaser; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -18,15 +20,37 @@ public interface StorageStrategy default void write(ChatHomeService chatHomeService) { + Phaser writtenChatRooms = new Phaser(1); + AtomicInteger numErrors = new AtomicInteger(); + writeChatRoomInfo( chatHomeService .getChatRoomInfo() - .doOnNext(chatRoomInfo -> - writeChatRoomData( - chatRoomInfo.getId(), - chatHomeService - .getChatRoomData(chatRoomInfo.getId()) - .flatMapMany(chatRoomData -> chatRoomData.getMessages())))); + .doOnNext(chatRoomInfo -> writtenChatRooms.register()) + .doOnNext(chatRoomInfo -> writeChatRoomData( + chatRoomInfo.getId(), + chatHomeService + .getChatRoomData(chatRoomInfo.getId()) + .flatMapMany(chatRoomData -> chatRoomData.getMessages()), + (chatRoomId) -> + { + logSuccess(chatRoomId); + writtenChatRooms.arriveAndDeregister(); + }, + (chatRoomId, throwable) -> + { + logFailure(chatRoomId, throwable); + numErrors.incrementAndGet(); + writtenChatRooms.arriveAndDeregister(); + }))); + + writtenChatRooms.arriveAndAwaitAdvance(); + if (numErrors.get() > 0) + { + throw new RuntimeException("Could not write all chat-rooms for " + chatHomeService); + } + + log.info("All chat-rooms were written successfully for {}", chatHomeService); } default void writeChatRoomData(UUID chatRoomId, Flux messageFlux) @@ -34,8 +58,8 @@ public interface StorageStrategy writeChatRoomData( chatRoomId, messageFlux, - (id) -> log.info("Successfully stored chat-room {}", id), - (id, throwable) -> log.error("Could not store chat-room {}: {}", id, throwable)); + (id) -> logSuccess(id), + (id, throwable) -> logFailure(id, throwable)); } void writeChatRoomInfo(Flux chatRoomInfoFlux); @@ -49,4 +73,14 @@ public interface StorageStrategy interface SuccessCallback extends Consumer {} interface FailureCallback extends BiConsumer {} + + 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); + } } diff --git a/src/test/java/de/juplo/kafka/chat/backend/AbstractConfigurationIT.java b/src/test/java/de/juplo/kafka/chat/backend/AbstractConfigurationIT.java index 41618e56..652c9b93 100644 --- a/src/test/java/de/juplo/kafka/chat/backend/AbstractConfigurationIT.java +++ b/src/test/java/de/juplo/kafka/chat/backend/AbstractConfigurationIT.java @@ -2,7 +2,10 @@ package de.juplo.kafka.chat.backend; import com.fasterxml.jackson.databind.ObjectMapper; import de.juplo.kafka.chat.backend.api.ChatRoomInfoTo; +import de.juplo.kafka.chat.backend.domain.ChatHomeService; +import de.juplo.kafka.chat.backend.implementation.StorageStrategy; import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; @@ -53,8 +56,18 @@ public abstract class AbstractConfigurationIT .expectStatus().isOk() .expectBody().jsonPath("$.status").isEqualTo("UP"); }); + } + @AfterAll + static void waitForCompletionOfStorage( + @Autowired StorageStrategy storageStrategy, + @Autowired ChatHomeService chatHomeService) + { + storageStrategy.write(chatHomeService); + } + + @Test @DisplayName("Restored chat-rooms can be listed") void testRestoredChatRoomsCanBeListed()