[Kotlin] Android MQTT + MQTTX 설정하기

MQTT구조

MQTT는 크게 세 가지 Broker, Publisher, Subscriber 구조로 이루어져 있습니다.

Publisher는 메시지(데이터)를 발행하는 역할

Subscriber는 메세지(데이터)를 구독하는 역할

Borker는 메시지를 Publisher로부터 받아와 Subscriber가 가져갈 수 있도록 전달하는 역할

이라고 생각하시면 됩니다.

 

아래 구조도를 보면서 설명하겠습니다.

(▲간략한 구조도)

 

 

1. Gradle 추가

// MQTT eclipse
implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5'
// MQTT hannesa2
implementation 'com.github.hannesa2:paho.mqtt.android:3.5.1'

 

 

2. 코드 작성

class TempFragment : Fragment() {
    
    // mqtt 관련
    private val serverUri = "tcp://mqtt@broker.emqx.io:1883" // 서버 URI (=브로커 URI)
    
    // Publish a message
    private val topic = "mia/topic" // 토픽
    private val qos = 1
    private val retained = false

    private var sendText = ""
    private var receiveText = ""
    
    // Mqtt 방식의 통신을 지원하는 클래스, MqttAndroidClient 객체 생성
    private lateinit var mqttClient: MqttAndroidClient

    override fun onCreateView(
        inflater: LayoutInflater,
        container: ViewGroup?,
        savedInstanceState: Bundle?
    ): View? {
        BleDebugLog.i(logTag, "onCreateView-()")
        binding = DataBindingUtil.inflate(inflater, R.layout.fragment_temp, container, false)
        with(binding) {
            viewModel = tempViewModel
            lifecycleOwner = viewLifecycleOwner
        }
        return binding.root
    }

    override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
        super.onViewCreated(view, savedInstanceState)
        BleDebugLog.i(logTag, "onViewCreated-()")
        
        // 연결
        binding.connectBtn.setOnClickListener {
            connectMqtt()
        }
        // 전송
        binding.sendBtn.setOnClickListener {
            sendMessageMqtt()
        }
        // 연결 해제
        binding.disconnectBtn.setOnClickListener {
            disconnectMqtt()
        }
        
    }
  }

 

 

3. MQTT 연결하기

/**
 * MQTT 연결
 */
private fun connectMqtt() {
    BleDebugLog.i(logTag, "connectMqtt-()")
    // MqttAndroidClient 초기화
    // → 기본적으로 AUTO_ACT로 동작함 : 메시지가 반환되면 프로세스에 메시지가 도착했다고 즉시 알림
    mqttClient = MqttAndroidClient(requireContext(), serverUri, MqttClient.generateClientId())
    // MqttConnectOptions는 Mqtt의 Client가 서버에 연결하는 방법을 제어하는 클래스
    val connectOptions = MqttConnectOptions()
    connectOptions.isCleanSession = true
    // Mqtt 서버와 연결
    // 연결 결과 콜백 등록 → callbackConnectResult
    mqttClient.connect(connectOptions, null, callbackConnectResult)
}

MQTT serverUri (브로커 Uri) 를 넣어서 초기화 한다.

connect() 연결한다.

 

MQTTX 설정은 아래와 같다.

 

 

 

4. 연결 결과 콜백

/**
 * connect 결과 처리
 */
private var callbackConnectResult = object : IMqttActionListener {
    override fun onSuccess(asyncActionToken: IMqttToken?) {
        BleDebugLog.i(logTag, "onSuccess-()  :: 연결 성공")
        BleDebugLog.d(logTag, "asyncActionToken: $asyncActionToken")
        // 연결에 성공하면 해당 토픽 구독
        mqttClient.subscribe(topic, 1)
        mqttCallBack()
        sendMessageMqtt()
    }

    override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
        BleDebugLog.i(logTag, "onFailure-() :: 연결 실패")
        BleDebugLog.e("$exception")
    }
}

연결 성공하면, 토픽을 subcribe(구독)한다.

모바일 = Subscriber 구독자

 

 

5. 메시지 전송

/**
 * 메시지 전송
 */
private fun sendMessageMqtt() {
    mqttClient.publish(topic, binding.msg.text.toString().toByteArray(), qos, retained) // 메세지 전송
    sendText = sendText + binding.msg.text.toString() + "\n"
    binding.firstTextView.text = sendText
}

현재는,

실제 서버<-> 브로커 서버<-> 모바일 이 아닌,

브로커 서버 <-> 모바일 이기 때문에

브로커 서버에서 데이터를 publish (발행) 하는 걸로 하였다.

 

+ 모바일에서 메시지 전송 (데이터 publish), MQTTX 에서 데이터 전송 (데이터 publish) 둘 다 할 수 있는 구조이다. 

 

 

6. 메시지 상태 콜백

/**
 * 메시지 상태 콜백
 */
private fun mqttCallBack() {
    // 콜백 설정
    mqttClient.setCallback(object : MqttCallback {
        // 연결이 끊겼을 경우
        override fun connectionLost(p0: Throwable?) {
            BleDebugLog.i(logTag, "connectionLost-() :: 연결 끊어짐")
        }

        // 메세지가 도착했을 때
        override fun messageArrived(topic: String?, message: MqttMessage?) {
            BleDebugLog.i(logTag, "messageArrived-() :: 메시지 도착")
            BleDebugLog.d(logTag, "topic: $topic")
            BleDebugLog.d(logTag, "message: $message")
            receiveText = receiveText + message.toString() + "\n\n"
            binding.secondTextView.text = receiveText
        }

        // 메시지 전송이 성공했을 때
        override fun deliveryComplete(p0: IMqttDeliveryToken?) {
            BleDebugLog.i(logTag, "deliveryComplete-() :: 메시지 전송 성공")
        }
    })
}

 

 

1. 모바일에서 데이터 발행 했을 때, Logcat

 

2. MQTTX에서 데이터 발행 했을 때, Logcat

 

 

MQTT 연결 중일 때,

데이터 발행이 이뤄지면 이를 구독하고 있던 모바일은 mqttClient에 등록된 콜백에서 메시지를 실시간으로 받는다!

 

 

 

 

출처: https://ogyong.tistory.com/29