Setting up a Java MQTT Client

try {
client.disconnect();
System.out.println("Disconnected from broker.");
} catch (MqttException e) {
System.out.println("Error disconnecting from broker: " + e.getMessage());
e.printStackTrace();
}

MQTT 3.1.1

//paho mqttv3
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.SSLContext;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.security.KeyStore;
import java.io.FileInputStream;
import java.io.IOException;
import java.security.cert.CertificateException;
import java.security.NoSuchAlgorithmException;
import javax.net.ssl.TrustManagerFactory;
import java.security.KeyManagementException;
import java.util.Random;
import java.util.Scanner;
public class MqttClientv3 {
public static void main(String[] args) {
String client_id = "your_client_id";
String hostip = "public-mqtt-broker.bevywise.com";
boolean TLS = false; // set true to Enable TLS
boolean AUTH = true;
String client_username = "Access-Key";
String client_password = "Access-Token";
boolean websocket_connection=false; //set true to Enable websocket
String topic = "";
int qos = 0;
boolean retain = false;
String brokerUrl= "";
int port;
if(websocket_connection)
port= TLS ? 11433:10433;
else
port = TLS ? 8883 : 1883; // Use 8883 for SSL/TLS, 1883 for non-SSL

// Creating a new MQTT client instance
MqttClient client;
if(websocket_connection)
brokerUrl=String.format("%s://%s:%d", TLS ? "wss" : "ws", hostip, port);
else
brokerUrl = String.format("%s://%s:%d", TLS ? "ssl" : "tcp", hostip, port);

MemoryPersistence persistence = new MemoryPersistence();
try {
client = new MqttClient(brokerUrl, client_id, persistence);
// Setting callbacks
client.setCallback(new MqttCallback() {
@Override
public void connectionLost(Throwable cause) {
System.out.println("Connection lost: " + cause.getMessage());
cause.printStackTrace();
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
System.out.println("Received message on topic: " + topic);
System.out.println("Message: " + new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
System.out.println("Message delivered successfully.");
}
});
// Setting connection options
MqttConnectOptions connOpts = new MqttConnectOptions();
if (TLS) {
try {
connOpts.setSocketFactory(getSocketFactory("/path/to/your/root.crt")); //
Replace with your root cert path
} catch (Exception e) {
System.out.println("Error setting up SSL socket factory: " + e.getMessage());
e.printStackTrace();
}
}
if (AUTH) {
connOpts.setUserName(client_username);
connOpts.setPassword(client_password.toCharArray());
}
connOpts.setCleanSession(true);
// Connect to MQTT broker
System.out.println("Connecting to broker: " + brokerUrl);
client.connect(connOpts);
System.out.println("Connected");
// Example topic to subscribe
final String subscribeTopic = "will_topic/device1";
// Subscribing to a topic
client.subscribe(subscribeTopic);
// Publishing messages or subscribing/unsubscribing based on user input
Scanner scanner = new Scanner(System.in);
while (true) {
System.out.println("1. Publish\n2. Subscribe\n3. Disconnect\n4. Unsubscribe");
int choice = scanner.nextInt();
scanner.nextLine(); // consume newline
switch (choice) {
case 1:
System.out.print("Enter Topic QoS(0/1/2) Retain(0/1) Count: ");
String[] input = scanner.nextLine().split(" ");
topic = input[0];
qos = Integer.parseInt(input[1]);
retain = Integer.parseInt(input[2]) == 1;
int count = Integer.parseInt(input[3]);
for (int i = 0; i < count; i++) { String message="{\" temp\":" + (new
Random().nextInt(50) + 1) + ",\" pressure\":" + (new Random().nextInt(100) +
1) + ",\" status\":" + (new Random().nextInt(2)) + "}" ; try {
client.publish(topic, message.getBytes(), qos, retain);
System.out.println("Published message: " + message);
} catch (MqttException e) {
System.out.println(" Error publishing message: " + e.getMessage());
e.printStackTrace();
}
Thread.sleep(6000); // 6 seconds
}
break;
case 2:
System.out.print(" Enter Topic QoS(0/1/2): ");
String[] subInput = scanner.nextLine().split(" ");
topic = subInput[0];
qos = Integer.parseInt(subInput[1]);
try {
client.subscribe(topic, qos);
System.out.println(" Subscribed to topic: " + topic);
} catch (MqttException e) {
System.out.println(" Error subscribing to topic: " + e.getMessage());
e.printStackTrace();
}
break;
case 3:
try {
client.disconnect();
System.out.println(" Disconnected from broker."); } catch (MqttException e) { System.out.println("Error disconnecting
from broker: " + e.getMessage());
e.printStackTrace();
}
break;
case 4:
System.out.print(" Enter Topic to unsubscribe: ");
topic = scanner.nextLine();
try {
client.unsubscribe(topic);
System.out.println(" Unsubscribed from topic: " + topic);
} catch (MqttException e) {
System.out.println(" Error unsubscribing from topic: " + e.getMessage());
e.printStackTrace();
}
break;
default:
System.out.println(" Invalid choice."); } } } catch (MqttException | InterruptedException e) { System.out.println("An
error occurred: " + e.getMessage());
e.printStackTrace();
}
}
// Method to create SSL socket factory (if needed)
private static SSLSocketFactory getSocketFactory(String caCrtFile) throws Exception {
CertificateFactory cf = CertificateFactory.getInstance(" X.509"); FileInputStream caInput=new
FileInputStream(caCrtFile); X509Certificate caCert=(X509Certificate)
cf.generateCertificate(caInput); caInput.close(); KeyStore
keyStore=KeyStore.getInstance(KeyStore.getDefaultType());
keyStore.load(null, null); keyStore.setCertificateEntry("ca-certificate",
caCert); TrustManagerFactory
tmf=TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(keyStore); SSLContext context=SSLContext.getInstance("TLS");
context.init(null, tmf.getTrustManagers(), null); return
context.getSocketFactory(); } }

MQTT 5

//paho mqttv5
import org.eclipse.paho.mqttv5.client.IMqttClient;
import org.eclipse.paho.mqttv5.client.MqttClient;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
import org.eclipse.paho.mqttv5.client.MqttCallback;
import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
import org.eclipse.paho.mqttv5.client.IMqttToken;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.SSLContext;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import java.security.KeyStore;
import java.io.FileInputStream;
import java.io.IOException;
import java.security.cert.CertificateException;
import java.security.NoSuchAlgorithmException;
import javax.net.ssl.TrustManagerFactory;
import javax.net.ssl.KeyManagerFactory;
import java.security.KeyManagementException;
import java.util.List;
import java.util.Random;
import java.util.Scanner;
public class MqttClientv5 {
public static void main(String[] args)throws Exception{
String client_id = "your_client_id";
String hostip = "public-mqtt-broker.bevywise.com";
boolean AUTH = true;
String client_username = "Access-Key";
String client_password = "Access-Token";
boolean websocket_connection=false; //set true to Enable websocket
String topic = "";
int qos = 0;
boolean retain = false;
String brokerUrl= "";
int port;
if(websocket_connection)
port= 10433;
else
port = 1883; // Use 8883 for SSL/TLS, 1883 for non-SSL

// Creating a new MQTT client instance
MqttClient client;
if(websocket_connection)
brokerUrl=String.format("%s://%s:%d", "ws", hostip, port);
else
brokerUrl = String.format("%s://%s:%d", "tcp", hostip, port);

MemoryPersistence persistence = new MemoryPersistence();
try {
client = new MqttClient(brokerUrl, client_id, persistence);
// Setting callbacks
client.setCallback(new MqttCallback() {
@Override
public void disconnected(MqttDisconnectResponse disconnectResponse) {
System.out.println("Disconnected from broker.");
}
@Override
public void mqttErrorOccurred(MqttException exception) {
System.out.println("MQTT Error: " + exception.getMessage());
exception.printStackTrace();
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
System.out.println("Received message on topic: " + topic);
System.out.println("Message: " + new String(message.getPayload()));
}
@Override
public void deliveryComplete(IMqttToken token) {
System.out.println("Message delivered successfully.");
}
@Override
public void authPacketArrived(int reasonCode, MqttProperties properties) {
System.out.println("Auth packet arrived with reason code: " + reasonCode);
}
@Override
public void connectComplete(boolean reconnect, String serverURI) {
System.out.println("Connected to broker: " + serverURI);
}
// Setting connection options
MqttConnectionOptions connOpts = new MqttConnectionOptions();
if (TLS) {
try {
connOpts.setSocketFactory(getSocketFactory("/path/to/your/root.crt")); //
Replace with your root cert path
} catch (Exception e) {
System.out.println("Error setting up SSL socket factory: " + e.getMessage());
e.printStackTrace();
}
}
if (AUTH) {
connOpts.setUserName(client_username);
connOpts.setPassword(client_password.getBytes()); // Convert to byte array
}
connOpts.setCleanStart(true); // Clean start for MQTTv5
connOpts.setMaxInflight(50);
// Connect to MQTT broker
System.out.println("Connecting to broker: " + brokerUrl);
client.connect(connOpts);
System.out.println("Connected");
// Example topic to subscribe
final String subscribeTopic = "will_topic/device1";
// Subscribing to a topic
client.subscribe(subscribeTopic, qos);
// Publishing messages or subscribing/unsubscribing based on user input
Scanner scanner = new Scanner(System.in);
while (true) {
System.out.println("1. Publish\n2. Subscribe\n3. Disconnect\n4. Unsubscribe");
int choice = scanner.nextInt();
scanner.nextLine(); // consume newline
switch (choice) {
case 1:
System.out.print("Enter Topic QoS(0/1/2) Retain(0/1) Count: ");
String[] input = scanner.nextLine().split(" ");
topic = input[0];
qos = Integer.parseInt(input[1]);
retain = Integer.parseInt(input[2]) == 1;
int count = Integer.parseInt(input[3]);
for (int i = 0; i < count; i++) { //String message="{\" temp\":" + (new
Random().nextInt(50) + 1) + ",\" pressure\":" + (new Random().nextInt(100) +
1) + ",\" status\":" + (new Random().nextInt(2)==0 ? "ON" : "OFF" ) + "}" ;
String message="hello world" ; MqttMessage mqttMessage=new
MqttMessage(message.getBytes()); mqttMessage.setQos(qos);
mqttMessage.setRetained(retain); System.out.println("Message: "+message);
client.publish(topic, mqttMessage);
System.out.println(" Packet: "+i+" - sent"); } break; case 2: System.out.print("Enter Topic QoS(0/1/2): ");
input = scanner.nextLine().split(" ");
topic = input[0];
qos = Integer.parseInt(input[1]);
client.subscribe(topic, qos);
break;
case 3:
client.disconnect();
System.out.println(" Disconnected from broker."); return; case 4: System.out.print("Enter Topic: ");
topic = scanner.nextLine();
client.unsubscribe(topic);
break;
default:
System.out.println(" Invalid choice. Please try again."); } } } catch (MqttException e) { System.out.println("Error
connecting to MQTT broker: " + e.getMessage());
e.printStackTrace();
}
}
private static SSLSocketFactory getSocketFactory(String rootCertPath) throws Exception {
CertificateFactory cf = CertificateFactory.getInstance(" X.509"); FileInputStream fis=new
FileInputStream(rootCertPath); X509Certificate caCert=(X509Certificate)
cf.generateCertificate(fis); fis.close(); KeyStore
ks=KeyStore.getInstance("JKS"); ks.load(null, null);
ks.setCertificateEntry("ca-certificate", caCert); TrustManagerFactory
tmf=TrustManagerFactory.getInstance("SunX509"); tmf.init(ks); SSLContext
sslContext=SSLContext.getInstance("TLSv1.2"); sslContext.init(null,
tmf.getTrustManagers(), null); return sslContext.getSocketFactory(); } }