From: Kai Moritz <kai@juplo.de>
Date: Sun, 18 Feb 2024 18:44:05 +0000 (+0100)
Subject: WIP:mongodb map vs subscribe - subscribe rausgezogen
X-Git-Tag: rebase--2024-02-19--08-59~3
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=16a5f36a912c0ed308c102bac52e87077f52ea6f;p=demos%2Fkafka%2Fchat

WIP:mongodb map vs subscribe - subscribe rausgezogen
---

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 9dd76250..a62f4082 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
@@ -31,21 +31,21 @@ public interface StorageStrategy
                     this::logFailure)));
   }
 
-  void writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux);
+  Flux<ChatRoomInfo> writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux);
   Flux<ChatRoomInfo> readChatRoomInfo();
-  default void writeChatRoomData(
+  default Flux<Message> writeChatRoomData(
       UUID chatRoomId,
       Flux<Message> messageFlux,
       SuccessCallback successCallback,
       FailureCallback failureCallback)
   {
-    writeChatRoomData(
+    return writeChatRoomData(
         chatRoomId,
         messageFlux
             .doOnComplete(() -> successCallback.accept(chatRoomId))
             .doOnError(throwable -> failureCallback.accept(chatRoomId, throwable)));
   }
-  void writeChatRoomData(UUID chatRoomId, Flux<Message> messageFlux);
+  Flux<Message> writeChatRoomData(UUID chatRoomId, Flux<Message> messageFlux);
   Flux<Message> readChatRoomData(UUID chatRoomId);
 
   interface SuccessCallback extends Consumer<UUID> {}
diff --git a/src/main/java/de/juplo/kafka/chat/backend/storage/files/FilesStorageStrategy.java b/src/main/java/de/juplo/kafka/chat/backend/storage/files/FilesStorageStrategy.java
index 7e04a964..cdb4f0d7 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/storage/files/FilesStorageStrategy.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/storage/files/FilesStorageStrategy.java
@@ -35,7 +35,7 @@ public class FilesStorageStrategy implements StorageStrategy
 
 
   @Override
-  public void writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux)
+  public Flux<ChatRoomInfo> writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux)
   {
     Path path = chatroomsPath();
     log.info("Writing chatrooms to {}", path);
@@ -48,7 +48,7 @@ public class FilesStorageStrategy implements StorageStrategy
               .getFactory()
               .createGenerator(Files.newBufferedWriter(path, CREATE, TRUNCATE_EXISTING));
 
-      chatRoomInfoFlux
+      return chatRoomInfoFlux
           .log()
           .doFirst(() ->
           {
@@ -86,8 +86,7 @@ public class FilesStorageStrategy implements StorageStrategy
             {
               throw new RuntimeException(e);
             }
-          })
-          .subscribe();
+          });
     }
     catch (IOException e)
     {
@@ -121,7 +120,7 @@ public class FilesStorageStrategy implements StorageStrategy
   }
 
   @Override
-  public void writeChatRoomData(
+  public Flux<Message> writeChatRoomData(
       UUID chatRoomId,
       Flux<Message> messageFlux)
   {
@@ -136,7 +135,7 @@ public class FilesStorageStrategy implements StorageStrategy
               .getFactory()
               .createGenerator(Files.newBufferedWriter(path, CREATE, TRUNCATE_EXISTING));
 
-      messageFlux
+      return messageFlux
           .log()
           .doFirst(() ->
           {
@@ -174,8 +173,7 @@ public class FilesStorageStrategy implements StorageStrategy
             {
               throw new RuntimeException(e);
             }
-          })
-          .subscribe();
+          });
     }
     catch (IOException e)
     {
diff --git a/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/ChatRoomTo.java b/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/ChatRoomTo.java
index 853ee1cf..47596a2e 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/ChatRoomTo.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/ChatRoomTo.java
@@ -5,6 +5,8 @@ import lombok.*;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.mongodb.core.mapping.Document;
 
+import java.util.UUID;
+
 
 @AllArgsConstructor
 @NoArgsConstructor
@@ -19,6 +21,13 @@ public class ChatRoomTo
   private String id;
   private String name;
 
+  public ChatRoomInfo toChatRoomInfo()
+  {
+    return new ChatRoomInfo(
+        UUID.fromString(id),
+        name);
+  }
+
   public static ChatRoomTo from(ChatRoomInfo chatRoomInfo)
   {
     return new ChatRoomTo(
diff --git a/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/MongoDbStorageStrategy.java b/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/MongoDbStorageStrategy.java
index 780d64be..1428119e 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/MongoDbStorageStrategy.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/storage/mongodb/MongoDbStorageStrategy.java
@@ -21,12 +21,12 @@ public class MongoDbStorageStrategy implements StorageStrategy
 
 
   @Override
-  public void writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux)
+  public Flux<ChatRoomInfo> writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux)
   {
-    chatRoomInfoFlux
+    return chatRoomInfoFlux
         .map(ChatRoomTo::from)
         .map(chatroomTo -> chatRoomRepository.save(chatroomTo))
-        .subscribe();
+        .map(ChatRoomTo::toChatRoomInfo);
   }
 
   @Override
@@ -42,12 +42,12 @@ public class MongoDbStorageStrategy implements StorageStrategy
   }
 
   @Override
-  public void writeChatRoomData(UUID chatRoomId, Flux<Message> messageFlux)
+  public Flux<Message> writeChatRoomData(UUID chatRoomId, Flux<Message> messageFlux)
   {
-    messageFlux
+    return messageFlux
         .map(message -> MessageTo.from(chatRoomId, message))
         .map(messageTo -> messageRepository.save(messageTo))
-        .subscribe();
+        .map(MessageTo::toMessage);
   }
 
   @Override
diff --git a/src/main/java/de/juplo/kafka/chat/backend/storage/nostorage/NoStorageStorageConfiguration.java b/src/main/java/de/juplo/kafka/chat/backend/storage/nostorage/NoStorageStorageConfiguration.java
index 1b20aa37..62150695 100644
--- a/src/main/java/de/juplo/kafka/chat/backend/storage/nostorage/NoStorageStorageConfiguration.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/storage/nostorage/NoStorageStorageConfiguration.java
@@ -32,7 +32,10 @@ public class NoStorageStorageConfiguration
     return new StorageStrategy()
     {
       @Override
-      public void writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux) {}
+      public Flux<ChatRoomInfo> writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux)
+      {
+        return chatRoomInfoFlux;
+      }
 
       @Override
       public Flux<ChatRoomInfo> readChatRoomInfo()
@@ -41,7 +44,10 @@ public class NoStorageStorageConfiguration
       }
 
       @Override
-      public void writeChatRoomData(UUID chatRoomId, Flux<Message> messageFlux) {}
+      public Flux<Message> writeChatRoomData(UUID chatRoomId, Flux<Message> messageFlux)
+      {
+        return messageFlux;
+      }
 
       @Override
       public Flux<Message> readChatRoomData(UUID chatRoomId)