Go 搭建 Coolpy+MQTT 服务

# MQTT 服务配置
mqtt:
  broker: "tcp://localhost:1883"  # MQTT  Broker 地址
  client_id: "go-coolpy-client"   # 客户端 ID
  username: "admin"               # 用户名（若有）
  password: "123456"              # 密码（若有）
  topic: "iot/devices/data"       # 订阅和发布的主题
# Coolpy 配置
coolpy:
  api_key: "your-coolpy-api-key"  # 从 Coolpy 控制台获取的 API 密钥
  device_id: "dev-001"            # 设备 ID

package main

import (
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    MQTT "github.com/eclipse/paho.mqtt.golang"
    "gopkg.in/yaml.v3"
)

// Config 定义配置结构体，对应 config.yaml 文件
type Config struct {
    MQTT   MQTTConfig   `yaml:"mqtt"`
    Coolpy CoolpyConfig `yaml:"coolpy"`
}

type MQTTConfig struct {
    Broker   string `yaml:"broker"`
    ClientID string `yaml:"client_id"`
    Username string `yaml:"username"`
    Password string `yaml:"password"`
    Topic    string `yaml:"topic"`
}

type CoolpyConfig struct {
    APIKey   string `yaml:"api_key"`
    DeviceID string `yaml:"device_id"`
}

var config Config
var mqttClient MQTT.Client

// 加载配置文件
func loadConfig() error {
    file, err := os.Open("config.yaml")
    if err != nil {
        return err
    }
    defer file.Close()

    decoder := yaml.NewDecoder(file)
    return decoder.Decode(&config)
}

// 初始化 MQTT 客户端
func initMQTT() error {
    opts := MQTT.NewClientOptions()
    opts.AddBroker(config.MQTT.Broker)
    opts.SetClientID(config.MQTT.ClientID)
    opts.SetUsername(config.MQTT.Username)
    opts.SetPassword(config.MQTT.Password)
    // 设置连接超时时间
    opts.SetConnectTimeout(5 * time.Second)
    // 设置重连间隔
    opts.SetAutoReconnect(true)
    opts.SetMaxReconnectInterval(3 * time.Second)
    // 设置消息接收回调函数
    opts.SetDefaultPublishHandler(messageHandler)

    mqttClient = MQTT.NewClient(opts)
    if token := mqttClient.Connect(); token.Wait() && token.Error() != nil {
        return token.Error()
    }

    log.Println("MQTT 客户端连接成功")
    return nil
}

// 消息接收处理函数
func messageHandler(client MQTT.Client, msg MQTT.Message) {
    log.Printf("收到消息 - 主题: %s, 内容: %s", msg.Topic(), string(msg.Payload()))
    // 这里可添加消息转发到 Coolpy 的逻辑
    forwardToCoolpy(string(msg.Payload()))
}

// 转发消息到 Coolpy
func forwardToCoolpy(data string) {
    // 调用 Coolpy API 转发消息，这里以模拟为例，实际需替换为真实 API 调用
    log.Printf("转发消息到 Coolpy - 设备 ID: %s, 数据: %s", config.Coolpy.DeviceID, data)
    // 真实场景示例：
    // req, err := http.NewRequest("POST", "https://api.coolpy.com/device/data", strings.NewReader(data))
    // req.Header.Set("Authorization", "Bearer "+config.Coolpy.APIKey)
    // 后续处理请求...
}

// 发布消息到 MQTT 主题
func publishMessage(data string) error {
    token := mqttClient.Publish(config.MQTT.Topic, 1, false, data)
    token.Wait()
    return token.Error()
}

// 订阅 MQTT 主题
func subscribeTopic() error {
    token := mqttClient.Subscribe(config.MQTT.Topic, 1, nil)
    token.Wait()
    if token.Error() != nil {
        return token.Error()
    }
    log.Printf("订阅主题成功: %s", config.MQTT.Topic)
    return nil
}

func main() {
    // 加载配置
    if err := loadConfig(); err != nil {
        log.Fatalf("加载配置失败: %v", err)
    }

    // 初始化 MQTT
    if err := initMQTT(); err != nil {
        log.Fatalf("初始化 MQTT 失败: %v", err)
    }
    defer mqttClient.Disconnect(250)

    // 订阅主题
    if err := subscribeTopic(); err != nil {
        log.Fatalf("订阅主题失败: %v", err)
    }

    // 模拟发布一条测试消息
    testData := `{"temperature": 25.5, "humidity": 60}`
    if err := publishMessage(testData); err != nil {
        log.Fatalf("发布消息失败: %v", err)
    }
    log.Printf("发布测试消息成功: %s", testData)

    // 监听退出信号，优雅关闭服务
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
    <-sigChan
    log.Println("服务正在关闭...")
}

// 模拟 Coolpy API 调用工具函数（实际项目可根据需求扩展）
func coolpyAPICall(method, url, data string) error {
    // 此处实现真实的 Coolpy API 调用逻辑
    return nil
}