From 803a9193c8b1ad1ec68f1818fc091291a60c8f4c Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Tue, 20 Feb 2024 16:12:00 +0100 Subject: [PATCH] refactor: RED - Refined success/error-handling for restore-operations * This innocent little change discloses a severe missconception in the implementation of the storage strategies. * The call to `Mono.block()`, though not really changing the behaviour during the restore-process, triggers a sanity-check from io.projectractor. --- .../inmemory/InMemoryChatMessageService.java | 7 ++++++- .../inmemory/SimpleChatHomeService.java | 15 ++++++++++----- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/inmemory/InMemoryChatMessageService.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/inmemory/InMemoryChatMessageService.java index 8f3e4956..0345c008 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/implementation/inmemory/InMemoryChatMessageService.java +++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/inmemory/InMemoryChatMessageService.java @@ -20,7 +20,12 @@ public class InMemoryChatMessageService implements ChatMessageService { log.debug("Creating InMemoryChatMessageService"); messages = new LinkedHashMap<>(); - messageFlux.subscribe(message -> messages.put(message.getKey(), message)); + messageFlux + .doOnNext(message -> messages.put(message.getKey(), message)) + .then() + .doOnSuccess(empty -> log.info("Restored InMemoryChatMessageService")) + .doOnError(throwable -> log.error("Could not restore InMemoryChatMessageService")) + .block(); } @Override diff --git a/src/main/java/de/juplo/kafka/chat/backend/implementation/inmemory/SimpleChatHomeService.java b/src/main/java/de/juplo/kafka/chat/backend/implementation/inmemory/SimpleChatHomeService.java index 93593d86..dfe8567a 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/implementation/inmemory/SimpleChatHomeService.java +++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/inmemory/SimpleChatHomeService.java @@ -40,11 +40,12 @@ public class SimpleChatHomeService implements ChatHomeService Clock clock, int bufferSize) { - this.shard = shard; - log.info("Created {}", this); + log.debug("Creating SimpleChatHomeService"); + this.shard = shard; this.chatRoomInfo = new HashMap<>(); this.chatRoomData = new HashMap<>(); + storageStrategy .readChatRoomInfo() .filter(info -> @@ -62,8 +63,7 @@ public class SimpleChatHomeService implements ChatHomeService return false; } }) - .toStream() - .forEach(info -> + .doOnNext(info -> { UUID chatRoomId = info.getId(); chatRoomInfo.put(chatRoomId, info); @@ -75,7 +75,12 @@ public class SimpleChatHomeService implements ChatHomeService clock, new InMemoryChatMessageService(messageFlux), bufferSize)); - }); + }) + .then() + .doOnSuccess(empty -> log.info("Restored {}", this)) + .doOnError(throwable -> log.error("Could not restore {}", this)) + .block(); + this.clock = clock; this.bufferSize = bufferSize; } -- 2.20.1