From 974327a0e3a970af36e3ace4a8fb17d8a800b4a3 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Wed, 21 Dec 2022 18:57:23 +0100
Subject: [PATCH] feat: implemented a listen-method for the chat-service based
 on a Flux

- Switched the return-type of `Chatroom.addMessage()` to `Mono<Message>`.
- Added an inner method, that "persists" the message and also returns
  `Mono<Message>`.
- `addMessage()` calls `persistMessage()`, peeks into the subscription and
  emits the `Message` to an internal `Sink.multy()` of the `Chatroom`.
- `Chatroom.listen()` creates a `Flux` from the internal Sink, so that
  multiple subscribers can listen on the sink and all retrieve the current
  messages in parallel.
---
 pom.xml                                       |  2 +-
 .../chatroom/api/ChatroomController.java      | 22 ++++++++++++---
 .../juplo/kafka/chatroom/domain/Chatroom.java | 27 +++++++++++++++++--
 3 files changed, 44 insertions(+), 7 deletions(-)

diff --git a/pom.xml b/pom.xml
index 98ef4afd..22d1810b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -19,7 +19,7 @@
   <dependencies>
     <dependency>
       <groupId>org.springframework.boot</groupId>
-      <artifactId>spring-boot-starter-web</artifactId>
+      <artifactId>spring-boot-starter-webflux</artifactId>
     </dependency>
 
     <dependency>
diff --git a/src/main/java/de/juplo/kafka/chatroom/api/ChatroomController.java b/src/main/java/de/juplo/kafka/chatroom/api/ChatroomController.java
index e3146a9d..5fabf363 100644
--- a/src/main/java/de/juplo/kafka/chatroom/api/ChatroomController.java
+++ b/src/main/java/de/juplo/kafka/chatroom/api/ChatroomController.java
@@ -3,10 +3,15 @@ package de.juplo.kafka.chatroom.api;
 import de.juplo.kafka.chatroom.domain.Chatroom;
 import de.juplo.kafka.chatroom.domain.Message;
 import lombok.RequiredArgsConstructor;
+import org.springframework.http.MediaType;
 import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import java.time.Clock;
+import java.time.Duration;
 import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
@@ -36,19 +41,28 @@ public class ChatroomController
   }
 
   @PutMapping("post/{chatroomId}/{username}/{messageId}")
-  public MessageTo post(
+  public Mono<MessageTo> post(
       @PathVariable UUID chatroomId,
       @PathVariable String username,
       @PathVariable UUID messageId,
-      @RequestBody String message)
+      @RequestBody String text)
   {
-    return MessageTo.from(
+    return
         chatrooms
             .get(chatroomId)
             .addMessage(
                 messageId,
                 LocalDateTime.now(clock),
                 username,
-                message));
+                text)
+            .map(message -> MessageTo.from(message));
+  }
+
+  @GetMapping(
+      path = "listen/{chatroomId}",
+      produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+  public Flux<Message> listen(@PathVariable UUID chatroomId)
+  {
+    return chatrooms.get(chatroomId).listen();
   }
 }
diff --git a/src/main/java/de/juplo/kafka/chatroom/domain/Chatroom.java b/src/main/java/de/juplo/kafka/chatroom/domain/Chatroom.java
index fd7d73aa..f87a0883 100644
--- a/src/main/java/de/juplo/kafka/chatroom/domain/Chatroom.java
+++ b/src/main/java/de/juplo/kafka/chatroom/domain/Chatroom.java
@@ -2,6 +2,10 @@ package de.juplo.kafka.chatroom.domain;
 
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
 
 import java.time.LocalDateTime;
 import java.util.LinkedList;
@@ -11,6 +15,7 @@ import java.util.stream.Stream;
 
 
 @RequiredArgsConstructor
+@Slf4j
 public class Chatroom
 {
   @Getter
@@ -18,8 +23,19 @@ public class Chatroom
   @Getter
   private final String name;
   private final List<Message> messages = new LinkedList<>();
+  private final Sinks.Many<Message> sink = Sinks.many().multicast().onBackpressureBuffer();
 
-  synchronized public Message addMessage(
+  synchronized public Mono<Message> addMessage(
+      UUID id,
+      LocalDateTime timestamp,
+      String user,
+      String text)
+  {
+    return persistMessage(id, timestamp, user, text)
+        .doOnNext(message -> sink.tryEmitNext(message).orThrow());
+  }
+
+  private Mono<Message> persistMessage(
       UUID id,
       LocalDateTime timestamp,
       String user,
@@ -27,7 +43,14 @@ public class Chatroom
   {
     Message message = new Message(id, (long)messages.size(), timestamp, user, text);
     messages.add(message);
-    return message;
+    return Mono
+        .fromSupplier(() -> message)
+        .log();
+  }
+
+  public Flux<Message> listen()
+  {
+    return sink.asFlux();
   }
 
   public Stream<Message> getMessages(long firstMessage)
-- 
2.20.1