refactor: Streamlined the API of the services

- `ChatRoomService` and `ChatHomeService` both return only reactive types.
- The decission stems from the wish, to become reactive all the way from
  the client to the technical implementation of the backend-services.
- This faciliates the mapping of the results in `ChatBackendController`.
- The refactoring already simplified lots of code, where a `Flux` has
  to derived from the `Stream`, yielding a good feeling about the plan,
  that is pursued with this refactoring.
- The refactoring also lead to the decision that `UnknownChatroomException`
  realy is a business-logic-exception (that according refactoring was
  already performed in the previous commits, to ceap this commit clean).

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 dbd12b0..f98a02a 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/ChatBackendApplication.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/ChatBackendApplication.java
@@ -8,7 +8,6 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.web.reactive.config.CorsRegistry;
 import org.springframework.web.reactive.config.WebFluxConfigurer;
-import reactor.core.publisher.Flux;
 
 
 @SpringBootApplication
@@ -33,7 +32,7 @@ public class ChatBackendApplication implements WebFluxConfigurer
        @PreDestroy
        public void onExit()
        {
-               storageStrategy.writeChatrooms(Flux.fromStream(chatHome.list()));
+               storageStrategy.writeChatrooms(chatHome.list());
        }
 
        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 31a2035..68d056b 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
@@ -2,7 +2,6 @@ package de.juplo.kafka.chat.backend.api;
 
 import de.juplo.kafka.chat.backend.domain.ChatHome;
 import de.juplo.kafka.chat.backend.domain.ChatRoom;
-import de.juplo.kafka.chat.backend.domain.UnknownChatroomException;
 import de.juplo.kafka.chat.backend.persistence.StorageStrategy;
 import lombok.RequiredArgsConstructor;
 import org.springframework.http.codec.ServerSentEvent;
@@ -10,9 +9,7 @@ import org.springframework.web.bind.annotation.*;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import java.util.Optional;
 import java.util.UUID;
-import java.util.stream.Stream;
 
 
 @RestController
@@ -24,13 +21,13 @@ public class ChatBackendController
 
 
   @PostMapping("create")
-  public ChatRoomTo create(@RequestBody String name)
+  public Mono<ChatRoomTo> create(@RequestBody String name)
   {
-    return ChatRoomTo.from(chatHome.createChatroom(name));
+    return chatHome.createChatroom(name).map(ChatRoomTo::from);
   }
 
   @GetMapping("list")
-  public Stream<ChatRoomTo> list()
+  public Flux<ChatRoomTo> list()
   {
     return chatHome.list().map(chatroom -> ChatRoomTo.from(chatroom));
   }
@@ -40,14 +37,13 @@ public class ChatBackendController
   {
     return chatHome
         .getChatroom(chatroomId)
-        .map(chatroom -> chatroom
+        .flatMapMany(chatroom -> chatroom
             .getMessages()
-            .map(MessageTo::from))
-        .get();
+            .map(MessageTo::from));
   }
 
   @GetMapping("get/{chatroomId}")
-  public Optional<ChatRoomTo> get(@PathVariable UUID chatroomId)
+  public Mono<ChatRoomTo> get(@PathVariable UUID chatroomId)
   {
     return chatHome.getChatroom(chatroomId).map(chatroom -> ChatRoomTo.from(chatroom));
   }
@@ -62,8 +58,7 @@ public class ChatBackendController
     return
         chatHome
             .getChatroom(chatroomId)
-            .map(chatroom -> put(chatroom, username, messageId, text))
-            .orElseThrow(() -> new UnknownChatroomException(chatroomId));
+            .flatMap(chatroom -> put(chatroom, username, messageId, text));
   }
 
   public Mono<MessageTo> put(
@@ -91,8 +86,7 @@ public class ChatBackendController
     return
         chatHome
             .getChatroom(chatroomId)
-            .map(chatroom -> get(chatroom, username, messageId))
-            .orElseThrow(() -> new UnknownChatroomException(chatroomId));
+            .flatMap(chatroom -> get(chatroom, username, messageId));
   }
 
   private Mono<MessageTo> get(
@@ -111,8 +105,7 @@ public class ChatBackendController
   {
     return chatHome
         .getChatroom(chatroomId)
-        .map(chatroom -> listen(chatroom))
-        .orElseThrow(() -> new UnknownChatroomException(chatroomId));
+        .flatMapMany(chatroom -> listen(chatroom));
   }
 
   private Flux<ServerSentEvent<MessageTo>> listen(ChatRoom chatroom)
@@ -132,6 +125,6 @@ public class ChatBackendController
   @PostMapping("/store")
   public void store()
   {
-    storageStrategy.writeChatrooms(Flux.fromStream(chatHome.list()));
+    storageStrategy.writeChatrooms(chatHome.list());
   }
 }

diff --git a/src/main/java/de/juplo/kafka/chat/backend/domain/ChatHome.java b/src/main/java/de/juplo/kafka/chat/backend/domain/ChatHome.java
index b31dede..13f18b9 100644 (file)
--- a/src/main/java/de/juplo/kafka/chat/backend/domain/ChatHome.java
+++ b/src/main/java/de/juplo/kafka/chat/backend/domain/ChatHome.java
@@ -2,9 +2,9 @@ package de.juplo.kafka.chat.backend.domain;
 
 import lombok.extern.slf4j.Slf4j;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import java.util.*;
-import java.util.stream.Stream;
 
 
 @Slf4j
@@ -21,20 +21,22 @@ public class ChatHome
     chatroomFlux.subscribe(chatroom -> chatrooms.put(chatroom.getId(), chatroom));
   }
 
-  public ChatRoom createChatroom(String name)
+  public Mono<ChatRoom> createChatroom(String name)
   {
     ChatRoom chatroom = service.createChatroom(UUID.randomUUID(), name);
     chatrooms.put(chatroom.getId(), chatroom);
-    return chatroom;
+    return Mono.justOrEmpty(chatroom);
   }
 
-  public Optional<ChatRoom> getChatroom(UUID id)
+  public Mono<ChatRoom> getChatroom(UUID id)
   {
-    return Optional.ofNullable(chatrooms.get(id));
+    return Mono
+        .justOrEmpty(chatrooms.get(id))
+        .or(Mono.error(() -> new UnknownChatroomException(id)));
   }
 
-  public Stream<ChatRoom> list()
+  public Flux<ChatRoom> list()
   {
-    return chatrooms.values().stream();
+    return Flux.fromStream(chatrooms.values().stream());
   }
 }

diff --git a/src/test/java/de/juplo/kafka/chat/backend/api/ChatBackendControllerTest.java b/src/test/java/de/juplo/kafka/chat/backend/api/ChatBackendControllerTest.java
index 08361dd..404fab2 100644 (file)
--- a/src/test/java/de/juplo/kafka/chat/backend/api/ChatBackendControllerTest.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/api/ChatBackendControllerTest.java
@@ -2,7 +2,6 @@ package de.juplo.kafka.chat.backend.api;
 
 import de.juplo.kafka.chat.backend.domain.*;
 import lombok.extern.slf4j.Slf4j;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -11,6 +10,7 @@ import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.MockBean;
 import org.springframework.http.MediaType;
 import org.springframework.test.web.reactive.server.WebTestClient;
+import reactor.core.publisher.Mono;
 
 import java.time.LocalDateTime;
 import java.util.Optional;
@@ -29,14 +29,14 @@ public class ChatBackendControllerTest
   @MockBean
   ChatHome chatHome;
 
-  @Disabled
   @Test
   @DisplayName("Assert expected problem-details for unknown chatroom on GET /list/{chatroomId}")
   void testUnknownChatroomExceptionForListChatroom(@Autowired WebTestClient client)
   {
     // Given
     UUID chatroomId = UUID.randomUUID();
-    when(chatHome.getChatroom(any(UUID.class))).thenReturn(Optional.empty());
+    when(chatHome.getChatroom(any(UUID.class)))
+        .thenReturn(Mono.error(() -> new UnknownChatroomException(chatroomId)));
 
     // When
     WebTestClient.ResponseSpec responseSpec = client
@@ -50,14 +50,14 @@ public class ChatBackendControllerTest
   }
 
 
-  @Disabled
   @Test
   @DisplayName("Assert expected problem-details for unknown chatroom on GET /get/{chatroomId}")
   void testUnknownChatroomExceptionForGetChatroom(@Autowired WebTestClient client)
   {
     // Given
     UUID chatroomId = UUID.randomUUID();
-    when(chatHome.getChatroom(any(UUID.class))).thenReturn(Optional.empty());
+    when(chatHome.getChatroom(any(UUID.class)))
+        .thenReturn(Mono.error(() -> new UnknownChatroomException(chatroomId)));
 
     // When
     WebTestClient.ResponseSpec responseSpec = client
@@ -78,7 +78,8 @@ public class ChatBackendControllerTest
     UUID chatroomId = UUID.randomUUID();
     String username = "foo";
     Long messageId = 66l;
-    when(chatHome.getChatroom(any(UUID.class))).thenReturn(Optional.empty());
+    when(chatHome.getChatroom(any(UUID.class)))
+        .thenReturn(Mono.error(() -> new UnknownChatroomException(chatroomId)));
 
     // When
     WebTestClient.ResponseSpec responseSpec = client
@@ -104,7 +105,8 @@ public class ChatBackendControllerTest
     UUID chatroomId = UUID.randomUUID();
     String username = "foo";
     Long messageId = 66l;
-    when(chatHome.getChatroom(any(UUID.class))).thenReturn(Optional.empty());
+    when(chatHome.getChatroom(any(UUID.class)))
+        .thenReturn(Mono.error(() -> new UnknownChatroomException(chatroomId)));
 
     // When
     WebTestClient.ResponseSpec responseSpec = client
@@ -127,13 +129,14 @@ public class ChatBackendControllerTest
   {
     // Given
     UUID chatroomId = UUID.randomUUID();
-    when(chatHome.getChatroom(any(UUID.class))).thenReturn(Optional.empty());
+    when(chatHome.getChatroom(any(UUID.class)))
+        .thenReturn(Mono.error(() -> new UnknownChatroomException(chatroomId)));
 
     // When
     WebTestClient.ResponseSpec responseSpec = client
         .get()
         .uri("/listen/{chatroomId}", chatroomId)
-        .accept(MediaType.TEXT_EVENT_STREAM, MediaType.APPLICATION_JSON)
+        // .accept(MediaType.TEXT_EVENT_STREAM, MediaType.APPLICATION_JSON) << TODO: Does not work!
         .exchange();
 
     // Then
@@ -161,7 +164,7 @@ public class ChatBackendControllerTest
     Long messageId = 66l;
     ChatRoom chatRoom = mock(ChatRoom.class);
     when(chatHome.getChatroom(any(UUID.class)))
-        .thenReturn(Optional.of(chatRoom));
+        .thenReturn(Mono.just(chatRoom));
     Message.MessageKey key = Message.MessageKey.of("foo", 1l);
     LocalDateTime timestamp = LocalDateTime.now();
     Message mutated = new Message(key, 0l, timestamp, "Mutated!");

diff --git a/src/test/java/de/juplo/kafka/chat/backend/persistence/LocalJsonFilesStorageStrategyIT.java b/src/test/java/de/juplo/kafka/chat/backend/persistence/LocalJsonFilesStorageStrategyIT.java
index c769222..a4ff04f 100644 (file)
--- a/src/test/java/de/juplo/kafka/chat/backend/persistence/LocalJsonFilesStorageStrategyIT.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/persistence/LocalJsonFilesStorageStrategyIT.java
@@ -9,7 +9,6 @@ import de.juplo.kafka.chat.backend.domain.Message;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import reactor.core.publisher.Flux;
 
 import java.io.IOException;
 import java.nio.file.Files;
@@ -43,7 +42,7 @@ public class LocalJsonFilesStorageStrategyIT
 
   void stop()
   {
-    storageStrategy.writeChatrooms(Flux.fromStream(chathome.list()));
+    storageStrategy.writeChatrooms(chathome.list());
   }
 
   @Test
@@ -51,30 +50,28 @@ public class LocalJsonFilesStorageStrategyIT
   {
     start();
 
-    assertThat(chathome.list()).hasSize(0);
+    assertThat(chathome.list().toStream()).hasSize(0);
 
-    ChatRoom chatroom = chathome.createChatroom("FOO");
+    ChatRoom chatroom = chathome.createChatroom("FOO").block();
     Message m1 = chatroom.addMessage(1l,"Peter", "Hallo, ich heiße Peter!").block();
     Message m2 = chatroom.addMessage(1l, "Ute", "Ich bin Ute...").block();
     Message m3 = chatroom.addMessage(2l, "Peter", "Willst du mit mir gehen?").block();
     Message m4 = chatroom.addMessage(1l, "Klaus", "Ja? Nein? Vielleicht??").block();
 
-    assertThat(chathome.list()).containsExactlyElementsOf(List.of(chatroom));
-    assertThat(chathome.getChatroom(chatroom.getId())).contains(chatroom);
+    assertThat(chathome.list().toStream()).containsExactlyElementsOf(List.of(chatroom));
+    assertThat(chathome.getChatroom(chatroom.getId())).emitsExactly(chatroom);
     assertThat(chathome
         .getChatroom(chatroom.getId())
-        .get()
-        .getMessages()).emitsExactly(m1, m2, m3, m4);
+        .flatMapMany(cr -> cr.getMessages())).emitsExactly(m1, m2, m3, m4);
 
     stop();
     start();
 
-    assertThat(chathome.list()).containsExactlyElementsOf(List.of(chatroom));
-    assertThat(chathome.getChatroom(chatroom.getId())).contains(chatroom);
+    assertThat(chathome.list().toStream()).containsExactlyElementsOf(List.of(chatroom));
+    assertThat(chathome.getChatroom(chatroom.getId())).emitsExactly(chatroom);
     assertThat(chathome
         .getChatroom(chatroom.getId())
-        .get()
-        .getMessages()).emitsExactly(m1, m2, m3, m4);
+        .flatMapMany(cr -> cr.getMessages())).emitsExactly(m1, m2, m3, m4);
   }
 
   @BeforeEach