snippet/mqtt/mqtt.go

package mqtt

import (
	"encoding/json"
	"errors"

	MQTT "github.com/eclipse/paho.mqtt.golang"
)

var (
	clientSubscribe, clientDistribute MQTT.Client
	config                            *Config
)

func Init(c *Config) {
	config = c
	if config == nil {
		config = &Config{
			Host: "tcp://127.0.0.1:1883",
		}
	}
}

func Options() *MQTT.ClientOptions {
	opts := MQTT.NewClientOptions()
	opts.AddBroker(config.Host)
	opts.Username = config.Username
	opts.Password = config.Password
	opts.ClientID = config.ClientID
	return opts
}

func New() (MQTT.Client, error) {
	if clientDistribute != nil {
		return clientDistribute, nil
	}
	opts := Options()
	clientDistribute = MQTT.NewClient(opts)
	token := clientDistribute.Connect()
	if token.Wait() && token.Error() != nil {
		clientDistribute = nil
		return nil, token.Error()
	}

	return clientDistribute, nil
}

// 订阅，当事件为nil时可获取client
func Subscribe(connHandler MQTT.OnConnectHandler, lostHandler MQTT.ConnectionLostHandler) error {
	if connHandler == nil {
		return errors.New("handler is nil")
	}

	opts := Options()
	opts.OnConnect = connHandler //当连接成功后调用注册
	if lostHandler != nil {
		opts.OnConnectionLost = lostHandler
	}

	if clientSubscribe == nil {
		clientSubscribe = MQTT.NewClient(opts)
	}

	token := clientSubscribe.Connect()
	if token.Wait() && token.Error() != nil {
		clientSubscribe = nil
		return token.Error()
	}

	return nil
}

// 取消订阅
func UnSubscribe(topic string) {
	if clientSubscribe == nil {
		return
	}
	c := clientSubscribe
	c.Unsubscribe(topic)
	c.Disconnect(250)
}

// 分发
func Distribute(topic *string, qos int, retained bool, payload *[]byte) error {
	c, err := New()
	if err != nil {
		return err
	}

	token := c.Publish(*topic, byte(qos), retained, *payload)
	if token.Wait() && token.Error() != nil {
		return token.Error()
	}
	return nil
}

// 分发Object
func DistributeObject(topic *string, qos int, retained bool, obj interface{}) error {
	payload, err := json.Marshal(obj)
	if err != nil {
		return err
	}
	return Distribute(topic, qos, retained, &payload)
}