105 lines
2.0 KiB
Go
105 lines
2.0 KiB
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)
|
||
}
|