Breaking of circular references, tests

Author: grzegorzswistak

.github/workflows/manual_release.yml

@@ -7,10 +7,10 @@ jobs:
   build:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
 
       - name: Create tag by maven-release plugin
-        uses: qcastel/[email protected].2
+        uses: qcastel/[email protected].41
         with:
           release-branch-name: "master"
           git-release-bot-name: "auto"

.github/workflows/on_push.yml

@@ -8,10 +8,11 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v2
-    - name: Set up JDK 1.8
-      uses: actions/setup-java@v1
+    - uses: actions/checkout@v4
+    - name: Set up JDK
+      uses: actions/setup-java@v2
       with:
-        java-version: 1.8
+        distribution: 'zulu'
+        java-version: '21'
     - name: Build with Maven
       run: mvn clean test --file pom.xml

pom.xml

@@ -14,7 +14,7 @@
 	<description>Demo project for Spring Boot</description>
 
 	<properties>
-		<java.version>1.8</java.version>
+		<java.version>21</java.version>
 	</properties>
 
 	<repositories>
@@ -28,7 +28,7 @@
 		<dependency>
 			<groupId>com.github.DatavenueLiveObjects</groupId>
 			<artifactId>FIFO_SDK_LiveObjects</artifactId>
-			<version>v1.16</version>
+			<version>v2.0.2</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>

src/main/java/com/orange/lo/sample/mqtt2eventhub/evthub/EventHubClientFacade.java

@@ -22,6 +22,7 @@
 import java.nio.charset.Charset;
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
+import java.util.List;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 
@@ -59,16 +60,28 @@ private void init() {
         }
     }
 
+    public void sendSync(List<String> messages) throws EventHubClientFacadeException {
+        List<EventData> list = messages.stream().map(msg -> {
+            byte[] bytes = msg.getBytes(Charset.defaultCharset());
+            return EventData.create(bytes);
+        }).toList();
+        sendSync(list);
+    }
+
     public void sendSync(String msg) throws EventHubClientFacadeException {
         byte[] payloadBytes = msg.getBytes(Charset.defaultCharset());
         sendSync(payloadBytes);
     }
 
-    public void sendSync(byte[] msg) throws EventHubClientFacadeException {
-        EventData sendEvent = EventData.create(msg);
+    public void sendSync(byte[] payloadBytes) throws EventHubClientFacadeException {
+        EventData sendEvent = EventData.create(payloadBytes);
+        List<EventData> list = List.of(sendEvent);
+        sendSync(list);
+    }
 
+    public void sendSync(Iterable<EventData> eventDatas) throws EventHubClientFacadeException {
         try {
-            eventHubClient.sendSync(sendEvent);
+            eventHubClient.sendSync(eventDatas);
         } catch (NullPointerException e) {
             throw new EventHubClientFacadeException(EVENT_HUB_CLIENT_NOT_INITIALIZED_MESSAGE);
         } catch (EventHubException e) {

src/main/java/com/orange/lo/sample/mqtt2eventhub/evthub/EventHubProperties.java

@@ -27,6 +27,7 @@ public class EventHubProperties {
     @DurationUnit(ChronoUnit.MILLIS)
     private Duration throttlingDelay = Duration.ofMillis(1);
     private int maxSendAttempts = 3;
+    private Integer messageBatchSize;
 
     public String getNameSpace() {
         return nameSpace;
@@ -99,4 +100,12 @@ public int getMaxSendAttempts() {
     public void setMaxSendAttempts(int maxSendAttempts) {
         this.maxSendAttempts = maxSendAttempts;
     }
+
+    public Integer getMessageBatchSize() {
+        return messageBatchSize;
+    }
+
+    public void setMessageBatchSize(Integer messageBatchSize) {
+        this.messageBatchSize = messageBatchSize;
+    }
 }

src/main/java/com/orange/lo/sample/mqtt2eventhub/evthub/EventHubSender.java

@@ -7,49 +7,53 @@
 
 package com.orange.lo.sample.mqtt2eventhub.evthub;
 
-import com.orange.lo.sample.mqtt2eventhub.liveobjects.LoProperties;
+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 com.orange.lo.sdk.LOApiClient;
 import net.jodah.failsafe.Failsafe;
 import net.jodah.failsafe.RetryPolicy;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
 import java.lang.invoke.MethodHandles;
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
 import java.util.concurrent.ExecutorService;
 
 @Component
 public class EventHubSender {
 
     private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+    private static final int DEFAULT_BATCH_SIZE = 10;
 
     private final EventHubClientFacade eventHubClientFacade;
     private final Counters counters;
     private final EventHubProperties eventHubProperties;
-    private final LoProperties loProperties;
     private final ExecutorService executorService;
-    private final LOApiClient loApiClient;
     private final ConnectorHealthActuatorEndpoint connectorHealthActuatorEndpoint;
+    private final LoService loService;
+    private final Queue<LoMessage> messageQueue;
 
     EventHubSender(EventHubClientFacade eventHubClientFacade, Counters counters, EventHubProperties eventHubProperties,
-                   LoProperties loProperties, ExecutorService executorService, LOApiClient loApiClient,
-                   ConnectorHealthActuatorEndpoint connectorHealthActuatorEndpoint) {
+                   ExecutorService executorService, ConnectorHealthActuatorEndpoint connectorHealthActuatorEndpoint,
+                   LoService loService, Queue<LoMessage> messageQueue) {
         this.eventHubClientFacade = eventHubClientFacade;
         this.counters = counters;
         this.eventHubProperties = eventHubProperties;
-        this.loProperties = loProperties;
         this.executorService = executorService;
-        this.loApiClient = loApiClient;
         this.connectorHealthActuatorEndpoint = connectorHealthActuatorEndpoint;
+        this.loService = loService;
+        this.messageQueue = messageQueue;
     }
 
-    public void send(int loMessageId, String msg) {
-
+    public void send(List<LoMessage> messages) {
         Failsafe.with(
                 new RetryPolicy<Void>()
                         .withMaxAttempts(eventHubProperties.getMaxSendAttempts())
@@ -61,17 +65,18 @@ public void send(int loMessageId, String msg) {
                         .onSuccess(r -> {
                             LOG.debug("Message was sent to Event Hub");
                             counters.getMesasageSentCounter().increment();
-                            loApiClient.getDataManagementFifo().sendAck(loMessageId, loProperties.getMessageQos());
+                            messages.forEach(m -> loService.sendAck(m.messageId()));
                         })
                         .onFailure(r -> {
-                            LOG.error("Cannot send message with id = {} to Event Hub because of {}", loMessageId, r.getFailure());
+                            LOG.error("Cannot send messages to Event Hub because of {}", r.getFailure());
                             counters.getMesasageSentFailedCounter().increment();
-                            loApiClient.getDataManagementFifo().sendAck(loMessageId, loProperties.getMessageQos());
+                            messages.forEach(m -> loService.sendAck(m.messageId()));
                         })
         ).with(executorService).run(execution -> {
             counters.getMesasageSentAttemptCounter().increment();
             try {
-                eventHubClientFacade.sendSync(msg);
+                List<String> messageContentList = messages.stream().map(LoMessage::message).toList();
+                eventHubClientFacade.sendSync(messageContentList);
                 connectorHealthActuatorEndpoint.setCloudConnectionStatus(true);
             } catch (EventHubClientFacadeException e) {
                 LOG.error("Problem with connection. Check Event Hub credentials. " + e.getMessage(), e);
@@ -90,4 +95,27 @@ private void checkConnection() {
             connectorHealthActuatorEndpoint.setCloudConnectionStatus(false);
         }
     }
+
+    @Scheduled(fixedRateString = "${azure.evt-hub.synchronization-interval}")
+    public void send() {
+        if (!messageQueue.isEmpty()) {
+            LOG.info("Start sending messages...");
+
+            int batchSize = eventHubProperties.getMessageBatchSize() != null
+                    ? eventHubProperties.getMessageBatchSize() : DEFAULT_BATCH_SIZE;
+
+            List<LoMessage> messageBatch = new ArrayList<>(batchSize);
+            while (!messageQueue.isEmpty()) {
+                messageBatch.add(messageQueue.poll());
+                if (messageBatch.size() == batchSize) {
+                    send(new ArrayList<>(messageBatch));
+                    messageBatch.clear();
+                }
+            }
+            if (!messageBatch.isEmpty()) {
+                send(new ArrayList<>(messageBatch));
+            }
+        }
+    }
+
 }

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

@@ -9,6 +9,8 @@
 
 import com.orange.lo.sdk.LOApiClient;
 import com.orange.lo.sdk.LOApiClientParameters;
+import com.orange.lo.sdk.fifomqtt.DataManagementFifoCallback;
+import com.orange.lo.sdk.mqtt.DataManagementReconnectCallback;
 import org.apache.maven.model.Model;
 import org.apache.maven.model.io.xpp3.MavenXpp3Reader;
 import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
@@ -25,12 +27,14 @@
 public class LoConfig {
 
 	private final LoProperties loProperties;
-	private final LoMqttHandler loMqttHandler;
+	private final DataManagementFifoCallback loMqttHandler;
+    private final DataManagementReconnectCallback reconnectHandler;
 
-	public LoConfig(LoProperties loProperties, LoMqttHandler loMqttHandler) {
+    public LoConfig(LoProperties loProperties, DataManagementFifoCallback loMqttHandler, DataManagementReconnectCallback reconnectHandler) {
 		this.loProperties = loProperties;
 		this.loMqttHandler = loMqttHandler;
-	}
+        this.reconnectHandler = reconnectHandler;
+    }
 
 	@Bean
 	public LOApiClient loApiClient() {
@@ -51,6 +55,7 @@ private LOApiClientParameters loApiClientParameters() {
                 .dataManagementMqttCallback(loMqttHandler)
                 .connectorType(loProperties.getConnectorType())
                 .connectorVersion(getConnectorVersion())
+				.dataManagementReconnectCallback(reconnectHandler)
                 .build();
     }
 

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

@@ -7,43 +7,35 @@
 
 package com.orange.lo.sample.mqtt2eventhub.liveobjects;
 
-import com.orange.lo.sample.mqtt2eventhub.evthub.EventHubSender;
 import com.orange.lo.sample.mqtt2eventhub.utils.Counters;
-import com.orange.lo.sdk.LOApiClient;
 import com.orange.lo.sdk.fifomqtt.DataManagementFifoCallback;
+import io.micrometer.core.instrument.Counter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
 
 import java.lang.invoke.MethodHandles;
+import java.util.Queue;
 
 @Component
 public class LoMqttHandler implements DataManagementFifoCallback {
 
     private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
-    private final EventHubSender eventHubSender;
-    private final Counters counters;
-    private final LOApiClient loApiClient;
-    private final LoProperties loProperties;
+    private final Counter mesasageReadCounter;
+    private final Queue<LoMessage> messageQueue;
 
-    public LoMqttHandler(@Lazy EventHubSender eventHubSender, @Lazy LOApiClient loApiClient, Counters counters, LoProperties loProperties) {
-        this.eventHubSender = eventHubSender;
-        this.counters = counters;
-        this.loApiClient = loApiClient;
-        this.loProperties = loProperties;
+    public LoMqttHandler(Counters counterProvider, Queue<LoMessage> messageQueue) {
+        LOG.info("LoMqttHandler init...");
+
+        this.mesasageReadCounter = counterProvider.getMesasageReadCounter();
+        this.messageQueue = messageQueue;
     }
 
     @Override
-    public void onMessage(int loMessageId, String message) {
-        counters.getMesasageReadCounter().increment();
-        try {
-            eventHubSender.send(loMessageId, message);
-        } catch (Exception e) {
-            LOG.error("Cannot send message with id = {} to Event Hub because of {}", loMessageId, e);
-            counters.getMesasageSentFailedCounter().increment();
-            loApiClient.getDataManagementFifo().sendAck(loMessageId, loProperties.getMessageQos());
-        }
+    public void onMessage(int messageId, String message) {
+        LOG.debug("Received message with id = {}", messageId);
+        mesasageReadCounter.increment();
+        messageQueue.add(new LoMessage(messageId, message));
     }
 }

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

@@ -15,7 +15,7 @@
 public class LoProperties {
 
 	private static final String CONNECTOR_TYPE = "LO_AZURE_EVENTHUB_ADAPTER";
-	
+
     private final String hostname;
     private final String apiKey;
     private final String topic;
@@ -47,7 +47,7 @@ public LoProperties(
     public String getConnectorType() {
         return CONNECTOR_TYPE;
     }
-    
+
     public String getHostname() {
         return hostname;
     }

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

@@ -25,11 +25,14 @@ public class LoService {
     private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
     private final DataManagementFifo dataManagementFifo;
+    private final LoProperties loProperties;
 
     private final ConnectorHealthActuatorEndpoint connectorHealthActuatorEndpoint;
 
-    public LoService(LOApiClient loApiClient, ConnectorHealthActuatorEndpoint connectorHealthActuatorEndpoint) {
+    public LoService(LOApiClient loApiClient, ConnectorHealthActuatorEndpoint connectorHealthActuatorEndpoint,
+                     LoProperties loProperties) {
         this.dataManagementFifo = loApiClient.getDataManagementFifo();
+        this.loProperties = loProperties;
         this.connectorHealthActuatorEndpoint = connectorHealthActuatorEndpoint;
     }
 
@@ -50,4 +53,9 @@ public void start() {
     public void stop() {
         dataManagementFifo.disconnect();
     }
+
+    public void sendAck(int loMessageId) {
+        dataManagementFifo.sendAck(loMessageId, loProperties.getMessageQos());
+    }
+
 }