juplo.de Git - demos/kafka/chat/blob - src/test/java/de/juplo/kafka/chat/backend/domain/ChatRoomDataTest.java

   1 package de.juplo.kafka.chat.backend.domain;
   2
   3 import org.awaitility.Awaitility;
   4 import org.junit.jupiter.api.BeforeEach;
   5 import org.junit.jupiter.api.DisplayName;
   6 import org.junit.jupiter.api.Test;
   7 import reactor.core.publisher.Flux;
   8 import reactor.core.publisher.Mono;
   9
  10 import java.time.*;
  11 import java.util.LinkedList;
  12 import java.util.List;
  13
  14 import static org.mockito.Mockito.*;
  15 import static pl.rzrz.assertj.reactor.Assertions.assertThat;
  16
  17
  18 public class ChatRoomDataTest
  19 {
  20   Clock now;
  21   ChatMessageService chatMessageService;
  22   ChatRoomData chatRoomData;
  23
  24   String user;
  25   Long messageId;
  26   Message.MessageKey key;
  27   LocalDateTime timestamp;
  28
  29
  30   @BeforeEach
  31   public void setUp()
  32   {
  33     now = Clock.fixed(Instant.now(), ZoneId.systemDefault());
  34     chatMessageService = mock(ChatMessageService.class);
  35     chatRoomData = new ChatRoomData(
  36         now,
  37         chatMessageService,
  38         8);
  39     chatRoomData.activate();
  40
  41     user = "foo";
  42     messageId = 1l;
  43     key = Message.MessageKey.of(user, messageId);
  44     timestamp = LocalDateTime.now(now);
  45   }
  46
  47
  48   @Test
  49   @DisplayName("Assert, that Mono emits expected message, if it exists")
  50   void testGetExistingMessage()
  51   {
  52     // Given
  53     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
  54         .thenReturn(Mono.just(someMessage()));
  55
  56     // When
  57     Mono<Message> mono = chatRoomData.getMessage(user, messageId);
  58
  59     // Then
  60     assertThat(mono).emitsExactly(someMessage());
  61   }
  62
  63   @Test
  64   @DisplayName("Assert, that Mono is empty, if message does not exists")
  65   void testGetNonExistentMessage()
  66   {
  67     // Given
  68     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
  69         .thenReturn(Mono.empty());
  70
  71     // When
  72     Mono<Message> mono = chatRoomData.getMessage(user, messageId);
  73
  74     // Then
  75     assertThat(mono).emitsCount(0);
  76   }
  77
  78   @Test
  79   @DisplayName("Assert, that Mono emits the persisted message, if a new message is added")
  80   void testAddNewMessageEmitsPersistedMessage()
  81   {
  82     // Given
  83     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
  84         .thenReturn(Mono.empty());
  85     when(chatMessageService.persistMessage(any(Message.MessageKey.class), any(LocalDateTime.class), any(String.class)))
  86         .thenReturn(Mono.just(someMessage()));
  87
  88     // When
  89     Mono<Message> mono = chatRoomData.addMessage(messageId, user, "Some Text");
  90
  91     // Then
  92     assertThat(mono).emitsExactly(someMessage());
  93   }
  94
  95   @Test
  96   @DisplayName("Assert, that ChatMessageService.persistMessage() is called correctly, if a new message is added")
  97   void testAddNewMessageTriggersPersistence()
  98   {
  99     // Given
 100     String messageText = "Bar";
 101     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 102         .thenReturn(Mono.empty());
 103     when(chatMessageService.persistMessage(any(Message.MessageKey.class), any(LocalDateTime.class), any(String.class)))
 104         .thenReturn(Mono.just(someMessage()));
 105
 106     // When
 107     chatRoomData
 108         .addMessage(messageId, user, messageText)
 109         .block();
 110
 111     // Then
 112     verify(chatMessageService, times(1)).persistMessage(eq(key), eq(timestamp), eq(messageText));
 113   }
 114
 115   @Test
 116   @DisplayName("Assert, that Mono emits the already persisted message, if an unchanged message is added")
 117   void testAddUnchangedMessageEmitsAlreadyPersistedMessage()
 118   {
 119     // Given
 120     String messageText = "Bar";
 121     Message existingMessage = new Message(key, 0l, timestamp, messageText);
 122     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 123         .thenReturn(Mono.just(existingMessage));
 124
 125     // When
 126     Mono<Message> mono = chatRoomData.addMessage(messageId, user, messageText);
 127
 128     // Then
 129     assertThat(mono).emitsExactly(existingMessage);
 130   }
 131
 132   @Test
 133   @DisplayName("Assert, that ChatMessageService.persistMessage() is not called, if an unchanged message is added")
 134   void testAddUnchangedMessageDoesNotTriggerPersistence()
 135   {
 136     // Given
 137     String messageText = "Bar";
 138     Message existingMessage = new Message(key, 0l, timestamp, messageText);
 139     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 140         .thenReturn(Mono.just(existingMessage));
 141
 142     // When
 143     chatRoomData
 144         .addMessage(messageId, user, messageText)
 145         .block();
 146
 147     // Then
 148     verify(chatMessageService, never()).persistMessage(any(), any(), any());
 149   }
 150
 151   @Test
 152   @DisplayName("Assert, that Mono sends an error, if a message is added again with mutated text")
 153   void testAddMutatedMessageSendsError()
 154   {
 155     // Given
 156     String messageText = "Bar";
 157     String mutatedText = "Boom!";
 158     Message existingMessage = new Message(key, 0l, timestamp, messageText);
 159     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 160         .thenReturn(Mono.just(existingMessage));
 161
 162     // When
 163     Mono<Message> mono = chatRoomData.addMessage(messageId, user, mutatedText);
 164
 165     // Then
 166     assertThat(mono).sendsError();
 167   }
 168
 169   @Test
 170   @DisplayName("Assert, that ChatMessageService.persistMessage() is not called, if a message is added again with mutated text")
 171   void testAddMutatedDoesNotTriggerPersistence()
 172   {
 173     // Given
 174     String messageText = "Bar";
 175     String mutatedText = "Boom!";
 176     Message existingMessage = new Message(key, 0l, timestamp, messageText);
 177     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 178         .thenReturn(Mono.just(existingMessage));
 179
 180     // When
 181     chatRoomData
 182         .addMessage(messageId, user, mutatedText)
 183         .onErrorResume((throwable) -> Mono.empty())
 184         .block();
 185
 186     // Then
 187     verify(chatMessageService, never()).persistMessage(any(), any(), any());
 188   }
 189
 190   @Test
 191   @DisplayName("Assert, that Mono sends an error, if a message is sent to a closed chat-room")
 192   void testAddMessageToClosedChatRoomSendsError()
 193   {
 194     // Given
 195     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 196         .thenReturn(Mono.empty());
 197     when(chatMessageService.persistMessage(any(Message.MessageKey.class), any(LocalDateTime.class), any(String.class)))
 198         .thenReturn(Mono.just(someMessage()));
 199
 200     chatRoomData.deactivate();
 201
 202     // When
 203     Mono<Message> mono = chatRoomData.addMessage(messageId, user, "Some text");
 204
 205     // Then
 206     assertThat(mono).sendsError();
 207   }
 208
 209   @Test
 210   @DisplayName("Assert, that ChatMessageService.persistMessage() is not called if a message is sent to a closed chat-room")
 211   void testAddMessageToClosedChatRoomDoesNotTriggerPersistence()
 212   {
 213     // Given
 214     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 215         .thenReturn(Mono.empty());
 216     when(chatMessageService.persistMessage(any(Message.MessageKey.class), any(LocalDateTime.class), any(String.class)))
 217         .thenReturn(Mono.just(someMessage()));
 218
 219     chatRoomData.deactivate();
 220
 221     // When
 222     chatRoomData
 223         .addMessage(messageId, user, "Some text")
 224         .onErrorResume((throwable) -> Mono.empty())
 225         .block();
 226
 227     // Then
 228     verify(chatMessageService, never()).persistMessage(any(), any(), any());
 229   }
 230
 231   @Test
 232   @DisplayName("Assert, that a listener receives a message, that was added after the listening had started")
 233   void testListenerReceivesMessageAddedAfterListeningStarts()
 234   {
 235     // Given
 236     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 237         .thenReturn(Mono.empty());
 238     when(chatMessageService.persistMessage(any(Message.MessageKey.class), any(LocalDateTime.class), any(String.class)))
 239         .thenReturn(Mono.just(someMessage()));
 240
 241     // When
 242     List<Message> receivedMessages = new LinkedList<>();
 243     chatRoomData
 244         .listen()
 245         .subscribe(receivedMessage -> receivedMessages.add(receivedMessage));
 246     Message sentMessage = chatRoomData
 247         .addMessage(messageId, user, "Some Text")
 248         .block();
 249
 250     // Then
 251     Awaitility
 252         .await()
 253         .atMost(Duration.ofSeconds(1))
 254         .untilAsserted(() -> assertThat(receivedMessages).contains(sentMessage));
 255   }
 256
 257   @Test
 258   @DisplayName("Assert, that a listener receives a message, that was added before the listening had started")
 259   void testListenerReceivesMessageAddedBeforeListeningStarts()
 260   {
 261     // Given
 262     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 263         .thenReturn(Mono.empty());
 264     when(chatMessageService.persistMessage(any(Message.MessageKey.class), any(LocalDateTime.class), any(String.class)))
 265         .thenReturn(Mono.just(someMessage()));
 266
 267     // When
 268     Message sentMessage = chatRoomData
 269         .addMessage(messageId, user, "Some Text")
 270         .block();
 271     List<Message> receivedMessages = new LinkedList<>();
 272     chatRoomData
 273         .listen()
 274         .subscribe(receivedMessage -> receivedMessages.add(receivedMessage));
 275
 276     // Then
 277     Awaitility
 278         .await()
 279         .atMost(Duration.ofSeconds(1))
 280         .untilAsserted(() -> assertThat(receivedMessages).contains(sentMessage));
 281   }
 282
 283   @Test
 284   @DisplayName("Assert, that a listener receives several messages, that were added before and after the listening had started, in correct order")
 285   void testListenerReceivesMessagesFromBeforeAndAfterListeningHadStartedInCorrectOrder()
 286   {
 287     // Given
 288     Message message1 = new Message(key, 1l, timestamp, "#1");
 289     Message message2 = new Message(key, 2l, timestamp, "#2");
 290     Message message3 = new Message(key, 3l, timestamp, "#3");
 291     Message message4 = new Message(key, 4l, timestamp, "#4");
 292     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 293         .thenReturn(Mono.empty());
 294     when(chatMessageService.persistMessage(any(Message.MessageKey.class), any(LocalDateTime.class), any(String.class)))
 295         .thenReturn(Mono.just(message1))
 296         .thenReturn(Mono.just(message2))
 297         .thenReturn(Mono.just(message3))
 298         .thenReturn(Mono.just(message4));
 299
 300     // When
 301     Message[] sentMessages = new Message[4];
 302     sentMessages[0] = chatRoomData.addMessage(messageId, user, "Some Text").block();
 303     sentMessages[1] = chatRoomData.addMessage(messageId, user, "Some Text").block();
 304     List<Message> receivedMessages = new LinkedList<>();
 305     chatRoomData
 306         .listen()
 307         .subscribe(receivedMessage -> receivedMessages.add(receivedMessage));
 308     sentMessages[2] = chatRoomData.addMessage(messageId, user, "Some Text").block();
 309     sentMessages[3] = chatRoomData.addMessage(messageId, user, "Some Text").block();
 310
 311     // Then
 312     Awaitility
 313         .await()
 314         .atMost(Duration.ofSeconds(1))
 315         .untilAsserted(() -> assertThat(receivedMessages).contains(sentMessages));
 316   }
 317
 318   @Test
 319   @DisplayName("Assert, that multiple listeners can receive an added message")
 320   void testMultipleListeners()
 321   {
 322     // Given
 323     Message message1 = new Message(key, 1l, timestamp, "#1");
 324     Message message2 = new Message(key, 2l, timestamp, "#2");
 325     Message message3 = new Message(key, 3l, timestamp, "#3");
 326     Message message4 = new Message(key, 4l, timestamp, "#4");
 327     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 328         .thenReturn(Mono.empty());
 329     when(chatMessageService.persistMessage(any(Message.MessageKey.class), any(LocalDateTime.class), any(String.class)))
 330         .thenReturn(Mono.just(message1))
 331         .thenReturn(Mono.just(message2))
 332         .thenReturn(Mono.just(message3))
 333         .thenReturn(Mono.just(message4));
 334
 335     // When
 336     Message[] sentMessages = new Message[4];
 337     List<Message> messagesReceivedByListener1 = new LinkedList<>();
 338     chatRoomData
 339         .listen()
 340         .subscribe(receivedMessage -> messagesReceivedByListener1.add(receivedMessage));
 341     sentMessages[0] = chatRoomData.addMessage(messageId, user, "Some Text").block();
 342     sentMessages[1] = chatRoomData.addMessage(messageId, user, "Some Text").block();
 343     List<Message> messagesReceivedByListener2 = new LinkedList<>();
 344     chatRoomData
 345         .listen()
 346         .subscribe(receivedMessage -> messagesReceivedByListener2.add(receivedMessage));
 347     sentMessages[2] = chatRoomData.addMessage(messageId, user, "Some Text").block();
 348     List<Message> messagesReceivedByListener3 = new LinkedList<>();
 349     chatRoomData
 350         .listen()
 351         .subscribe(receivedMessage -> messagesReceivedByListener3.add(receivedMessage));
 352     sentMessages[3] = chatRoomData.addMessage(messageId, user, "Some Text").block();
 353     List<Message> messagesReceivedByListener4 = new LinkedList<>();
 354     chatRoomData
 355         .listen()
 356         .subscribe(receivedMessage -> messagesReceivedByListener4.add(receivedMessage));
 357
 358     // Then
 359     Awaitility
 360         .await()
 361         .atMost(Duration.ofSeconds(1))
 362         .untilAsserted(() ->
 363         {
 364           assertThat(messagesReceivedByListener1).contains(sentMessages);
 365           assertThat(messagesReceivedByListener2).contains(sentMessages);
 366           assertThat(messagesReceivedByListener3).contains(sentMessages);
 367           assertThat(messagesReceivedByListener4).contains(sentMessages);
 368         });
 369   }
 370
 371   @Test
 372   @DisplayName("Assert, that a listended to chat-room emits completed, if it is closed")
 373   void testListenedToChatRoomEmitsCompletedIfItIsClosed()
 374   {
 375     // Given
 376     Message message1 = new Message(key, 1l, timestamp, "#1");
 377     Message message2 = new Message(key, 2l, timestamp, "#2");
 378     Message message3 = new Message(key, 3l, timestamp, "#3");
 379     Message message4 = new Message(key, 4l, timestamp, "#4");
 380     when(chatMessageService.getMessage(any(Message.MessageKey.class)))
 381         .thenReturn(Mono.empty());
 382     when(chatMessageService.persistMessage(any(Message.MessageKey.class), any(LocalDateTime.class), any(String.class)))
 383         .thenReturn(Mono.just(message1))
 384         .thenReturn(Mono.just(message2))
 385         .thenReturn(Mono.just(message3))
 386         .thenReturn(Mono.just(message4));
 387
 388     chatRoomData.addMessage(messageId, user, "Some Text").block();
 389     chatRoomData.addMessage(messageId, user, "Some Text").block();
 390     chatRoomData.addMessage(messageId, user, "Some Text").block();
 391     chatRoomData.addMessage(messageId, user, "Some Text").block();
 392
 393     // When
 394     Flux<Message> listenFlux = chatRoomData.listen();
 395     chatRoomData.deactivate();
 396
 397     // Then
 398     assertThat(listenFlux).emitsExactly(
 399         message1,
 400         message2,
 401         message3,
 402         message4);
 403   }
 404
 405   @Test
 406   @DisplayName("Assert, that a listended to chat-room emits completed, if it is closed")
 407   void testListeningToClosedChatRoomSendsError()
 408   {
 409     // Given
 410     chatRoomData.deactivate();
 411
 412     // When
 413     Flux<Message> listenFlux = chatRoomData.listen();
 414
 415     // Then
 416     assertThat(listenFlux).sendsError();
 417   }
 418
 419
 420   /**
 421    * This message is used, when methods of {@link ChatMessageService} are mocked,
 422    * that return a {@link Message}.
 423    * The contents of the message are set to arbitrary values, in order to underline
 424    * the fact, that the test can only assert, that the message that was returned
 425    * by {@link ChatMessageService} is handed on by {@link ChatRoomData} correctly.
 426    * @return a message.
 427    */
 428   private Message someMessage()
 429   {
 430     return new Message(
 431         Message.MessageKey.of("FOO", 666l),
 432         666l,
 433         LocalDateTime.of(2024, 3, 8, 12, 13, 00),
 434         "Just some message...");
 435   }
 436 }