From: Kai Moritz Date: Sun, 8 Jan 2023 09:34:02 +0000 (+0100) Subject: refactore: Renamed `ChatroomService` to `ChatRoomService` -- Move X-Git-Tag: wip~69 X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=c1e01738f2a0b941f1a9f20fcc982cb4de6ab8cd;p=demos%2Fkafka%2Fchat refactore: Renamed `ChatroomService` to `ChatRoomService` -- Move --- diff --git a/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoomService.java b/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoomService.java new file mode 100644 index 00000000..1e7e8c72 --- /dev/null +++ b/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoomService.java @@ -0,0 +1,19 @@ +package de.juplo.kafka.chat.backend.domain; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.time.LocalDateTime; + + +public interface ChatroomService +{ + Mono persistMessage( + Message.MessageKey key, + LocalDateTime timestamp, + String text); + + Mono getMessage(Message.MessageKey key); + + Flux getMessages(long first, long last); +} diff --git a/src/main/java/de/juplo/kafka/chat/backend/domain/ChatroomService.java b/src/main/java/de/juplo/kafka/chat/backend/domain/ChatroomService.java deleted file mode 100644 index 1e7e8c72..00000000 --- a/src/main/java/de/juplo/kafka/chat/backend/domain/ChatroomService.java +++ /dev/null @@ -1,19 +0,0 @@ -package de.juplo.kafka.chat.backend.domain; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import java.time.LocalDateTime; - - -public interface ChatroomService -{ - Mono persistMessage( - Message.MessageKey key, - LocalDateTime timestamp, - String text); - - Mono getMessage(Message.MessageKey key); - - Flux getMessages(long first, long last); -} diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/InMemoryChatRoomService.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/InMemoryChatRoomService.java new file mode 100644 index 00000000..e06709ff --- /dev/null +++ b/src/main/java/de/juplo/kafka/chat/backend/persistence/InMemoryChatRoomService.java @@ -0,0 +1,76 @@ +package de.juplo.kafka.chat.backend.persistence; + +import de.juplo.kafka.chat.backend.domain.Message; +import de.juplo.kafka.chat.backend.domain.MessageMutationException; +import de.juplo.kafka.chat.backend.domain.ChatroomService; +import lombok.extern.slf4j.Slf4j; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.time.LocalDateTime; +import java.util.LinkedHashMap; + + +@Slf4j +public class InMemoryChatroomService implements ChatroomService +{ + private final LinkedHashMap messages; + + + public InMemoryChatroomService(LinkedHashMap messages) + { + this.messages = messages; + } + + public InMemoryChatroomService(Flux messageFlux) + { + log.debug("Creating InMemoryChatroomService"); + messages = new LinkedHashMap<>(); + messageFlux.subscribe(message -> persistMessage(message)); + } + + @Override + public Mono persistMessage( + Message.MessageKey key, + LocalDateTime timestamp, + String text) + { + Message message = new Message(key, (long)messages.size(), timestamp, text); + return Mono.justOrEmpty(persistMessage(message)); + } + + private Message persistMessage(Message message) + { + Message.MessageKey key = message.getKey(); + Message existing = messages.get(key); + if (existing != null) + { + log.info("Message with key {} already exists; {}", key, existing); + if (!message.equals(existing)) + throw new MessageMutationException(message, existing); + return null; + } + + messages.put(key, message); + return message; + } + + @Override + public Mono getMessage(Message.MessageKey key) + { + return Mono.fromSupplier(() -> messages.get(key)); + } + + @Override + public Flux getMessages(long first, long last) + { + return Flux.fromStream(messages + .values() + .stream() + .filter(message -> + { + long serial = message.getSerialNumber(); + return serial >= first && serial <= last; + })); + } +} diff --git a/src/main/java/de/juplo/kafka/chat/backend/persistence/InMemoryChatroomService.java b/src/main/java/de/juplo/kafka/chat/backend/persistence/InMemoryChatroomService.java deleted file mode 100644 index e06709ff..00000000 --- a/src/main/java/de/juplo/kafka/chat/backend/persistence/InMemoryChatroomService.java +++ /dev/null @@ -1,76 +0,0 @@ -package de.juplo.kafka.chat.backend.persistence; - -import de.juplo.kafka.chat.backend.domain.Message; -import de.juplo.kafka.chat.backend.domain.MessageMutationException; -import de.juplo.kafka.chat.backend.domain.ChatroomService; -import lombok.extern.slf4j.Slf4j; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import java.time.LocalDateTime; -import java.util.LinkedHashMap; - - -@Slf4j -public class InMemoryChatroomService implements ChatroomService -{ - private final LinkedHashMap messages; - - - public InMemoryChatroomService(LinkedHashMap messages) - { - this.messages = messages; - } - - public InMemoryChatroomService(Flux messageFlux) - { - log.debug("Creating InMemoryChatroomService"); - messages = new LinkedHashMap<>(); - messageFlux.subscribe(message -> persistMessage(message)); - } - - @Override - public Mono persistMessage( - Message.MessageKey key, - LocalDateTime timestamp, - String text) - { - Message message = new Message(key, (long)messages.size(), timestamp, text); - return Mono.justOrEmpty(persistMessage(message)); - } - - private Message persistMessage(Message message) - { - Message.MessageKey key = message.getKey(); - Message existing = messages.get(key); - if (existing != null) - { - log.info("Message with key {} already exists; {}", key, existing); - if (!message.equals(existing)) - throw new MessageMutationException(message, existing); - return null; - } - - messages.put(key, message); - return message; - } - - @Override - public Mono getMessage(Message.MessageKey key) - { - return Mono.fromSupplier(() -> messages.get(key)); - } - - @Override - public Flux getMessages(long first, long last) - { - return Flux.fromStream(messages - .values() - .stream() - .filter(message -> - { - long serial = message.getSerialNumber(); - return serial >= first && serial <= last; - })); - } -}