[BAEL-9541] Introduction to HiveMQ MQTT Client

libraries-server/pom.xml

@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project xmlns="http://maven.apache.org/POM/4.0.0"
-    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <artifactId>libraries-server</artifactId>
     <version>0.0.1-SNAPSHOT</version>
@@ -55,6 +55,12 @@
             <artifactId>netty-all</artifactId>
             <version>${netty.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>com.hivemq</groupId>
+            <artifactId>hivemq-mqtt-client</artifactId>
+            <version>${hivemq.mqtt.client.version}</version>
+        </dependency>
     </dependencies>
 
     <properties>
@@ -63,6 +69,7 @@
         <netty.version>4.1.20.Final</netty.version>
         <eclipse.paho.client.mqttv3.version>1.2.0</eclipse.paho.client.mqttv3.version>
         <logback.configurationFileName>logback.xml</logback.configurationFileName>
+        <hivemq.mqtt.client.version>1.3.12</hivemq.mqtt.client.version>
     </properties>
 
 </project>

libraries-server/src/test/java/com/baeldung/mqtt/HiveMqMqttClientIntegrationTest.java

@@ -0,0 +1,72 @@
+package com.baeldung.mqtt;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+
+import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5BlockingClient;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
+
+class HiveMqMqttClientIntegrationTest {
+
+    private static final String PUBLIC_BROKER_HOST = "broker.hivemq.com";
+    private static final int PUBLIC_BROKER_PORT = 1883;
+
+    @Test
+    void givenSubscriber_whenMessageIsPublished_thenItIsReceived() throws Exception {
+        String topic = "baeldung/hivemq/test/" + UUID.randomUUID();
+        String payload = "Hello from Baeldung";
+
+        Mqtt5AsyncClient subscriber = Mqtt5Client.builder()
+            .identifier("baeldung-sub-" + UUID.randomUUID())
+            .serverHost(PUBLIC_BROKER_HOST)
+            .serverPort(PUBLIC_BROKER_PORT)
+            .buildAsync();
+
+        subscriber.connect()
+            .join();
+
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<String> receivedMessage = new AtomicReference<>();
+
+        subscriber.publishes(MqttGlobalPublishFilter.SUBSCRIBED, publish -> {
+            String message = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
+            receivedMessage.set(message);
+            latch.countDown();
+        });
+
+        subscriber.subscribeWith()
+            .topicFilter(topic)
+            .send()
+            .join();
+
+        Mqtt5BlockingClient publisher = Mqtt5Client.builder()
+            .identifier("baeldung-pub-" + UUID.randomUUID())
+            .serverHost(PUBLIC_BROKER_HOST)
+            .serverPort(PUBLIC_BROKER_PORT)
+            .buildBlocking();
+
+        publisher.connect();
+
+        publisher.publishWith()
+            .topic(topic)
+            .payload(payload.getBytes(StandardCharsets.UTF_8))
+            .send();
+
+        assertTrue(latch.await(2, TimeUnit.SECONDS));
+        assertEquals(payload, receivedMessage.get());
+
+        publisher.disconnect();
+        subscriber.disconnect()
+            .join();
+    }
+}