FranklinWH-Toolbox/service/utils/mqttClient/mqttClient.go

package mqttClient

import (
	"crypto/tls"
	"errors"
	"fmt"
	"log"
	"net/url"
	"os"
	"reflect"
	"sync"
	"time"
	"unsafe"

	mqtt "github.com/eclipse/paho.mqtt.golang"
	"go.uber.org/zap"
)

type MqttClientOpt struct {
	Broker   string // 服务端地址
	ClientID string // (可选) 客户端名称
	TLS      *tls.Config
}

type MqttClient struct {
	client        mqtt.Client
	clientOptions *mqtt.ClientOptions
	subscribeMap  map[string]func(payload []byte)
	subscribeMu   sync.Mutex // 保护 subscribeMap（OnConnect 回调与调用方在不同 goroutine）
	clientID      string     // 客户端名称
}

const (
	connectTimeout      = 30 * time.Second // 30s
	reconnectTimeout    = 10 * time.Second // 10s
	disconnectTimeout   = 10 * 1000        // 10s
	subscribeMsgTimeout = 30 * time.Second // 30s
	sendMsgTimeout      = 30 * time.Second // 30s
)

// 创建客户端
func CreateMqttClient(opt MqttClientOpt) (*MqttClient, error) {
	// 初始化前设置（放在 InitFwhMqttClient 开头或 init() 中）
	mqtt.DEBUG = log.New(os.Stdout, "[paho-DEBUG] ", log.Ltime|log.Lmicroseconds)
	mqtt.WARN = log.New(os.Stdout, "[paho-WARN]  ", log.Ltime|log.Lmicroseconds)
	mqtt.ERROR = log.New(os.Stderr, "[paho-ERROR] ", log.Ltime|log.Lmicroseconds)

	// 创建客户端选项
	opts := mqtt.NewClientOptions()
	opts.AddBroker(opt.Broker)
	clientID := "go_mqtt_client"
	if len(opt.ClientID) > 0 {
		clientID = opt.ClientID
	}
	opts.SetClientID(clientID)

	opts.SetAutoReconnect(true) // 自动重连
	opts.SetCleanSession(true)  // 不保留会话
	opts.SetTLSConfig(opt.TLS)

	mqttClient := MqttClient{
		subscribeMap: map[string]func(payload []byte){},
		clientID:     clientID,
	}

	// 连接成功回调
	opts.OnConnect = func(c mqtt.Client) {
		zap.L().Info("Connected to MQTT broker", zap.String("clientID", clientID))

		// 重新订阅丢失的 Subscribe
		mqttClient.ResumeSubscribe()
	}

	// 连接丢失回调
	opts.OnConnectionLost = func(c mqtt.Client, err error) {
		fmt.Printf("%s Connection lost: %s\n", clientID, err.Error())
		zap.L().Info("MQTT Connection lost", zap.String("clientID", clientID), zap.String("err", err.Error()))
	}

	// 重连回调
	opts.OnReconnecting = func(c mqtt.Client, opts *mqtt.ClientOptions) {
		zap.L().Info("MQTT Reconnecting", zap.String("clientID", clientID))
	}

	// 尝试连接回调
	opts.OnConnectAttempt = func(broker *url.URL, tlsCfg *tls.Config) *tls.Config {
		zap.L().Info("MQTT Attempt Connect", zap.String("clientID", clientID), zap.String("Broker", broker.String()))
		return tlsCfg
	}

	// 连接事件回调
	opts.OnConnectionNotification = func(c mqtt.Client, cn mqtt.ConnectionNotification) {
		zap.L().Info("MQTT Connecting Notification", zap.String("clientID", clientID), zap.Int64("Notification", int64(cn.Type())))
	}

	// 创建客户端
	mqttClient.client = mqtt.NewClient(opts)
	mqttClient.clientOptions = opts

	return &mqttClient, nil
}

func (c *MqttClient) RecreateClient() error {
	if c.clientOptions == nil {
		return errors.New("RecreateClient failed, ClientOptions is nil.")
	}

	// 尝试断开旧连接（非阻塞）
	oldClient := c.client
	go func() {
		if oldClient != nil {
			done := make(chan struct{})
			go func() {
				oldClient.Disconnect(disconnectTimeout)
				close(done)
			}()
			select {
			case <-done:
			case <-time.After(15 * time.Second):
				zap.L().Warn("[MQTT] old client disconnect timeout, abandoning")
			}
		}
	}()

	// 重新创建客户端
	c.client = mqtt.NewClient(c.clientOptions)
	zap.L().Info("[MQTT] create new client")
	return nil
}

func (c *MqttClient) Connect() error {
	isConnected := c.client.IsConnected()
	isOpen := c.client.IsConnectionOpen()

	zap.L().Debug("[MQTT] Connect() status check",
		zap.Bool("IsConnected", isConnected),
		zap.Bool("IsConnectionOpen", isOpen),
	)

	// 底层连接已活跃，无需再连
	if isOpen {
		return nil
	}

	// IsConnected=true 但连接不活跃 → 正在重连中，等待恢复
	if isConnected && !isOpen {
		zap.L().Info("[MQTT] client is reconnecting, waiting for connection to restore")
		deadline := time.Now().Add(reconnectTimeout)
		for time.Now().Before(deadline) {
			if c.client.IsConnectionOpen() {
				return nil
			}
			// 如果 IsConnected 也变成 false 了，说明重连失败/放弃了
			if !c.client.IsConnected() {
				zap.L().Warn("[MQTT] client lost connection during reconnect wait, will re-connect")
				break
			}
			time.Sleep(500 * time.Millisecond)
		}

		// 等待超时或重连失败，检查最终状态
		if c.client.IsConnectionOpen() {
			return nil
		}

		// 强制断开后重连（清除 disconnecting 卡死状态）
		// zap.L().Warn("[MQTT] reconnect failed, forcing disconnect and re-connect")
		// c.client.Disconnect(0)
		// time.Sleep(100 * time.Millisecond)
	}

	// 真正断开状态，执行连接
	token := c.client.Connect()
	if !token.WaitTimeout(connectTimeout) {
		if err := c.RecreateClient(); err != nil {
			return fmt.Errorf("recreate client failed (%s)", c.clientID)
		}

		token = c.client.Connect()
		if !token.WaitTimeout(connectTimeout) {
			return fmt.Errorf("connect timeout (%s)", c.clientID)
		}
	}

	return token.Error()
}

// 断开连接
func (c *MqttClient) Disconnect() {
	// 清空订阅
	c.CleanSubscribe()

	// 安全断开，等待最多 10 秒让队列消息发送完
	c.client.Disconnect(disconnectTimeout)
}

// 检查是否连接
func (c *MqttClient) IsConnected() bool {
	return c.client.IsConnected()
}

// 恢复订阅消息
func (c *MqttClient) ResumeSubscribe() {
	c.subscribeMu.Lock()
	subs := make(map[string]func(payload []byte), len(c.subscribeMap))
	for k, v := range c.subscribeMap {
		subs[k] = v
	}
	c.subscribeMu.Unlock()

	for k, v := range subs {
		if err := c.subscribeMessage(k, v); err != nil {
			zap.L().Error("[MQTT] resume subscribe failed", zap.String("topic", k), zap.String("err", err.Error()))
		}
	}
}

// 清空所有订阅
func (c *MqttClient) CleanSubscribe() {
	c.subscribeMu.Lock()
	defer c.subscribeMu.Unlock()
	for k := range c.subscribeMap {
		delete(c.subscribeMap, k)
	}
}

// 订阅消息
func (c *MqttClient) subscribeMessage(topic string, callback func(payload []byte)) error {
	if !c.client.IsConnected() {
		return fmt.Errorf("subscribe failed. (client not connected)")
	}

	token := c.client.Subscribe(topic, 1, func(c mqtt.Client, m mqtt.Message) {
		callback(m.Payload())
	})
	if !token.WaitTimeout(subscribeMsgTimeout) {
		return fmt.Errorf("subscribe timeout (topic:%s)", topic)
	}

	err := token.Error()
	if err != nil {
		return err
	}

	return nil
}

// 订阅消息
func (c *MqttClient) SubscribeMessage(topic string, callback func(payload []byte)) error {
	err := c.subscribeMessage(topic, callback)
	if err != nil {
		return err
	}

	// 备份subscribe
	c.subscribeMu.Lock()
	c.subscribeMap[topic] = callback
	c.subscribeMu.Unlock()

	return nil
}

// 取消订阅
func (c *MqttClient) UnSubscribeMessage(topic string) error {
	if !c.client.IsConnected() {
		return fmt.Errorf("unsubscribe failed. (client not connected)")
	}

	token := c.client.Unsubscribe(topic)
	if !token.WaitTimeout(subscribeMsgTimeout) {
		return fmt.Errorf("unsubscribe timeout (topic:%s)", topic)
	}

	err := token.Error()
	if err != nil {
		return err
	}

	c.subscribeMu.Lock()
	delete(c.subscribeMap, topic)
	c.subscribeMu.Unlock()

	return nil
}

// 发送消息，只尝试一次，不管是否送达
func (c *MqttClient) SendMessageOnce(topic string, msg []byte) error {
	if !c.client.IsConnected() {
		return fmt.Errorf("send message once failed. (client not connected)")
	}

	token := c.client.Publish(topic, 0, false, msg)
	if !token.WaitTimeout(sendMsgTimeout) {
		return fmt.Errorf("send message once timeout (topic:%s, msg:%s)", topic, msg)
	}

	return token.Error()
}

// 发送消息，没收到 ACK 就重发，至少送到一次，可能收到多次
func (c *MqttClient) SendMessage(topic string, msg []byte, retained bool) error {
	if !c.client.IsConnected() {
		return fmt.Errorf("send message failed. (client not connected)")
	}

	token := c.client.Publish(topic, 1, retained, msg)
	if !token.WaitTimeout(sendMsgTimeout) {
		return fmt.Errorf("send message timeout (topic:%s, msg:%s)", topic, msg)
	}

	return token.Error()
}

// 发送消息，严格一次，不重复，不丢失
func (c *MqttClient) SendMessageMustOnce(topic string, msg []byte, retained bool) error {
	if !c.client.IsConnected() {
		return fmt.Errorf("send message must once failed. (client not connected)")
	}

	token := c.client.Publish(topic, 2, retained, msg)
	if !token.WaitTimeout(sendMsgTimeout) {
		return fmt.Errorf("send message must once timeout (topic:%s, msg:%s)", topic, msg)
	}

	return token.Error()
}

// 发送消息，没收到 ACK 就重发，至少送到一次，可能收到多次，并保留消息在Broker，新定义的设备会收到消息
func (c *MqttClient) SendMessageRetained(topic string, msg []byte) error {
	if !c.client.IsConnected() {
		return fmt.Errorf("send message retained failed. (client not connected)")
	}

	token := c.client.Publish(topic, 1, true, msg)
	if !token.WaitTimeout(sendMsgTimeout) {
		return fmt.Errorf("send message retained timeout (topic:%s, msg:%s)", topic, msg)
	}

	return token.Error()
}

// 清除保留在Broker的消息
func (c *MqttClient) CleanMessageRetained(topic string) error {
	if !c.client.IsConnected() {
		return fmt.Errorf("clean message retained failed. (client not connected)")
	}

	token := c.client.Publish(topic, 1, true, nil)
	if !token.WaitTimeout(sendMsgTimeout) {
		return fmt.Errorf("clean message retained timeout (topic:%s)", topic)
	}

	return token.Error()
}

// getClientStatus 通过反射强制获取 paho client 内部的连接状态（仅调试用）
// 返回值: 0=disconnected, 1=disconnecting, 2=connecting, 3=reconnecting, 4=connected
func (c *MqttClient) GetClientStatus() string {
	statusMap := map[uint32]string{
		0: "disconnected",
		1: "disconnecting",
		2: "connecting",
		3: "reconnecting",
		4: "connected",
	}

	// mqtt.Client 是接口，需要先拿到底层 *client 指针
	rv := reflect.ValueOf(c.client).Elem() // *client -> client struct

	// 获取 status 字段 (类型: connectionStatus)
	statusField := rv.FieldByName("status")
	if !statusField.IsValid() {
		return "unknown(field not found)"
	}

	// connectionStatus 结构体中第一个字段之后是 status uint32
	// connectionStatus 内嵌了 sync.RWMutex，RWMutex 占据前面的空间
	// 实际上 status 是 connectionStatus 结构体中名为 "status" 的字段
	innerStatus := statusField.FieldByName("status")
	if !innerStatus.IsValid() {
		return "unknown(inner status not found)"
	}

	// 强制读取未导出字段
	ptr := unsafe.Pointer(innerStatus.UnsafeAddr())
	val := *(*uint32)(ptr)

	if name, ok := statusMap[val]; ok {
		return name
	}
	return fmt.Sprintf("unknown(%d)", val)
}