Rework Mqtt Implementation

api/src/main/java/MqttStarter.java

@@ -0,0 +1,45 @@
+import be.naaturel.boardmateapi.common.helpers.Logger;
+import be.naaturel.boardmateapi.services.MqttService;
+import jakarta.annotation.PreDestroy;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
+
+import java.util.Arrays;
+
+@Component
+public class MqttStarter {
+
+    private final MqttService service;
+
+    public MqttStarter(MqttService service){
+        this.service = service;
+    }
+
+    @EventListener(ApplicationReadyEvent.class)
+    public void start(){
+        try {
+            setCallback();
+            service.subscribe("/board-mate/+/telemetry");
+        } catch (Exception e){
+            System.err.println(Arrays.toString(e.getStackTrace()));
+        }
+    }
+
+    @PreDestroy
+    public void stop() {
+        service.disconnect();
+    }
+
+    private void setCallback(){
+        service.onConnectionLost((cause) -> {
+            Logger.displayError("Connection lost: " + cause.getMessage());
+        });
+
+        service.onMessageReceived((msg) -> {
+            Logger.displayInfo("Received message on topic " + msg.getTopic() + ": " + msg.getContent());
+        });
+
+        service.registerCallback();
+    }
+}

api/src/main/java/be/naaturel/boardmateapi/common/models/MqttMessageWrapper.java

@@ -0,0 +1,24 @@
+package be.naaturel.boardmateapi.common.models;
+
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+
+import java.nio.charset.StandardCharsets;
+
+public class MqttMessageWrapper {
+
+    private final String topic;
+    private final MqttMessage message;
+
+    public MqttMessageWrapper(String topic, MqttMessage message){
+        this.topic = topic;
+        this.message = message;
+    }
+
+    public String getContent() {
+        return new String(message.getPayload(), StandardCharsets.UTF_8);
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+}

api/src/main/java/be/naaturel/boardmateapi/configurations/configurations/MqttConfig.java

@@ -29,26 +29,7 @@ public class MqttConfig {
     @Bean("mqttSubscriber")
     public MqttClient mqttSubscriber() throws MqttException {
         String subscriberId = properties.getClientId() + "-sub";
-        MqttClient client = new MqttClient(properties.getBrokerUrl(), subscriberId);
-
-        client.setCallback(new MqttCallback() {
-            @Override
-            public void connectionLost(Throwable cause) {
-                Logger.displayError("Connection lost: " + cause.getMessage());
-            }
-
-            @Override
-            public void messageArrived(String topic, MqttMessage message) {
-                Logger.displayInfo("Received message on topic " + topic + ": " + message.toString());
-            }
-
-            @Override
-            public void deliveryComplete(IMqttDeliveryToken token) {
-                // Not needed for subscriber
-            }
-        });
-
-        return client;
+        return new MqttClient(properties.getBrokerUrl(), subscriberId);
     }
 
 }

api/src/main/java/be/naaturel/boardmateapi/configurations/properties/MqttProperies.java

@@ -6,9 +6,12 @@ import org.springframework.context.annotation.Configuration;
 @Configuration
 public class MqttProperies {
 
-    @Value("${mqtt.broker-url}")
+    @Value("${mqtt.plain.broker-url}")
     private String brokerUrl;
 
+    @Value("${mqtt.ssl.broker-url}")
+    private String brokerSecureUrl;
+
     @Value("${mqtt.client-id}")
     private String clientId;
 
@@ -18,12 +21,9 @@ public class MqttProperies {
     @Value("${mqtt.password}")
     private String password;
 
-    @Value("${mqtt.topic}")
-    private String topic;
-
     public String getBrokerUrl() { return brokerUrl; }
+    public String getBrokerSecureUrl() { return brokerSecureUrl; }
     public String getClientId() { return clientId; }
     public String getUsername() { return username; }
     public String getPassword() { return password; }
-    public String getTopic() { return topic; }
 }

api/src/main/java/be/naaturel/boardmateapi/controllers/BrokerController.java

@@ -21,7 +21,7 @@ public class BrokerController {
         this.service = service;
     }
 
-    @PostMapping("/publish/{topic}")
+    /*@PostMapping("/publish/{topic}")
     public ResponseEntity<ResponseBody<?>> publish(@PathVariable String topic, @RequestBody String message){
         ResponseBody<?> body = ResponseBody.createEmpty();
         try {
@@ -38,6 +38,6 @@ public class BrokerController {
                     .status(HttpStatus.INTERNAL_SERVER_ERROR)
                     .body(body);
         }
-    }
+    }*/
 }
 

api/src/main/java/be/naaturel/boardmateapi/services/MqttService.java

@@ -2,59 +2,110 @@ package be.naaturel.boardmateapi.services;
 
 import be.naaturel.boardmateapi.common.exceptions.ServiceException;
 import be.naaturel.boardmateapi.common.helpers.Logger;
+import be.naaturel.boardmateapi.common.models.MqttMessageWrapper;
 import org.eclipse.paho.client.mqttv3.*;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 
+import java.nio.charset.StandardCharsets;
+import java.util.function.Consumer;
+
 @Service
 public class MqttService {
 
     private final MqttClient publisher;
     private final MqttClient subscriber;
+    private Consumer<Throwable> onConnectionLost;
+    private Consumer<MqttMessageWrapper> onMessageReceived;
 
     public MqttService(
             @Qualifier("mqttPublisher") MqttClient publisher,
             @Qualifier("mqttSubscriber") MqttClient subscriber
-            ) {
+    ) {
         this.publisher = publisher;
         this.subscriber = subscriber;
+        this.onConnectionLost = null;
+        this.onMessageReceived = null;
     }
 
-    public void publish(String topic, String payload) throws ServiceException {
+    public void onConnectionLost(Consumer<Throwable> consumer){
+        this.onConnectionLost = consumer;
+    }
+
+    public void onMessageReceived(Consumer<MqttMessageWrapper> consumer){
+        this.onMessageReceived = consumer;
+    }
+
+    public void disconnect() {
         try {
-            MqttMessage message = new MqttMessage(payload.getBytes());
-            message.setQos(1);
-            if(!publisher.isConnected()){
-                publisher.connect();
-            }
-            publisher.publish(topic, message);
-            Logger.displayInfo("Published message: " + payload);
-        } catch (MqttException e) {
-            throw new ServiceException("Unable to publish message", e);
-        } finally {
-            try{
-                if (publisher.isConnected()) publisher.disconnect();
-            } catch (MqttException e){
-                Logger.displayError("Failed to disconnect MQTT client: " + e.getMessage());
-            }
+            disconnect(publisher);
+            disconnect(subscriber);
+        } catch (ServiceException e) {
+            Logger.displayError("Failed to disconnect MQTT client" + e.getMessage());
         }
     }
 
-    public void subscribe(String topic) throws ServiceException {
+    public void publish(String topic, String payload) {
+        try {
+            connect(publisher, null);
+            MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
+            message.setQos(1);
+            publisher.publish(topic, message);
+        } catch (MqttException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public void subscribe(String topic) {
         try {
             MqttConnectOptions options = new MqttConnectOptions();
             options.setAutomaticReconnect(true);
-            options.setCleanSession(true);
+            options.setCleanSession(false);
+
+            connect(subscriber, options);
 
-            if(!subscriber.isConnected()){
-                subscriber.connect(options);
-            }
             subscriber.subscribe(topic, 1);
-            Logger.displayInfo("Subscribed to topic: " + topic);
-
         } catch (MqttException e) {
-            throw new ServiceException("Unable to subscribe", e);
+            throw new RuntimeException(e);
         }
     }
 
+    public void registerCallback(){
+        this.subscriber.setCallback(new MqttCallback() {
+            @Override
+            public void connectionLost(Throwable cause) {
+                onConnectionLost.accept(cause);
+            }
+
+            @Override
+            public void messageArrived(String topic, MqttMessage message) {
+                MqttMessageWrapper msg = new MqttMessageWrapper(topic, message);
+                onMessageReceived.accept(msg);
+            }
+
+            @Override
+            public void deliveryComplete(IMqttDeliveryToken token) {
+                // Not needed for subscriber
+            }
+        });
+    }
+
+    private void disconnect(MqttClient client) throws ServiceException {
+        if(client.isConnected()){
+            try {
+                client.disconnect();
+            } catch (MqttException e) {
+                throw new ServiceException("Failed to disconnect the broker", e);
+            }
+        }
+    }
+
+    private void connect(MqttClient client, MqttConnectOptions options) throws MqttException {
+        if (client.isConnected()) return;
+        if (options == null) {
+            client.connect();
+        } else {
+            client.connect(options);
+        }
+    }
 }

api/src/main/resources/application.properties

@@ -19,13 +19,13 @@ server.ssl.key-store-type=PKCS12
 server.ssl.key-alias=board-mate-api
 
 #=============MQTT=============
-mqtt.broker-url=tcp://test.mosquitto.org:1883
-mqtt.client-id=board-mate-client
+mqtt.plain.broker-url=${BROKER_URL}
+mqtt.ssl.broker-url=${BROKER_SECURE_URL}
+mqtt.client-id=board-mate-api
 
-mqtt.topic=board-mate-test/topic
+mqtt.username=${BROKER_USERNAME}
+mqtt.password=${BROKER_PASSWORD}
 
-mqtt.username=yourUsername
-mqtt.password=yourPassword
 #=============METRICS=============
 management.endpoint.health.show-details=always
 management.endpoints.web.exposure.include=*