package xn.xxp.mqtt

import android.os.Handler
import android.os.HandlerThread
import android.os.Looper
import android.os.Message
import android.util.Log
import com.blankj.utilcode.util.LogUtils
import org.eclipse.paho.client.mqttv3.IMqttActionListener
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken
import org.eclipse.paho.client.mqttv3.IMqttToken
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.persist.MqttDefaultFilePersistence

/**
 * info
 *
 * @author ReiChin_
 */
class MqttWorker(
    serverURI: String,
    clientId: String,
    private val uName: String,
    private val uPwd: String
) {

    companion object {
        private const val MSG_CONNECT = 233
        private const val MSG_RELEASE = 234
    }

    private var mReleased: Boolean = false
    private var mReconnectPost: Boolean = false

    private var mClient: LabMqttClient? = null
    private val mConnectOpt: MqttConnectOptions
    private val mInnerHandler: InnerHandler
    private val mHandlerThread = HandlerThread("mqtt-handler-thread")

    init {
        mHandlerThread.start()
        mInnerHandler = InnerHandler(mHandlerThread.looper, this)

        // 连接参数
        mConnectOpt = MqttConnectOptions().apply {
            // 清除缓存
            isCleanSession = true
            // 设置超时时间，单位：秒
            connectionTimeout = 60
            // 心跳包发送间隔，单位：秒
            keepAliveInterval = 60
            // 用户名
            userName = uName
            // 密码
            password = uPwd.toCharArray()
        }

        try {
            val tmpDir = System.getProperty("java.io.tmpdir")
            mClient = LabMqttClient(serverURI, clientId, MqttDefaultFilePersistence(tmpDir))
        } catch (e: Exception) {
            e.printStackTrace()
            // 正常来说，这里不可达，真出现就再把异常抛出去
//            throw RuntimeException(e)
        }
    }

    private class InnerHandler(looper: Looper, var mqttWorker: MqttWorker?) : Handler(looper) {

        override fun handleMessage(msg: Message) {
            if (null == mqttWorker) return

            when (msg.what) {
                MSG_CONNECT -> mqttWorker?.connect()
                MSG_RELEASE -> mqttWorker?.release()
            }
        }
    }

    /**
     * mqtt丢失连接，接收消息等回调
     */
    private val mMqttCallback = object : MqttCallback {
        /**
         * 连接丢失，重新连接
         */
        override fun connectionLost(cause: Throwable?) {
            reconnect()
        }

        /**
         * 消息到达
         */
        override fun messageArrived(topic: String?, message: MqttMessage?) {
            mListener?.messageArrived(topic, message)
        }

        /**
         * 消息投递成功
         */
        override fun deliveryComplete(token: IMqttDeliveryToken?) {
        }
    }

    /**
     * 连接回调
     */
    private val mMqttActionListener = object : IMqttActionListener {

        /**
         * 连接成功
         */
        override fun onSuccess(asyncActionToken: IMqttToken?) {
            mListener?.onConnectSuccess()
        }

        /**
         * 连接失败
         */
        override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
            reconnect()
        }
    }

    @Synchronized
    private fun reconnect() {
        if (mReleased || mReconnectPost) {
            return
        }

        // 设置已经post重新连接
        mReconnectPost = true

        mInnerHandler.removeCallbacks(mRunReconnect)
        mInnerHandler.postDelayed(mRunReconnect, 2000)
    }

    /**
     * 重连
     */
    private val mRunReconnect = Runnable {
        synchronized(MqttWorker) {
            disconnect()
            connect()
            // 设置重连完成
            mReconnectPost = false
        }
    }

    /**
     * 连接
     */
    @Synchronized
    fun connect() {
        if (Thread.currentThread() != mHandlerThread) {
            mInnerHandler.sendEmptyMessage(MSG_CONNECT)
            return
        }
        if (null == mClient) return

        if (mClient!!.isConnected) {
            return
        }

        try {
            mClient!!.setCallback(mMqttCallback)
            mClient!!.actionListener = mMqttActionListener
            mClient!!.connect(mConnectOpt)
        } catch (e: MqttException) {
            e.printStackTrace()
            mMqttActionListener.onFailure(null, e)
        }
    }

    /**
     * 断开连接
     */
    @Synchronized
    private fun disconnect() {
        try {
            if (null == mClient) return

            mClient?.let {
                if (it.isConnected) it.disconnect()
            }
        } catch (e: MqttException) {
            e.printStackTrace()
        }
    }

    /**
     * 订阅主题
     *
     * @param topicFilter
     * @param qos 1.发送者最少发送一次消息，确保消息到达Broker，Broker需要返回确认消息PUBACK。在Qos1情况下，Broker可能接受到重复消息
     * 2.Qos2使用两阶段确认来保证消息的不丢失和不重复。在Qos2情况下，Broker肯定会收到消息，且只收到一次
     * 0.发送者只发送一次消息，不进行重试，Broker不会返回确认消息。在Qos0情况下，Broker可能没有接受到消息
     */
    fun subscribe(topicFilter: String, qos: Int) {

        if (Thread.currentThread() != mHandlerThread) {
            mInnerHandler.post { subscribe(topicFilter, qos) }
            return
        }

        if (null == mClient) return

        try {
            mClient!!.subscribe(topicFilter, qos)
        } catch (e: MqttException) {
            LogUtils.e(Log.getStackTraceString(e))
        }
    }

    /**
     * 订阅主题
     *
     * @param topics
     * @param qos 1.发送者最少发送一次消息，确保消息到达Broker，Broker需要返回确认消息PUBACK。在Qos1情况下，Broker可能接受到重复消息
     * 2.Qos2使用两阶段确认来保证消息的不丢失和不重复。在Qos2情况下，Broker肯定会收到消息，且只收到一次
     * 0.发送者只发送一次消息，不进行重试，Broker不会返回确认消息。在Qos0情况下，Broker可能没有接受到消息
     */
    fun subscribe(topics: Array<String>, qos: Int) {

        if (Thread.currentThread() != mHandlerThread) {
            mInnerHandler.post { subscribe(topics, qos) }
            return
        }

        if (null == mClient) return

        try {
            topics.forEach { topic ->
                mClient!!.subscribe(topic, qos)
            }
        } catch (e: MqttException) {
            LogUtils.e(Log.getStackTraceString(e))
        }
    }

    /**
     * 发布消息
     *
     * @param topic
     * @param message
     * @param qos 1.发送者最少发送一次消息，确保消息到达Broker，Broker需要返回确认消息PUBACK。在Qos1情况下，Broker可能接受到重复消息
     * 2.Qos2使用两阶段确认来保证消息的不丢失和不重复。在Qos2情况下，Broker肯定会收到消息，且只收到一次
     * 0.发送者只发送一次消息，不进行重试，Broker不会返回确认消息。在Qos0情况下，Broker可能没有接受到消息
     */
    fun publish(topic: String, message: String, qos: Int) {
        if (Thread.currentThread() != mHandlerThread) {
            mInnerHandler.post { publish(topic, message, qos) }
            return
        }

        val mqttMessage = MqttMessage(message.toByteArray())
        mqttMessage.qos = qos

        if (null == mClient) return

        if (mClient!!.isConnected) {
            try {
                mClient!!.publish(topic, mqttMessage)
            } catch (e: MqttException) {
                mMqttCallback.connectionLost(e)
            }
        } else {
            mMqttCallback.connectionLost(IllegalStateException("MQTT未连接"))
        }
    }

    @Synchronized
    fun release() {
        if (Thread.currentThread() != mHandlerThread) {
            mInnerHandler.sendEmptyMessage(MSG_RELEASE)
            return
        }
        mReleased = true
        mInnerHandler.mqttWorker = null
        disconnect()
        mHandlerThread.quitSafely()
    }

    interface Listener {

        /**
         * 连接成功
         */
        fun onConnectSuccess()

        /**
         * 消息到达
         *
         * @param topic
         * @param message
         */
        fun messageArrived(topic: String?, message: MqttMessage?)
    }

    private var mListener: Listener? = null

    fun setListener(listener: Listener?) {
        mListener = listener
    }

}