JAVA - MQTT Client 구현하기, 토픽간 연결하기

1. 개요

  PAHO 라이브러리를 이용한 MQTT 기본 기능을 구현합니다. 추가로 Topic과 Topic을 연결 해주는 프로그램을 구현해봅니다. MQTT 기본 설명은 http://twinw.tistory.com/158에서 보시면 됩니다.

PAHO mqtt 라이브러리는 https://eclipse.org/paho/에서 다운받으실 수 있으며 아래 첨부해 두었습니다.

org.eclipse.paho.client.mqttv3_fat.jar

2. 프로그램 시나리오

  소스코드는 기본적인 MQTT Client 기능을 구현한 MQTT.java 파일과 Broker와 Broker간 연결을 해주는 MqttToMqtt.java 파일이 있습니다.

어떠한 것이 MQTTServer1로 메세지를 보내면 MqttToMqtt.class는 MqttServer1를 Subscribe를 한 후 MqttServer2로 읽은 메세지로 Publish합니다.

3. Source Code - MQTT.java

 MQTT Client 소스 코드입니다.

아래 해당 소스 코드를 수정하여 Publish에 대한 동작을 정의하실 수 있습니다.

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.MqttPersistenceException; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; public class MQTT implements MqttCallback{ private static String Broker; private static String Client_ID; private static String UserName; private static String Passwd; private static MqttAsyncClient Client; private static MqttMessage message; private static MemoryPersistence persistence; private static MqttConnectOptions connOpts; private static String topic; public MQTT(String broker, String client_id,String username, String passwd){ this.Broker = broker; this.Client_ID = client_id; this.UserName = username; this.Passwd = passwd; } public void init(String topic){ this.topic = topic; this.persistence = new MemoryPersistence(); try { Client = new MqttAsyncClient(this.Broker, this.Client_ID, this.persistence); Client.setCallback(this); connOpts = new MqttConnectOptions(); //if(Client_ID!=null && Passwd != null){ connOpts.setUserName(this.UserName); connOpts.setPassword(this.Passwd.toCharArray()); //} connOpts.setCleanSession(true); System.out.println("Connecting to broker: "+this.Broker); Client.connect(connOpts); System.out.println("Connected"); message = new MqttMessage(); } catch(MqttException me) { System.out.println("reason "+me.getReasonCode()); System.out.println("msg "+me.getMessage()); System.out.println("loc "+me.getLocalizedMessage()); System.out.println("cause "+me.getCause()); System.out.println("excep "+me); me.printStackTrace(); } try { Thread.sleep(3000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } public void disconnect(){ try { Client.disconnect(); Client.close(); } catch (MqttException e) { // TODO Auto-generated catch block e.printStackTrace(); } } public void publish(String msg, int qos){ message.setQos(qos); message.setPayload(msg.getBytes()); try { Client.publish(topic, message); } catch (MqttPersistenceException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (MqttException e) { // TODO Auto-generated catch block e.printStackTrace(); } } public void subscribe(int qos){ try { Client.subscribe(topic,qos); } catch (MqttException e) { // TODO Auto-generated catch block e.printStackTrace(); } } public String getTopic(){ return topic; } @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { System.out.println("Message arrived : " + new String(mqttMessage.getPayload(), "UTF-8")); } @Override public void connectionLost(Throwable cause) { System.out.println("Lost Connection." + cause.getCause()); } @Override public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { System.out.println("Message with " + iMqttDeliveryToken + " delivered."); } }

4. Source Code - MqttToMqtt.java

 MQTT Client 소스를 조금 수정하여 만든 소스 코드입니다.

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.MqttPersistenceException; import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; public class MqttToMqtt implements MqttCallback{ private static String Broker; private static String Client_ID; private static String UserName; private static String Passwd; private static MqttAsyncClient Client; private static MqttMessage message; private static MemoryPersistence persistence; private static MqttConnectOptions connOpts; private static String Topic; private static MQTT recvMQTT; public MqttToMqtt(String broker, String client_id,String username, String passwd, MQTT recvMQTT){ this.Broker = broker; this.Client_ID = client_id; this.UserName = username; this.Passwd = passwd; this.recvMQTT = recvMQTT; } public void init(){ this.Topic = recvMQTT.getTopic(); this.persistence = new MemoryPersistence(); try { Client = new MqttAsyncClient(this.Broker, this.Client_ID, this.persistence); Client.setCallback(this); connOpts = new MqttConnectOptions(); //if(Client_ID!=null && Passwd != null){ connOpts.setUserName(this.UserName); connOpts.setPassword(this.Passwd.toCharArray()); //} connOpts.setCleanSession(true); System.out.println("Connecting to broker: "+this.Broker); Client.connect(connOpts); System.out.println("Connected"); message = new MqttMessage(); } catch(MqttException me) { System.out.println("reason "+me.getReasonCode()); System.out.println("msg "+me.getMessage()); System.out.println("loc "+me.getLocalizedMessage()); System.out.println("cause "+me.getCause()); System.out.println("excep "+me); me.printStackTrace(); } try { Thread.sleep(1000); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } this.subscribe(0); } public void disconnect(){ try { Client.disconnect(); Client.close(); } catch (MqttException e) { // TODO Auto-generated catch block e.printStackTrace(); } } private void publish(String msg, int qos){ message.setQos(qos); message.setPayload(msg.getBytes()); try { Client.publish(Topic, message); } catch (MqttPersistenceException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (MqttException e) { // TODO Auto-generated catch block e.printStackTrace(); } } public void subscribe(int qos){ try { Client.subscribe(Topic,qos); } catch (MqttException e) { // TODO Auto-generated catch block e.printStackTrace(); } } @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { System.out.println("Message arrived : " + new String(mqttMessage.getPayload(), "UTF-8")); String msg = new String(mqttMessage.getPayload(), "UTF-8"); recvMQTT.publish(msg,0); } @Override public void connectionLost(Throwable cause) { System.out.println("Lost Connection." + cause.getCause()); } @Override public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { System.out.println("Message with " + iMqttDeliveryToken + " delivered."); } }

5. Sample Code

 Sample 코드입니다. 

MQTT.zip

public class Main{ public static void main(String[] args){ String MqttServer1 = "tcp://52.52.52.52:1883"; String MqttServer2 = "ssl://86.86.86.86:8883"; String client_id = "Input your ClientID"; String username = "Input your ssl_username"; String passwd = "Input your ssl_passwd"; String topic = "Input your TOPIC"; String msg = "Input your Message"; //Receive message from Mqtt not Machine MQTT ReadFromOtherMQTT = new MQTT(MqttServer2, client_id, username, passwd); ReadFromOtherMQTT.init(topic); sleep(1000); ReadFromOtherMQTT.subscribe(0); sleep(1000); //Receive message from Machine and Send to other MQTT MqttToMqtt ReadAndSend = new MqttToMqtt(MqttServer1, client_id, username, passwd, ReadFromOtherMQTT); ReadAndSend.init(); } static void sleep(int time){ try { Thread.sleep(time); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }