WIP

diff --git a/pom.xml b/pom.xml
index 7521e14..4d3d183 100644 (file)
--- a/pom.xml
+++ b/pom.xml
@@ -5,7 +5,7 @@
        <parent>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-parent</artifactId>
-               <version>3.0.0</version>
+               <version>3.0.1</version>
                <relativePath/> <!-- lookup parent from repository -->
        </parent>
        <groupId>de.juplo.kafka</groupId>
@@ -40,8 +40,8 @@
                        <optional>true</optional>
                </dependency>
                <dependency>
-                       <groupId>org.springframework.data</groupId>
-                       <artifactId>spring-data-mongodb</artifactId>
+                       <groupId>org.springframework.boot</groupId>
+                       <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
                </dependency>
                <dependency>
                        <groupId>org.springframework.boot</groupId>
@@ -59,6 +59,12 @@
                        <artifactId>mockito-core</artifactId>
                        <scope>test</scope>
                </dependency>
+               <dependency>
+                       <groupId>de.flapdoodle.embed</groupId>
+                       <artifactId>de.flapdoodle.embed.mongo</artifactId>
+                       <version>4.3.2</version>
+                       <scope>test</scope>
+               </dependency>
        </dependencies>
 
        <build>

diff --git a/src/main/java/de/juplo/kafka/chat/backend/ChatBackendConfiguration.java b/src/main/java/de/juplo/kafka/chat/backend/ChatBackendConfiguration.java
index 87965ab..0cca42d 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/ChatBackendConfiguration.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/ChatBackendConfiguration.java
@@ -10,6 +10,7 @@ import de.juplo.kafka.chat.backend.persistence.inmemory.InMemoryChatRoomService;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories;
 
 import java.nio.file.Paths;
 import java.time.Clock;
@@ -17,6 +18,7 @@ import java.time.Clock;
 
 @Configuration
 @EnableConfigurationProperties(ChatBackendProperties.class)
+@EnableReactiveMongoRepositories
 public class ChatBackendConfiguration
 {
   @Bean

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/ChatHomeRepository.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/ChatHomeRepository.java
index cc038c1..afb30fc 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/ChatHomeRepository.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/ChatHomeRepository.java
@@ -1,11 +1,8 @@
 package de.juplo.kafka.chat.backend.persistence.storage.mongodb;
 
-import org.springframework.data.mongodb.repository.MongoRepository;
+import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
 
-import java.util.List;
 
-
-public interface ChatHomeRepository extends MongoRepository<ChatRoomTo, String>
+public interface ChatHomeRepository extends ReactiveMongoRepository<ChatRoomTo, String>
 {
-  public List<ChatRoomTo> findAll();
 }

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/ChatRoomTo.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/ChatRoomTo.java
index 701998e..f4f3ac6 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/ChatRoomTo.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/ChatRoomTo.java
@@ -6,7 +6,6 @@ import org.springframework.data.annotation.Id;
 import org.springframework.data.mongodb.core.mapping.DBRef;
 
 import java.util.List;
-import java.util.UUID;
 
 
 @AllArgsConstructor
@@ -23,7 +22,6 @@ public class ChatRoomTo
   @DBRef
   private List<MessageTo> messages;
 
-
   public static ChatRoomTo from(ChatRoom chatroom)
   {
     return new ChatRoomTo(

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/MessageTo.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/MessageTo.java
index c61a1cc..8e37166 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/MessageTo.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/MessageTo.java
@@ -4,7 +4,6 @@ import de.juplo.kafka.chat.backend.domain.Message;
 import lombok.*;
 
 import java.time.LocalDateTime;
-import java.util.UUID;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/MongoDbStorageStrategy.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/MongoDbStorageStrategy.java
index ac249fc..73f5e8a 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/MongoDbStorageStrategy.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/storage/mongodb/MongoDbStorageStrategy.java
@@ -2,32 +2,45 @@ package de.juplo.kafka.chat.backend.persistence.storage.mongodb;
 
 import de.juplo.kafka.chat.backend.domain.ChatRoom;
 import de.juplo.kafka.chat.backend.persistence.StorageStrategy;
+import de.juplo.kafka.chat.backend.persistence.storage.files.ChatRoomServiceFactory;
+import lombok.RequiredArgsConstructor;
 import reactor.core.publisher.Flux;
 
-import java.io.IOException;
+import java.time.Clock;
+import java.util.UUID;
 
 
+@RequiredArgsConstructor
 public class MongoDbStorageStrategy implements StorageStrategy
 {
+  private final ChatHomeRepository repository;
+  private final Clock clock;
+  private final int bufferSize;
+  private final ChatRoomServiceFactory factory;
+
+
   @Override
   public void write(Flux<ChatRoom> chatroomFlux)
   {
     chatroomFlux
         .log()
         .map(ChatRoomTo::from)
-        .subscribe(chatroom ->
-        {
-        });
-  }
-    catch (IOException e)
-  {
-    throw new RuntimeException(e);
-  }
+        .subscribe(chatroom -> repository.save(chatroom));
   }
 
   @Override
   public Flux<ChatRoom> read()
   {
-    return null;
+    return repository
+        .findAll()
+        .map(chatRoomTo -> new ChatRoom(
+            UUID.fromString(chatRoomTo.getId()),
+            chatRoomTo.getName(),
+            clock,
+            factory.create(
+                Flux
+                    .fromIterable(chatRoomTo.getMessages())
+                    .map(messageTo -> messageTo.toMessage())),
+            bufferSize));
   }
 }