improve simple messageBus test and compine with MessageBusService; fi…

…x issue with observationStoppedThrough -> put execution to AutoClosable close() method!

src/main/java/com/recom/api/messagebus/MessageBusController.java

@@ -85,22 +85,24 @@ public ResponseEntity<ResponseBodyEmitter> getMessagesJSON(
         log.debug("Requested POST /api/v1/map/message-bus (JSON)");
         assertionService.assertMapExists(messageBusLongPollRequestDto.getMapName());
 
-        final MessageLongPollObserver messageLongPollObserver = MessageLongPollObserver.builder()
+        try (final MessageLongPollObserver messageLongPollObserver = MessageLongPollObserver.builder()
                 .timeout(RECOM_CURL_TIMEOUT.toMillis())
                 .asyncTaskExecutor(asyncConfiguration.provideClusterGeneratorExecutor())
                 .messagePersistenceLayer(messagePersistenceLayer)
-                .build();
-        messageLongPollObserver.observe(messageBusService.getSubject());
+                .build()
+        ) {
+            messageLongPollObserver.observe(messageBusService.getSubject());
 
-        messageLongPollObserver.scheduleTestResponse(messageBusLongPollRequestDto.getMapName(), Duration.ofSeconds(5), messageBusService.getSubject(), asyncConfiguration);
+            messageLongPollObserver.scheduleTestResponse(messageBusLongPollRequestDto.getMapName(), Duration.ofSeconds(5), messageBusService.getSubject(), asyncConfiguration);
 
-        final HttpHeaders httpHeaders = new HttpHeaders();
-        httpHeaders.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE);
+            final HttpHeaders httpHeaders = new HttpHeaders();
+            httpHeaders.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE);
 
-        return ResponseEntity.status(HttpStatus.OK)
-                .headers(httpHeaders)
-                .cacheControl(CacheControl.noCache())
-                .body(messageLongPollObserver.provideResponseEmitter());
+            return ResponseEntity.status(HttpStatus.OK)
+                    .headers(httpHeaders)
+                    .cacheControl(CacheControl.noCache())
+                    .body(messageLongPollObserver.provideResponseEmitter());
+        }
     }
 
     @Operation(

src/main/java/com/recom/observer/ObserverTemplate.java

@@ -5,13 +5,14 @@
 import java.util.ArrayList;
 import java.util.List;
 
-public abstract class ObserverTemplate<T> implements Observing<T> {
+public abstract class ObserverTemplate<T> implements Observing<T>, AutoCloseable {
 
     @NonNull
     protected final List<Subjective<T>> subjects = new ArrayList<>();
 
     @Override
     public void observe(@NonNull final Subjective<T> subject) {
+        subject.beObservedBy(this);
         subjects.add(subject);
     }
 
@@ -26,4 +27,9 @@ public void takeDeathNoticeFrom(@NonNull final Subjective<T> subject) {
         subjects.remove(subject);
     }
 
+    @Override
+    public void close() {
+        subjects.forEach(subject -> subject.observationStoppedThrough(this));
+    }
+
 }

src/main/java/com/recom/service/messagebus/MessageLongPollObserver.java

@@ -78,7 +78,6 @@ public void takeNotice(
             log.error(e.getMessage(), e);
             responseBodyEmitter.completeWithError(e);
         } finally {
-            subject.observationStoppedThrough(this);
             responseBodyEmitter.complete();
         }
     }

src/test/java/com/recom/service/messagebus/MessageBusServiceTest.java

@@ -0,0 +1,48 @@
+package com.recom.service.messagebus;
+
+import com.recom.model.message.MessageContainer;
+import com.recom.observer.Subjective;
+import com.recom.persistence.message.MessagePersistenceLayer;
+import lombok.NonNull;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+@ExtendWith(MockitoExtension.class)
+class MessageBusServiceTest {
+
+    @Mock
+    @NonNull
+    private MessagePersistenceLayer messagePersistenceLayer;
+    @InjectMocks
+    private MessageBusService serviceUnderTest;
+
+    @Test
+    void sendMessage() {
+        // Arrange
+        // Act
+        serviceUnderTest.sendMessage("mapName", new MessageContainer());
+        // Assert
+    }
+
+    @Test
+    void listMessagesSince() {
+        // Arrange
+        // Act
+        serviceUnderTest.listMessagesSince("mapName", 0L);
+        // Assert
+    }
+
+    @Test
+    void getSubject() {
+        // Arrange
+        // Act
+        final Subjective<MessageContainer> resultToTest = serviceUnderTest.getSubject();
+        // Assert
+        assertNotNull(resultToTest);
+    }
+}

src/test/java/com/recom/service/messagebus/MessageLongPollObserverTest.java

@@ -1,13 +1,11 @@
 package com.recom.service.messagebus;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.recom.configuration.AsyncConfiguration;
 import com.recom.model.message.MessageContainer;
 import com.recom.model.message.MessageType;
 import com.recom.model.message.OneMessage;
-import com.recom.observer.Subject;
+import com.recom.observer.Subjective;
 import com.recom.persistence.message.MessagePersistenceLayer;
-import com.recom.property.RECOMAsyncProperties;
 import com.recom.service.provider.StaticObjectMapperProvider;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -16,7 +14,6 @@
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.springframework.core.task.AsyncTaskExecutor;
 
-import java.time.Duration;
 import java.util.List;
 
 import static org.mockito.Mockito.*;
@@ -29,7 +26,6 @@ class MessageLongPollObserverTest {
     @Mock
     private MessagePersistenceLayer messagePersistenceLayer;
     private MessageBusService messageBusService;
-    private AsyncConfiguration asyncConfiguration;
     private MessageLongPollObserver observerUnderTest;
 
     @BeforeEach
@@ -49,20 +45,13 @@ public void setUp() {
         final ObjectMapper objectMapper = new ObjectMapper();
         final StaticObjectMapperProvider staticObjectMapperProvider = new StaticObjectMapperProvider(objectMapper);
         staticObjectMapperProvider.postConstruct();
-
-        // Prepare AsyncConfiguration
-        final RECOMAsyncProperties properties = RECOMAsyncProperties.builder()
-                .corePoolSize(1)
-                .maxPoolSize(1)
-                .build();
-        asyncConfiguration = new AsyncConfiguration(properties);
     }
 
     @Test
-    public void test() throws InterruptedException {
+    public void sendMessage_withOneMessage_messageIsPersisted() throws InterruptedException {
         // Arrange
-        observerUnderTest.scheduleTestResponse("TestMap", Duration.ofMillis(5), new Subject<>(), asyncConfiguration);
-        observerUnderTest.observe(messageBusService.getSubject());
+        final Subjective<MessageContainer> messageBusSubjectSpy = spy(messageBusService.getSubject());
+        observerUnderTest.observe(messageBusSubjectSpy);
 
         final OneMessage testMessage = OneMessage.builder()
                 .messageType(MessageType.TEST)
@@ -78,10 +67,13 @@ public void test() throws InterruptedException {
                         .build()
         );
 
+        // run autocloseable implementation;
+        observerUnderTest.close();
+
         // Assert
-        // Sleep for a duration longer than the scheduled task to complete
-        Thread.sleep(200);
         verify(messagePersistenceLayer, times(1)).saveAll(anyList());
+        verify(messageBusSubjectSpy, times(1)).beObservedBy(observerUnderTest);
+        verify(messageBusSubjectSpy, times(1)).observationStoppedThrough(any()); // executed by autocloseable implementation
     }
 
 }