package mqttClient

import (
	"PPOSUtils/utils"
	"encoding/hex"
	"encoding/json"
	//"errors"
	"fmt"
	mqttPkg "github.com/eclipse/paho.mqtt.golang"
	"time"
)

type MqttMessageHander interface {
	MessageHandler() mqttPkg.MessageHandler
}

func NewMqttSubMessage(hander func(deviceID, topic, msgID string, data utils.Mp)) *MqttSubMessage {
	return &MqttSubMessage{hander}
}

type MqttSubMessage struct {
	handler func(deviceID, topic, msgID string, mp utils.Mp)
}

func (that *MqttSubMessage) MessageHandler() mqttPkg.MessageHandler {
	return func(client mqttPkg.Client, msg mqttPkg.Message) {
		_topic := msg.Topic()
		_byte := []byte{
			byte(msg.MessageID()),
		}
		_msgID := hex.EncodeToString(_byte)
		_data := msg.Payload()
		_mp := utils.NewMP()
		if _err := json.Unmarshal(_data, &_mp); _err != nil {
			return
		}
		_slice := []string{
			"id",
			"deviceID",
		}
		for _, _v := range _slice {
			if !_mp.Has(_v) {
				return
			}
		}
		_deviceID := utils.ToString(_mp.Get("deviceID"))
		if !msg.Retained() {
			that.handler(_deviceID, _topic, _msgID, _mp)
		}
	}
}

///////////////////////////////////////////////////////////////

func NewMqttClient(host, userName, password string) *MqttClient {
	return &MqttClient{
		host:     host,
		userName: userName,
		password: password,
	}
}

type MqttClient struct {
	host       string
	userName   string
	password   string
	client     mqttPkg.Client
	hander     MqttMessageHander
	isPublish  bool
	subMap     map[string]bool
	deviceType string
}

func (that *MqttClient) SetPublishHandler(handler MqttMessageHander) *MqttClient {
	that.hander = handler
	that.isPublish = true
	return that
}

func (that *MqttClient) SetDeviceType(deviceType string) *MqttClient {
	that.deviceType = deviceType
	return that
}

func (that *MqttClient) Conn(clientID string) error {
	that.subMap = map[string]bool{}
	//设置连接参数
	clinetOptions := mqttPkg.NewClientOptions().AddBroker(that.host).SetUsername(that.userName).SetPassword(that.password)
	//设置客户端ID
	//_clientID := fmt.Sprintf("%d-%d", clientID, time.Now().Unix())
	_clientID := clientID
	clinetOptions.SetClientID(_clientID)
	//设置连接超时
	clinetOptions.SetConnectTimeout(time.Duration(60) * time.Second)
	//设置断开重链
	clinetOptions.SetAutoReconnect(true)
	//
	//clinetOptions.SetCleanSession(true)
	//
	if that.isPublish {
		clinetOptions.SetDefaultPublishHandler(that.hander.MessageHandler())
	}
	//创建客户端连接
	client := mqttPkg.NewClient(clinetOptions)
	//客户端连接判断
	token := client.Connect()
	token.WaitTimeout(time.Duration(60) * time.Second)
	token.Wait()
	if _err := token.Error(); _err != nil {
		return _err
	}
	that.client = client
	return nil
}

func (that *MqttClient) ReConn() {

}

func (that *MqttClient) getTopic(deviceCode string) string {
	_topic := that.deviceType + "_" + deviceCode
	return _topic
}

func (that *MqttClient) UnSubMsg(deviceCode string) {
	_topic := that.getTopic(deviceCode)
	_, _ok := that.subMap[_topic]
	if !_ok {
		return
	}
	_token := that.client.Unsubscribe(_topic)
	_token.WaitTimeout(time.Duration(1) * time.Second)
	_token.Wait()
	if _err := _token.Error(); _err != nil {
		fmt.Println("######## mqtt Unsubscribe error ######", _err.Error())
	}
	delete(that.subMap, _topic)
}

//SubMsg 消息消费
func (that *MqttClient) SubMsg(topic string, hander MqttMessageHander) error {
	_topic := that.getTopic(topic)
	_, _ok := that.subMap[_topic]
	if _ok {
		return nil
	}
	//消息处理函数
	_token := that.client.Subscribe(_topic, 2, hander.MessageHandler())
	_token.WaitTimeout(time.Duration(3) * time.Second)
	_token.Wait()
	if _err := _token.Error(); _err != nil {
		return _err
	}
	that.subMap[_topic] = true
	return nil
}

//SubMsg 消息消费
func (that *MqttClient) PubMsg(topic string, cmd interface{}) error {
	_topic := that.getTopic(topic)
	_token := that.client.Publish(_topic, 2, false, cmd)
	_token.WaitTimeout(time.Duration(3) * time.Second)
	_token.Wait()
	if _err := _token.Error(); _err != nil {
		return _err
	}
	return nil
}

//Close 关闭链接
func (that *MqttClient) Close() {
	that.client.Disconnect(250)
}
