From ce7f5a067b0af16442255903cb06e97dfa4fd352 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Wed, 25 Jan 2023 18:51:38 +0100 Subject: [PATCH] WIP:refactor --- .../kafka/chat/backend/domain/ChatRoom.java | 40 +++++++++---------- 1 file changed, 20 insertions(+), 20 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 9ea4cef3..da5eba2a 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 @@ -66,30 +66,30 @@ public class ChatRoom throw new InvalidUsernameException(user); Message.MessageKey key = Message.MessageKey.of(user, id); - return service - .getMessage(key) - .handle((Message existing, SynchronousSink sink) -> - { - if (existing.getMessageText().equals(text)) - { - sink.next(existing); - } - else - { - sink.error(new MessageMutationException(existing, text)); - } - }) - .switchIfEmpty(Mono.defer(() -> service - .persistMessage(key, LocalDateTime.now(clock), text) - .doOnNext(m -> + .getMessage(key) + .handle((Message existing, SynchronousSink sink) -> { - Sinks.EmitResult result = sink.tryEmitNext(m); - if (result.isFailure()) + if (existing.getMessageText().equals(text)) + { + sink.next(existing); + } + else { - log.warn("Emitting of message failed with {} for {}", result.name(), m); + sink.error(new MessageMutationException(existing, text)); } - }))); + }) + .switchIfEmpty( + Mono + .defer(() -> service.persistMessage(key, LocalDateTime.now(clock), text)) + .doOnNext(m -> + { + Sinks.EmitResult result = sink.tryEmitNext(m); + if (result.isFailure()) + { + log.warn("Emitting of message failed with {} for {}", result.name(), m); + } + })); } -- 2.20.1