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.

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 8f3e49565b0a95cdc240ccdeebdeaadafb5a03e4..0345c0083c924358da556f95846252239397b7ec 100644 (file)
--- 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 93593d86f11ec84cb410ac639fb5c1615bf2a436..dfe8567a8bcf3e123c1c7a719c8d620cc81edec7 100644 (file)
--- 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;
   }