Added the possibility to specify the MQTT ClientId both for source an…

…d destination brokers
wldt · May 27, 2021 · 977a106 · 977a106
1 parent c2cda23
commit 977a106
Show file tree

Hide file tree

Showing 5 changed files with 56 additions and 4 deletions.
diff --git a/conf/mqtt.yaml b/conf/mqtt.yaml
@@ -11,6 +11,7 @@ brokerClientUsername: null
 brokerClientPassword: null
 brokerSecureCommunicationRequired: false
 brokerServerCertPath: null
+brokerClientId: null #If null the clientId will be randomly generated by the library (Default Option)
 
 #Destination Broker Parameters
 destinationBrokerAddress: 127.0.0.1
@@ -20,6 +21,7 @@ destinationBrokerClientUsername: null
 destinationBrokerClientPassword: null
 destinationBrokerSecureCommunicationRequired: false
 destinationBrokerServerCertPath: null
+destinationBrokerClientId: null #If null the clientId will be randomly generated by the library (Default Option)
 
 #e.g., digitaltwin (without the final /)
 destinationBrokerBaseTopic: null

diff --git a/pom.xml b/pom.xml
@@ -6,7 +6,7 @@
 
     <groupId>it.unimore.dipi.iot</groupId>
     <artifactId>wldt-core</artifactId>
-    <version>0.1.3.5</version>
+    <version>0.1.3.6</version>
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

diff --git a/src/main/java/it/unimore/dipi/iot/wldt/worker/mqtt/Mqtt2MqttConfiguration.java b/src/main/java/it/unimore/dipi/iot/wldt/worker/mqtt/Mqtt2MqttConfiguration.java
@@ -30,6 +30,8 @@ public class Mqtt2MqttConfiguration implements WldtWorkerConfiguration {
 
     private String destinationBrokerTlsContext = "TLSv1.2";
 
+    private String destinationBrokerClientId = null;
+
     private String deviceId = null;
 
     private String brokerAddress = "127.0.0.1";
@@ -48,6 +50,8 @@ public class Mqtt2MqttConfiguration implements WldtWorkerConfiguration {
 
     private String brokerTlsContext = "TLSv1.2";
 
+    private String brokerClientId = null;
+
     private List<MqttTopicDescriptor> topicList;
 
     public Mqtt2MqttConfiguration() {
@@ -213,6 +217,22 @@ public void setBrokerTlsContext(String brokerTlsContext) {
         this.brokerTlsContext = brokerTlsContext;
     }
 
+    public String getBrokerClientId() {
+        return brokerClientId;
+    }
+
+    public void setBrokerClientId(String brokerClientId) {
+        this.brokerClientId = brokerClientId;
+    }
+
+    public String getDestinationBrokerClientId() {
+        return destinationBrokerClientId;
+    }
+
+    public void setDestinationBrokerClientId(String destinationBrokerClientId) {
+        this.destinationBrokerClientId = destinationBrokerClientId;
+    }
+
     @Override
     public String toString() {
         final StringBuffer sb = new StringBuffer("Mqtt2MqttConfiguration{");
@@ -226,6 +246,7 @@ public String toString() {
         sb.append(", destinationBrokerSecureCommunicationRequired=").append(destinationBrokerSecureCommunicationRequired);
         sb.append(", destinationBrokerServerCertPath='").append(destinationBrokerServerCertPath).append('\'');
         sb.append(", destinationBrokerTlsContext='").append(destinationBrokerTlsContext).append('\'');
+        sb.append(", destinationBrokerClientId='").append(destinationBrokerClientId).append('\'');
         sb.append(", deviceId='").append(deviceId).append('\'');
         sb.append(", brokerAddress='").append(brokerAddress).append('\'');
         sb.append(", brokerPort=").append(brokerPort);
@@ -235,6 +256,7 @@ public String toString() {
         sb.append(", brokerSecureCommunicationRequired=").append(brokerSecureCommunicationRequired);
         sb.append(", brokerServerCertPath='").append(brokerServerCertPath).append('\'');
         sb.append(", brokerTlsContext='").append(brokerTlsContext).append('\'');
+        sb.append(", brokerClientId='").append(brokerClientId).append('\'');
         sb.append(", topicList=").append(topicList);
         sb.append('}');
         return sb.toString();

diff --git a/src/main/java/it/unimore/dipi/iot/wldt/worker/mqtt/Mqtt2MqttManager.java b/src/main/java/it/unimore/dipi/iot/wldt/worker/mqtt/Mqtt2MqttManager.java
@@ -232,7 +232,21 @@ private void initPhysicalDeviceMqttBrokerClient() throws WldtMqttModuleException
 
         try {
 
-            physicalDeviceMqttBrokerClient = new MqttClient(getPhysicalThingMqttBrokerUrl(),String.format("%s%s", this.wldtId, "physical"), new MemoryPersistence());
+            String targetUrl = getPhysicalThingMqttBrokerUrl();
+
+            //Default client id
+            String mqttClientId = String.format("%s%s", this.wldtId, "physical");
+
+            //Check if a clientId is specified in the configuration and use it
+            if(this.mqtt2MqttConfiguration != null &&
+                    this.mqtt2MqttConfiguration.getBrokerClientId() != null &&
+                    this.mqtt2MqttConfiguration.getBrokerClientId().length() > 0
+            )
+                mqttClientId = this.mqtt2MqttConfiguration.getBrokerClientId();
+
+            logger.info("{} Client ({}) - Connecting to {} ...", TAG, mqttClientId, targetUrl);
+
+            physicalDeviceMqttBrokerClient = new MqttClient(targetUrl, mqttClientId, new MemoryPersistence());
 
             MqttConnectOptions options = new MqttConnectOptions();
             options.setAutomaticReconnect(true);
@@ -283,9 +297,19 @@ private void initDigitalTwinMqttBrokerClient() throws WldtMqttModuleException {
 
             String targetUrl = getDestinationMqttBrokerUrl();
 
-            logger.info("{} Connecting to {} ...", TAG, targetUrl);
+            //Default client id
+            String mqttClientId = String.format("%s%s", this.wldtId, "digital");
+
+            //Check if a clientId is specified in the configuration and use it
+            if(this.mqtt2MqttConfiguration != null &&
+                    this.mqtt2MqttConfiguration.getDestinationBrokerClientId() != null &&
+                    this.mqtt2MqttConfiguration.getDestinationBrokerClientId().length() > 0
+            )
+                mqttClientId = this.mqtt2MqttConfiguration.getDestinationBrokerClientId();
+
+            logger.info("{} Client ({}) - Connecting to {} ...", TAG, mqttClientId, targetUrl);
 
-            digitalTwinMqttBrokerClient = new MqttClient(targetUrl,String.format("%s%s", this.wldtId, "digital"), new MemoryPersistence());
+            digitalTwinMqttBrokerClient = new MqttClient(targetUrl, mqttClientId, new MemoryPersistence());
 
             MqttConnectOptions options = new MqttConnectOptions();
             options.setAutomaticReconnect(true);

diff --git a/src/test/java/it/unimore/wldt/test/mqtt/WldtExampleMqttProcess.java b/src/test/java/it/unimore/wldt/test/mqtt/WldtExampleMqttProcess.java
@@ -138,6 +138,10 @@ private static Mqtt2MqttConfiguration getMqttComplexProtocolConfiguration(){
         mqtt2MqttConfiguration.setDestinationBrokerPort(DESTINATION_BROKER_PORT);
         mqtt2MqttConfiguration.setDeviceId(DEVICE_ID);
 
+        //If Required Specify the ClientId
+        //mqtt2MqttConfiguration.setBrokerClientId("physicalBrokerTestClientId");
+        //mqtt2MqttConfiguration.setDestinationBrokerClientId("digitalBrokerTestClientId");
+
         //Specify Topic List Configuration
         mqtt2MqttConfiguration.setTopicList(
                 Arrays.asList(