WIP

[demos/kafka/chat] / src / main / java / de / juplo / kafka / chat / backend / persistence / kafka / KafkaChatRoomService.java

diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatRoomService.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatRoomService.java
new file mode 100644 (file)
index 0000000..981c11f
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/kafka/KafkaChatRoomService.java
@@ -0,0 +1,60 @@
+package de.juplo.kafka.chat.backend.persistence.kafka;
+
+import de.juplo.kafka.chat.backend.domain.ChatRoomService;
+import de.juplo.kafka.chat.backend.domain.Message;
+import de.juplo.kafka.chat.backend.domain.MessageMutationException;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.TopicPartition;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.time.LocalDateTime;
+import java.util.LinkedHashMap;
+
+
+@Slf4j
+@RequiredArgsConstructor
+public class KafkaChatRoomService implements ChatRoomService
+{
+  private final Producer<String, MessageTo> producer;
+  private final TopicPartition tp;
+
+  private final LinkedHashMap<Message.MessageKey, Message> messages = new LinkedHashMap<>();
+
+  private long offset = 0l;
+
+
+  @Override
+  public Message persistMessage(
+    Message.MessageKey key,
+    LocalDateTime timestamp,
+    String text)
+  {
+    
+    Mono.error(() -> new MessageMutationException(existing, text)));
+    Message message = new Message(key, (long)messages.size(), timestamp, text);
+    messages.put(message.getKey(), message);
+    return message;
+  }
+
+  @Override
+  public Mono<Message> getMessage(Message.MessageKey key)
+  {
+    return Mono.fromSupplier(() -> messages.get(key));
+  }
+
+  @Override
+  public Flux<Message> getMessages(long first, long last)
+  {
+    return Flux.fromStream(messages
+      .values()
+      .stream()
+      .filter(message ->
+      {
+        long serial = message.getSerialNumber();
+        return serial >= first && serial <= last;
+      }));
+  }
+}