ipole_edge_server/cloud/websvr/controllers/mqtthandler.go

package controllers

import (
	"context"
	"sync"
	"time"

	"github.com/astaxie/beego"

	"lc/common/mqtt"
)

var _mqtthandleronce sync.Once
var _mqtthandlersingle *MqttHandler

func GetMqttHandler() *MqttHandler {
	_mqtthandleronce.Do(func() {
		_mqtthandlersingle = NewMqttHandler(beego.AppConfig.String("mqtt.server"), beego.AppConfig.String("mqtt.user"),
			beego.AppConfig.String("mqtt.password"), 3000)
	})
	return _mqtthandlersingle
}

type MqttHandler struct {
	mqtt      *mqtt.Client
	mu        sync.Mutex
	mapTopics map[string]mqtt.QOS
	Timeout   uint //超时时间,毫秒为单位
}

func NewMqttHandler(server, user, password string, timeout uint) *MqttHandler {
	o := MqttHandler{
		mapTopics: make(map[string]mqtt.QOS),
		Timeout:   timeout,
	}
	client, err := mqtt.NewClient(mqtt.ClientOptions{
		Servers:       []string{server},
		ClientID:      beego.AppConfig.String("mqtt.id"),
		Username:      user,
		Password:      password,
		AutoReconnect: true,
	}, &o)
	if client == nil || err != nil {
		beego.Error("未配置 MQTT Server")
		panic("未配置 MQTT Server")
		return nil
	}
	o.mqtt = client
	ctx, cancel := o.Ctx()
	defer cancel()
	err = client.Connect(ctx)
	return &o
}

func (o *MqttHandler) ConnectionLostHandler(err error) {
	beego.Error("MqttHandler.ConnectionLostHandler:MQTT连接已经断开,原因:", err.Error())
}

func (o *MqttHandler) OnConnectHandler() {
	beego.Debug("MqttHandler.OnConnectHandler:MQTT连接成功")
	//连接成功则订阅主题
	for k, v := range o.mapTopics {
		o.Subscribe(k, v)
	}
}

func (o *MqttHandler) GetWill() (string, string) {
	return "", ""
}

func (o *MqttHandler) Connect() error {
	if !o.mqtt.IsConnected() {
		ctx, cancel := o.Ctx()
		defer cancel()
		return o.mqtt.Connect(ctx)
	}
	return nil
}

func (o *MqttHandler) IsConnected() bool {
	return o.mqtt.IsConnected()
}

func (o *MqttHandler) Publish(topic string, payload []byte, qos mqtt.QOS) error {
	ctx, cancel := o.Ctx()
	defer cancel()
	return o.mqtt.Publish(ctx, topic, payload, qos)
}

func (o *MqttHandler) PublishString(topic string, payload string, qos mqtt.QOS) error {
	ctx, cancel := o.Ctx()
	defer cancel()
	return o.mqtt.PublishString(ctx, topic, payload, qos)
}

func (o *MqttHandler) PublishJSON(topic string, payload interface{}, qos mqtt.QOS) error {
	ctx, cancel := o.Ctx()
	defer cancel()
	return o.mqtt.PublishJSON(ctx, topic, payload, qos)
}

func (o *MqttHandler) Subscribe(topic string, qos mqtt.QOS) error {
	o.mu.Lock()
	defer o.mu.Unlock()
	if _, ok := o.mapTopics[topic]; !ok {
		o.mapTopics[topic] = qos
	}
	ctx, cancel := o.Ctx()
	defer cancel()
	return o.mqtt.Subscribe(ctx, topic, qos)
}

func (o *MqttHandler) Unsubscribe(ctx context.Context, topic string) error {
	o.mu.Lock()
	defer o.mu.Unlock()
	if _, ok := o.mapTopics[topic]; ok {
		delete(o.mapTopics, topic)
	}
	ctx, cancel := o.Ctx()
	defer cancel()
	return o.mqtt.Unsubscribe(ctx, topic)
}

func (o *MqttHandler) Handle(topic string, handler mqtt.MessageHandler) mqtt.Route {
	return o.mqtt.Handle(topic, handler)
}

func (o *MqttHandler) Ctx() (context.Context, context.CancelFunc) {
	return context.WithTimeout(context.Background(), time.Millisecond*time.Duration(o.Timeout))
}