From 41e0a7cfd0025d5bea9722c8fc10c3b20a0fcad7 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Wed, 25 Jan 2023 22:08:11 +0100 Subject: [PATCH] refactor: Simplified the handling of `MessageMutationException` --- .../juplo/kafka/chat/backend/domain/ChatRoom.java | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoom.java b/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoom.java index 4f855b8c..02f5c08c 100644 --- a/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoom.java +++ b/src/main/java/de/juplo/kafka/chat/backend/domain/ChatRoom.java @@ -7,6 +7,7 @@ import lombok.extern.slf4j.Slf4j; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; +import reactor.core.publisher.SynchronousSink; import java.time.Clock; import java.time.LocalDateTime; @@ -67,9 +68,17 @@ public class ChatRoom Message.MessageKey key = Message.MessageKey.of(user, id); return service .getMessage(key) - .flatMap(existing -> text.equals(existing.getMessageText()) - ? Mono.just(existing) - : Mono.error(() -> new MessageMutationException(existing, text))) + .handle((Message existing, SynchronousSink sink) -> + { + if (existing.getMessageText().equals(text)) + { + sink.next(existing); + } + else + { + sink.error(new MessageMutationException(existing, text)); + } + }) .switchIfEmpty( Mono .fromSupplier(() ->service.persistMessage(key, LocalDateTime.now(clock), text)) -- 2.20.1