Missing classes

Author: grzegorzswistak

src/main/java/com/orange/lo/sample/mqtt2eventhub/ApplicationConfig.java

@@ -0,0 +1,25 @@
+/**
+ * Copyright (c) Orange. All Rights Reserved.
+ * <p>
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+package com.orange.lo.sample.mqtt2eventhub;
+
+import com.orange.lo.sample.mqtt2eventhub.liveobjects.LoMessage;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+@Configuration
+public class ApplicationConfig {
+
+    @Bean
+    public Queue<LoMessage> messageQueue() {
+        return new ConcurrentLinkedQueue<>();
+    }
+
+}

src/main/java/com/orange/lo/sample/mqtt2eventhub/liveobjects/LoMessage.java

@@ -0,0 +1,5 @@
+package com.orange.lo.sample.mqtt2eventhub.liveobjects;
+
+public record LoMessage(int messageId, String message) {
+
+}

src/main/java/com/orange/lo/sample/mqtt2eventhub/liveobjects/LoMqttReconnectHandler.java

@@ -0,0 +1,40 @@
+/**
+ * Copyright (c) Orange. All Rights Reserved.
+ * <p>
+ * This source code is licensed under the MIT license found in the
+ * LICENSE file in the root directory of this source tree.
+ */
+
+package com.orange.lo.sample.mqtt2eventhub.liveobjects;
+
+import com.orange.lo.sample.mqtt2eventhub.utils.ConnectorHealthActuatorEndpoint;
+import com.orange.lo.sdk.mqtt.DataManagementReconnectCallback;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.lang.invoke.MethodHandles;
+
+@Component
+public class LoMqttReconnectHandler implements DataManagementReconnectCallback {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+    private final ConnectorHealthActuatorEndpoint connectorHealthActuatorEndpoint;
+
+    public LoMqttReconnectHandler(ConnectorHealthActuatorEndpoint connectorHealthActuatorEndpoint) {
+        this.connectorHealthActuatorEndpoint = connectorHealthActuatorEndpoint;
+    }
+
+    @Override
+    public void connectComplete(boolean b, String s) {
+        LOG.info("Hahahaah moj handler #1 {}", connectorHealthActuatorEndpoint.isLoConnectionStatus());
+        connectorHealthActuatorEndpoint.setLoConnectionStatus(true);
+        LOG.info("Hahahaah moj handler #2 {}", connectorHealthActuatorEndpoint.isLoConnectionStatus());
+    }
+
+    @Override
+    public void connectionLost(Throwable throwable) {
+        connectorHealthActuatorEndpoint.setLoConnectionStatus(false);
+    }
+}

src/test/java/com/orange/lo/sample/mqtt2eventhub/ApplicationConfigTest.java

@@ -0,0 +1,27 @@
+package com.orange.lo.sample.mqtt2eventhub;
+
+import com.orange.lo.sample.mqtt2eventhub.liveobjects.LoMessage;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class ApplicationConfigTest {
+
+    private ApplicationConfig applicationConfig;
+
+    @BeforeEach
+    void setUp() {
+        applicationConfig = new ApplicationConfig();
+    }
+
+    @Test
+    void messageQueueShouldBeThreadSafe() {
+        Queue<LoMessage> loMessages = applicationConfig.messageQueue();
+
+        assertInstanceOf(ConcurrentLinkedQueue.class, loMessages);
+    }
+}

src/test/java/com/orange/lo/sample/mqtt2eventhub/evthub/EventHubSenderTest.java

@@ -0,0 +1,144 @@
+package com.orange.lo.sample.mqtt2eventhub.evthub;
+
+import com.orange.lo.sample.mqtt2eventhub.liveobjects.LoMessage;
+import com.orange.lo.sample.mqtt2eventhub.liveobjects.LoService;
+import com.orange.lo.sample.mqtt2eventhub.utils.ConnectorHealthActuatorEndpoint;
+import com.orange.lo.sample.mqtt2eventhub.utils.Counters;
+import io.micrometer.core.instrument.Counter;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentMatcher;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class EventHubSenderTest {
+    @Mock
+    private EventHubClientFacade eventHubClientFacade;
+    @Mock
+    private Counters counters;
+    @Mock
+    private Counter counter;
+    @Mock
+    private EventHubProperties eventHubProperties;
+    @Mock
+    private ExecutorService executorService;
+    @Mock
+    private ConnectorHealthActuatorEndpoint connectorHealthActuatorEndpoint;
+    @Mock
+    private LoService loService;
+
+    EventHubSender eventHubSender;
+
+    @BeforeEach
+    void setUp() {
+        prepareService(new LinkedList<>());
+    }
+
+    private void prepareService(LinkedList<LoMessage> messageQueue) {
+        eventHubSender = new EventHubSender(eventHubClientFacade, counters, eventHubProperties, executorService,
+                connectorHealthActuatorEndpoint, loService, messageQueue);
+    }
+
+    @Test
+    void shouldNotSendMessagesIfQueueIsEmpty() throws EventHubClientFacadeException {
+        // when
+        eventHubSender.send();
+
+        // then
+        verify(eventHubClientFacade, never()).sendSync(ArgumentMatchers.<String>anyList());
+    }
+
+    @Test
+    void shouldSendMessagesInOneBatchIfQueueNotExceedMessageBatchSizeProperty() throws EventHubClientFacadeException {
+        // given
+        int batchSize = 5;
+
+        when(eventHubProperties.getMessageBatchSize()).thenReturn(batchSize);
+        when(eventHubProperties.getMaxSendAttempts()).thenReturn(10);
+        when(eventHubProperties.getThrottlingDelay()).thenReturn(Duration.ofMillis(1));
+        when(counters.getMesasageSentAttemptCounter()).thenReturn(counter);
+
+        LinkedList<LoMessage> messageQueue = getExampleMessageQueue(batchSize);
+
+        prepareService(messageQueue);
+
+        List<String> expectedMessages = messageQueue.stream().map(LoMessage::message).toList();
+
+        // when
+        eventHubSender.send();
+
+        // then
+        verify(eventHubClientFacade, times(1)).sendSync(expectedMessages);
+    }
+
+    @Test
+    void shouldSplitMessagesIntoPacketsIfQueueExceedMessageBatchSizeProperty() throws EventHubClientFacadeException {
+        // given
+        int batchSize = 5;
+        int totalLength = batchSize + 1;
+
+        when(eventHubProperties.getMessageBatchSize()).thenReturn(batchSize);
+        when(eventHubProperties.getMaxSendAttempts()).thenReturn(10);
+        when(eventHubProperties.getThrottlingDelay()).thenReturn(Duration.ofMillis(1));
+        when(counters.getMesasageSentAttemptCounter()).thenReturn(counter);
+
+        LinkedList<LoMessage> messageQueue = getExampleMessageQueue(totalLength);
+
+        prepareService(messageQueue);
+
+        List<String> expectedMessages1 = (new LinkedList<>(messageQueue)).subList(0, batchSize)
+                .stream()
+                .map(LoMessage::message)
+                .toList();
+        List<String> expectedMessages2 = (new LinkedList<>(messageQueue)).subList(batchSize, totalLength)
+                .stream()
+                .map(LoMessage::message)
+                .toList();
+
+        // when
+        eventHubSender.send();
+
+        // then
+        verify(eventHubClientFacade, times(1)).sendSync(expectedMessages1);
+        verify(eventHubClientFacade, times(1)).sendSync(expectedMessages2);
+    }
+
+    private LinkedList<LoMessage> getExampleMessageQueue(int batchSize) {
+        return IntStream.range(1, batchSize + 1)
+                .mapToObj(i -> new LoMessage(i, String.format("Message %d", i)))
+                .collect(Collectors.toCollection(LinkedList::new));
+    }
+
+    class IsListWithGivenSize implements ArgumentMatcher<List> {
+        int size;
+
+        public IsListWithGivenSize(int size) {
+            this.size = size;
+        }
+
+        @Override
+        public boolean matches(List list) {
+            return list.size() == size;
+        }
+    }
+}

src/test/java/com/orange/lo/sample/mqtt2eventhub/liveobjects/LoConfigTest.java

@@ -0,0 +1,55 @@
+package com.orange.lo.sample.mqtt2eventhub.liveobjects;
+
+import com.orange.lo.sdk.LOApiClient;
+import com.orange.lo.sdk.fifomqtt.DataManagementFifoCallback;
+import com.orange.lo.sdk.mqtt.DataManagementReconnectCallback;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.ConfigDataApplicationContextInitializer;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+@ExtendWith({SpringExtension.class})
+@EnableConfigurationProperties(LoProperties.class)
+@ContextConfiguration(initializers = ConfigDataApplicationContextInitializer.class)
+@ActiveProfiles("unittest")
+class LoConfigTest {
+
+    private DataManagementFifoCallback dataManagementFifoCallback = (i, s) -> {
+
+    };
+    private final DataManagementReconnectCallback dataManagementReconnectCallback = new DataManagementReconnectCallback() {
+
+        @Override
+        public void connectComplete(boolean b, String s) {
+
+        }
+
+        @Override
+        public void connectionLost(Throwable throwable) {
+
+        }
+    };
+
+    @Autowired
+    private LoProperties loProperties;
+    private LoConfig loConfig;
+
+    @BeforeEach
+    void setUp() {
+        loConfig = new LoConfig(loProperties, dataManagementFifoCallback, dataManagementReconnectCallback);
+    }
+
+    @Test
+    void shouldInitLoApiClient() {
+        LOApiClient loApiClient = loConfig.loApiClient();
+
+        assertNotNull(loApiClient);
+    }
+}

src/test/java/com/orange/lo/sample/mqtt2eventhub/liveobjects/LoMqttReconnectHandlerTest.java

@@ -0,0 +1,38 @@
+package com.orange.lo.sample.mqtt2eventhub.liveobjects;
+
+import com.orange.lo.sample.mqtt2eventhub.utils.ConnectorHealthActuatorEndpoint;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+@ExtendWith(MockitoExtension.class)
+class LoMqttReconnectHandlerTest {
+
+    @Mock
+    private ConnectorHealthActuatorEndpoint connectorHealthActuatorEndpoint;
+    private LoMqttReconnectHandler loMqttReconnectHandler;
+
+    @BeforeEach
+    void setUp() {
+        this.loMqttReconnectHandler = new LoMqttReconnectHandler(connectorHealthActuatorEndpoint);
+    }
+
+    @Test
+    void shouldChangeLoConnectionStausWhenConnectComplete() {
+        loMqttReconnectHandler.connectComplete(false, "");
+
+        verify(connectorHealthActuatorEndpoint, times(1)).setLoConnectionStatus(true);
+    }
+
+    @Test
+    void shouldChangeLoConnectionStausWhenConnectionLost() {
+        loMqttReconnectHandler.connectionLost(new Exception());
+
+        verify(connectorHealthActuatorEndpoint, times(1)).setLoConnectionStatus(false);
+    }
+}

src/test/java/com/orange/lo/sample/mqtt2eventhub/liveobjects/LoPropertiesTest.java

@@ -0,0 +1,37 @@
+package com.orange.lo.sample.mqtt2eventhub.liveobjects;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.test.context.ConfigDataApplicationContextInitializer;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+@ExtendWith({SpringExtension.class})
+@EnableConfigurationProperties(LoProperties.class)
+@ContextConfiguration(initializers = ConfigDataApplicationContextInitializer.class)
+@ActiveProfiles("unittest")
+class LoPropertiesTest {
+
+    @Autowired
+    private LoProperties loProperties;
+
+    @Test
+    void shouldBindLoPropertiesFromYamlConfiguration() {
+        Assertions.assertAll(
+                () -> assertEquals("invalid.url.liveobjects.orange-business.com", this.loProperties.getHostname()),
+                () -> assertEquals("test", this.loProperties.getApiKey()),
+                () -> assertEquals("dev", this.loProperties.getTopic()),
+                () -> assertEquals(30, this.loProperties.getKeepAliveIntervalSeconds()),
+                () -> assertEquals(true, this.loProperties.getAutomaticReconnect()),
+                () -> assertEquals(1, this.loProperties.getMessageQos()),
+                () -> assertEquals("./temp/", this.loProperties.getMqttPersistenceDir()),
+                () -> assertEquals(30000, this.loProperties.getConnectionTimeout())
+        );
+    }
+}