WIP:mongodb map vs subscribe
authorKai Moritz <kai@juplo.de>
Sat, 3 Feb 2024 23:37:23 +0000 (00:37 +0100)
committerKai Moritz <kai@juplo.de>
Mon, 19 Feb 2024 13:56:54 +0000 (14:56 +0100)
src/main/java/de/juplo/kafka/chat/backend/storage/files/FilesStorageStrategy.java
src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/MongoDbStorageStrategy.java

index cb7dd31..7e04a96 100644 (file)
@@ -74,18 +74,20 @@ public class FilesStorageStrategy implements StorageStrategy
               throw new RuntimeException(e);
             }
           })
-          .subscribe(chatRoomInfo ->
+          .map(chatRoomInfo ->
           {
             try
             {
               ChatRoomInfoTo chatRoomInfoTo = ChatRoomInfoTo.from(chatRoomInfo);
               generator.writeObject(chatRoomInfoTo);
+              return chatRoomInfo;
             }
             catch (IOException e)
             {
               throw new RuntimeException(e);
             }
-          });
+          })
+          .subscribe();
     }
     catch (IOException e)
     {
@@ -160,18 +162,20 @@ public class FilesStorageStrategy implements StorageStrategy
               throw new RuntimeException(e);
             }
           })
-          .subscribe(message ->
+          .map(message ->
           {
             try
             {
               MessageTo messageTo = MessageTo.from(message);
               generator.writeObject(messageTo);
+              return message;
             }
             catch (IOException e)
             {
               throw new RuntimeException(e);
             }
-          });
+          })
+          .subscribe();
     }
     catch (IOException e)
     {
index b1bead9..780d64b 100644 (file)
@@ -25,7 +25,8 @@ public class MongoDbStorageStrategy implements StorageStrategy
   {
     chatRoomInfoFlux
         .map(ChatRoomTo::from)
-        .subscribe(chatroomTo -> chatRoomRepository.save(chatroomTo));
+        .map(chatroomTo -> chatRoomRepository.save(chatroomTo))
+        .subscribe();
   }
 
   @Override
@@ -45,7 +46,8 @@ public class MongoDbStorageStrategy implements StorageStrategy
   {
     messageFlux
         .map(message -> MessageTo.from(chatRoomId, message))
-        .subscribe(messageTo -> messageRepository.save(messageTo));
+        .map(messageTo -> messageRepository.save(messageTo))
+        .subscribe();
   }
 
   @Override