WIP:mongodb map vs subscribe - subscribe rausgezogen

diff --git a/src/main/java/de/juplo/kafka/chat/backend/ChatBackendApplication.java b/src/main/java/de/juplo/kafka/chat/backend/ChatBackendApplication.java
index 1eaa88c..76debbe 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/ChatBackendApplication.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/ChatBackendApplication.java
@@ -32,7 +32,9 @@ public class ChatBackendApplication implements WebFluxConfigurer
        @PreDestroy
        public void onExit()
        {
-               storageStrategy.write(chatHomeService);
+               storageStrategy
+                               .write(chatHomeService)
+                               .subscribe();
        }
 
        public static void main(String[] args)

diff --git a/src/main/java/de/juplo/kafka/chat/backend/api/ChatBackendController.java b/src/main/java/de/juplo/kafka/chat/backend/api/ChatBackendController.java
index f3efe79..acb84e6 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/api/ChatBackendController.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/api/ChatBackendController.java
@@ -137,6 +137,6 @@ public class ChatBackendController
   @PostMapping("/store")
   public void store()
   {
-    storageStrategy.write(chatHomeService);
+    storageStrategy.write(chatHomeService).subscribe();
   }
 }

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 a62f408..4b43f88 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/implementation/StorageStrategy.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/implementation/StorageStrategy.java
@@ -16,9 +16,9 @@ public interface StorageStrategy
 {
   Logger log = LoggerFactory.getLogger(StorageStrategy.class.getCanonicalName());
 
-  default void write(ChatHomeService chatHomeService)
+  default Flux<ChatRoomInfo> write(ChatHomeService chatHomeService)
   {
-    writeChatRoomInfo(
+    return writeChatRoomInfo(
         chatHomeService
             .getChatRoomInfo()
             .doOnNext(chatRoomInfo ->
@@ -28,7 +28,7 @@ public interface StorageStrategy
                         .getChatRoomData(chatRoomInfo.getId())
                         .flatMapMany(chatRoomData -> chatRoomData.getMessages()),
                     this::logSuccess,
-                    this::logFailure)));
+                    this::logFailure).subscribe()));
   }
 
   Flux<ChatRoomInfo> writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux);

diff --git a/src/test/java/de/juplo/kafka/chat/backend/AbstractStorageStrategyIT.java b/src/test/java/de/juplo/kafka/chat/backend/AbstractStorageStrategyIT.java
index 5eaf541..41e80ed 100644 (file)
--- a/src/test/java/de/juplo/kafka/chat/backend/AbstractStorageStrategyIT.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/AbstractStorageStrategyIT.java
@@ -28,7 +28,9 @@ public abstract class AbstractStorageStrategyIT
 
   protected void stop()
   {
-    getStorageStrategy().write(chathome);
+    getStorageStrategy()
+        .write(chathome)
+        .subscribe();
   }
 
   @Test