package mqtt;

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

public class MqttServerExample {

 public static void main(String[] args) {
 String broker = "tcp://broker.emqx.io:1883";
 String clientId = "mqtt-server-mqttx_1e1de6c0";
 String topic = "test/ddqqbb";//test/ddqqbb
 String message = "Hello, MQTT !topic1122";
 String topicFilter = "test/topic1122";//test/topic1122
 try {
 MqttClient client = new MqttClient(broker, clientId);
 MqttConnectOptions connOpts = new MqttConnectOptions();
 connOpts.setCleanSession(true);

 client.connect(connOpts);

 MqttMessage mqttMessage = new MqttMessage(message.getBytes());
 mqttMessage.setQos(0);

 client.publish(topic, mqttMessage);
 System.out.println("Message published successfully!");
 new Thread(new Runnable() {
 @Override
 public void run() {
 int i = 0;
 while (true) {
 try {
 client.publish(topic, mqttMessage);
// System.out.println("Message published successfully!" + i++);
 Thread.sleep(1000);
 } catch (Exception e) {
 throw new RuntimeException(e);
 }
 }
 }
 }).start();
 client.subscribe(topicFilter);
// client.disconnect();
 client.setCallback(new MqttCallback() {
 @Override
 public void connectionLost(Throwable cause) {
 System.out.println("connectionLost:" + cause.getMessage());
 }

 @Override
 public void messageArrived(String topic, MqttMessage message) throws Exception {
 System.out.println("messageArrived:" + message);
 }

 @Override
 public void deliveryComplete(IMqttDeliveryToken token) {
// System.out.println("deliveryComplete:" + token);
 }
 });
 } catch (MqttException me) {
 System.out.println("Error: " + me.getMessage());
 }
 }
}