Connect FlowMQ with MQTT Java Client

FlowMQ's native support for MQTT makes it a great choice for Java applications. This guide will walk you through using the popular Eclipse Paho Java Client to connect to FlowMQ, publish messages, and subscribe to topics.

Prerequisites

Before you start, ensure you have the following:

• A running instance of FlowMQ.
• The FlowMQ MQTT listener endpoint (e.g., your-flowmq-host:1883).
• A Java Development Kit (JDK) 8 or later.
• A dependency management tool like Maven or Gradle.

Dependencies

Add the Eclipse Paho client to your project's dependencies.

Maven

Add the following to your pom.xml:

<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>

Gradle

Add the following to your build.gradle:

implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5'

Connecting to FlowMQ

To get started, you need to create an MqttClient instance and connect it to your FlowMQ broker.

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;

public class Main {
    public static void main(String[] args) {
        String broker = "tcp://your-flowmq-host:1883";
        String clientId = "my-java-client";
        
        try {
            MqttClient client = new MqttClient(broker, clientId);
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);

            System.out.println("Connecting to broker: " + broker);
            client.connect(options);
            System.out.println("Connected.");

            // ... publish and subscribe logic here ...

        } catch (MqttException me) {
            me.printStackTrace();
        }
    }
}

Publishing Messages

Once connected, you can easily publish messages to any topic.

import org.eclipse.paho.client.mqttv3.MqttMessage;

// ... inside the try block after connecting ...
String topic = "flowmq/test/java";
String content = "Hello FlowMQ from Java Paho Client!";
int qos = 1;

System.out.println("Publishing message: " + content);
MqttMessage message = new MqttMessage(content.getBytes());
message.setQos(qos);
client.publish(topic, message);
System.out.println("Message published.");

Subscribing to Topics

To receive messages, subscribe to a topic and set up a callback handler.

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;

// ... inside the try block after connecting ...
client.setCallback(new MqttCallback() {
    public void connectionLost(Throwable cause) {
        System.out.println("Connection lost: " + cause.getMessage());
    }

    public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println("Message arrived. Topic: " + topic + " | Message: " + new String(message.getPayload()));
    }

    public void deliveryComplete(IMqttDeliveryToken token) {
        // Not used in this example
    }
});

client.subscribe("flowmq/test/java", 1);
System.out.println("Subscribed to topic: flowmq/test/java");

Full Example

Here is a complete example that connects, sets up a subscription, publishes a message, and then waits to receive it.

import org.eclipse.paho.client.mqttv3.*;

public class Main {
    public static void main(String[] args) {
        String broker = "tcp://your-flowmq-host:1883";
        String clientId = "JavaSampleClient";
        String topic = "flowmq/java/test";
        String content = "Hello from Paho!";
        int qos = 1;

        try {
            MqttClient client = new MqttClient(broker, clientId);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);

            // Set callback
            client.setCallback(new MqttCallback() {
                public void connectionLost(Throwable cause) {
                    System.out.println("Connection lost!");
                }
                
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("Received Message: " + new String(message.getPayload()));
                }
                
                public void deliveryComplete(IMqttDeliveryToken token) {
                    // Delivery is complete
                }
            });

            // Connect, Subscribe, Publish
            System.out.println("Connecting to broker...");
            client.connect(connOpts);
            System.out.println("Connected.");
            
            client.subscribe(topic, qos);
            System.out.println("Subscribed to topic: " + topic);
            
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(qos);
            client.publish(topic, message);
            System.out.println("Message published.");

            // Wait a bit to receive the message before disconnecting
            Thread.sleep(1000); 

            client.disconnect();
            System.out.println("Disconnected.");
            client.close();

        } catch (MqttException | InterruptedException me) {
            me.printStackTrace();
        }
    }
}

Additional Resources

• For more advanced features and configuration options, refer to the official Eclipse Paho project page.
• Explore FlowMQ's advanced features for MQTT.