eemqtt

eemqtt 是一个基于 github.com/eclipse/paho.golang 的 ego MQTT 组件。

核心目标：

• 保留 paho.golang/autopaho 的 MQTT 连接、重连、路由和 MQTT v5 能力
• 用 ego 的方式接入配置、日志、指标、链路、health 和 debug stats
• 对外暴露适合 ego 微服务使用的 BuildE / Publish / Subscribe / Close 组件接口

当前能力

• Load / Build / BuildE 配置化构建
• 启动阶段首连校验，避免“服务已启动但 MQTT 实际不可用”
• Publish / PublishViaQueue / Subscribe / SubscribeMany / Unsubscribe / Close
• StandardRouter 路由和离线订阅注册表
• 自动重连后的订阅恢复
• TLS 文件配置加载
• Health / HealthStatus / Ping / Stats / LoadInstance
• /debug/mqtt/stats governor 调试入口
• MQTT v5 UserProperties 链路透传

核心设计

• 组件生命周期分为 Load -> BuildE -> Running -> Reconnecting/Degraded -> Close
• BuildE() 会在启动阶段等待首次连接结果，避免服务已经启动但 MQTT 实际不可用
• 订阅使用 StandardRouter + 本地订阅注册表，保证重连后可恢复订阅
• 可观测性通过 elog / emetric / etrace / governor 接入，不破坏底层 MQTT 能力
• PublishViaQueue 用于对接 autopaho 的离线队列发布能力

快速开始

package main

import (
	"context"
	"encoding/json"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/ego-component/eemqtt"
	"github.com/gotomicro/ego"
	"github.com/gotomicro/ego/core/elog"
)

var client *eemqtt.Component

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	if err := ego.New(
		ego.WithHang(true),
		ego.WithAfterStopClean(closeMQTT),
	).Invoker(initMQTT, runPublisher).Run(); err != nil {
		elog.Error("startup", elog.FieldErr(err))
		return
	}
	<-ctx.Done()
}

func initMQTT() error {
	var err error
	client, err = eemqtt.Load("emqtt").BuildE()
	if err != nil {
		return err
	}
	return client.Subscribe("topic/demo", handleMessage, eemqtt.WithSubscribeQoS(1))
}

func runPublisher() error {
	go func() {
		ticker := time.NewTicker(5 * time.Second)
		defer ticker.Stop()

		for i := 0; ; i++ {
			select {
			case <-ticker.C:
				payload, _ := json.Marshal(map[string]any{"count": i})
				if err := client.Publish(context.Background(), "topic/demo", payload, eemqtt.WithQoS(1)); err != nil {
					elog.Error("publish", elog.FieldErr(err))
				}
			}
		}
	}()
	return nil
}

func handleMessage(ctx context.Context, msg *eemqtt.Message) error {
	elog.Info("receive mqtt message", elog.String("topic", msg.Topic), elog.ByteString("payload", msg.Payload))
	return nil
}

func closeMQTT() error {
	if client == nil {
		return nil
	}
	return client.Close()
}

配置示例

[emqtt]
debug = true
brokers = ["mqtt://127.0.0.1:1883"]
username = ""
password = ""
clientID = "demo-client"

keepAlive = 30
connectTimeout = "10s"
startupTimeout = "10s"
connectRetryDelay = "5s"
shutdownTimeout = "5s"

cleanStartOnInitialConnection = false
sessionExpiryInterval = 3600

enableAccessInterceptor = true
enableMetricInterceptor = true
enableTraceInterceptor = true
onFail = "panic"

[emqtt.authentication.tls]
caFile = "./certs/ca.crt"
certFile = "./certs/client.crt"
keyFile = "./certs/client.key"
serverName = "broker.example.com"
insecureSkipVerify = false

说明：

• brokers 是主入口；broker 和 addr 只保留为单 broker 兼容字段
• clientID 在同一个 broker 上必须保持唯一；如果两个进程复用同一个 clientID，Broker 可能会踢掉旧连接
• TLS 使用 [emqtt.authentication.tls]
• BuildE() 会在 startupTimeout 内等待首次连接成功
• cleanStartOnInitialConnection = false 表示默认保留 MQTT session；如果你希望每次启动都丢弃旧 session，请显式改成 true
• Publish 的 payload 只接受 []byte
• Publish 在断连期间会快速返回错误；如果你需要断连时继续接收发布请求并等待恢复后补发，请使用 PublishViaQueue
• 纯 client 进程示例建议配合 ego.WithHang(true)，否则 Invoker 执行完后进程会直接退出
• 推荐通过 ego.WithAfterStopClean 把 Close() 接到 ego 停机流程

API

func DefaultContainer() *Container
func Load(key string) *Container
func LoadInstance(name string) *Component

func (c *Container) Build(options ...Option) *Component
func (c *Container) BuildE(options ...Option) (*Component, error)

func (c *Component) Publish(ctx context.Context, topic string, payload []byte, opts ...PublishOption) error
func (c *Component) PublishViaQueue(ctx context.Context, topic string, payload []byte, opts ...PublishOption) error
func (c *Component) Subscribe(topic string, handler MessageHandler, opts ...SubscribeOption) error
func (c *Component) SubscribeMany(subs ...Subscription) error
func (c *Component) Unsubscribe(topics ...string) error

func (c *Component) Close() error
func (c *Component) Health() bool
func (c *Component) HealthStatus() HealthInfo
func (c *Component) Ping(ctx context.Context) error
func (c *Component) Stats() Stats

func (c *Component) Manager() *autopaho.ConnectionManager
func (c *Component) Router() *paho.StandardRouter

常用选项：

• WithOnFail("panic"|"error")
• WithTLSConfig(*tls.Config)
• WithConnectPacketBuilder(...)
• WithPahoPublishHook(...)
• WithQueue(...)
• WithOnConnectionUp(...)
• WithOnConnectionDown(...)
• WithOnConnectError(...)
• WithQoS(byte)
• WithRetain(bool)
• WithContentType(string)
• WithUserProperty(key, value)
• WithSubscribeQoS(byte)

示例

• 标准模板: examples/basic_pubsub/main.go
• 用户名密码连接: examples/user_auth/main.go
• 生命周期与 governor: examples/lifecycle_and_governor/main.go
• 高级透传能力: examples/advanced_options/main.go
• TLS 连接: examples/tls_auth/main.go
• 队列发布: examples/queue_publish/main.go

与 paho.golang 的差距和同步策略

eemqtt 不是 github.com/eclipse/paho.golang 的 1:1 镜像封装，而是一个面向 ego 微服务底座的 MQTT 组件。

当前已经对齐的方向：

• 保留 autopaho.ConnectionManager 作为连接核心
• 保留 StandardRouter、MQTT v5 UserProperties、Queue、连接回调、ConnectPacketBuilder、PublishHook 这些关键扩展点
• 用 ego 生命周期、日志、指标、链路、governor stats 把这些能力接成可治理组件

当前仍然存在的差距：

• 不是所有 paho.golang/autopaho 的底层能力都已经做成 eemqtt 的高层 helper 或示例
• 一些更底层的模型，例如更细粒度的 session/store 组合、特殊 transport、自定义收包链路、RPC 风格封装，当前仍然建议通过 typed passthrough 或直接使用底层包能力
• SubscribeTopics 目前主要承担配置化订阅注册表和重连恢复初始化，不负责在配置层直接绑定业务 handler

用户侧扩展路径：

• eemqtt 虽然不是 paho.golang 的完整镜像，但没有把高级能力封死
• 已开放的 Manager()、Router()、WithConnectPacketBuilder(...)、WithPahoPublishHook(...)、WithQueue(...)、连接生命周期回调，就是面向高级用户的原生扩展入口
• 如果业务需要更底层的 MQTT 能力，建议优先基于这些入口扩展，而不是绕过 eemqtt 自己再维护一套独立生命周期
• 这样既能继续复用 ego 的配置、日志、指标、trace、governor、health/stats，又不会丢失对上游工具包高级能力的接入空间

后续兼容方向：

• 会持续跟进 paho.golang 官方工具包的版本和稳定能力演进
• 对通用且适合 ego 组件边界的能力，优先补 typed passthrough，再视使用频率决定是否补高层 helper
• 对不适合被高层抽象稳定承诺的底层细节，继续保留直接访问 Manager() / Router() 的能力，避免为了“统一封装”损失原生能力
• README、示例和集成测试会持续同步官方工具包的关键能力，确保组件演进时不会和上游长期脱节

调试与验证

• 组件实例 stats: /debug/mqtt/stats
• 全量测试: go test ./...
• 集成测试需要显式设置以下环境变量后再运行：

export EEMQTT_INTEGRATION=1
export EEMQTT_INTEGRATION_BROKER='tcp://localhost:1883'
export EEMQTT_INTEGRATION_USERNAME='your-username'
export EEMQTT_INTEGRATION_PASSWORD='your-password'
export EEMQTT_INTEGRATION_CLIENT_ID_PREFIX='internal-eemqtt-integration'
go test . -run 'TestIntegrationEMQXPublishSubscribe|TestIntegrationEMQXUnsubscribeStopsDelivery|TestIntegrationEMQXReconnectResubscribes' -v