mochi-mqtt/server 实现一个mqtt bridge 功能

mochi-mqtt/server内置了一个比较有意思的功能InlineClient 通过此接口我们就可以实现一个类似nanomq 的mqtt bridge 功能，而且开发起来比较方便

参考实现机制

简单说明: 就是通过一个mqtt client 接受mqtt broker 的消息，结合配置的信息，讲数据发给mochi-mqtt/server 的client 或者给mqtt broker

参考代码

就是一个简单示例代码，代码比较简单，主要是一个简单的演示，讲mqtt broker 的消息转发给mochi-mqtt/server client

package mainimport ("fmt""log""os""os/signal""syscall""time"mqttclient "github.com/eclipse/paho.mqtt.golang"mqtt "github.com/mochi-mqtt/server/v2""github.com/mochi-mqtt/server/v2/hooks/auth""github.com/mochi-mqtt/server/v2/listeners"
)func main() {sigs := make(chan os.Signal, 1)done := make(chan bool, 1)signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)go func() {<-sigsdone <- true}()server := mqtt.New(&mqtt.Options{InlineClient: true, // 此参数是核心，默认没开启，同时注意此参数应该在mqtt 初始化的时候赋值，其他地方不行，目前会有问题})_ = server.AddHook(new(auth.AllowHook), nil)tcp := listeners.NewTCP(listeners.Config{ID: "t1", Address: ":1889"})err := server.AddListener(tcp)if err != nil {log.Fatal(err)}go func() {if err := server.Serve(); err != nil {log.Fatalf("[服务端启动失败] %v", err)}}()go func() {messagePubHandler := func(client mqttclient.Client, msg mqttclient.Message) {fmt.Printf("收到消息: [%s] %s\n", msg.Topic(), msg.Payload())server.Publish("dalongdemo", msg.Payload(), false, 0)}topicMap := map[string]byte{"topic/one":   0,"topic/two":   1,"topic/three": 2,}opts := mqttclient.NewClientOptions()opts.AddBroker("tcp://localhost:1886")opts.SetClientID("go_mqtt_client")opts.SetAutoReconnect(true)opts.SetDefaultPublishHandler(messagePubHandler)opts.SetConnectRetry(true)opts.SetKeepAlive(10 * time.Second)opts.SetConnectionLostHandler(func(client mqttclient.Client, err error) {fmt.Println("❌ 连接丢失:", err)})// 重连成功的回调messageHandler := func(client mqttclient.Client, msg mqttclient.Message) {defer func() {if r := recover(); r != nil {fmt.Printf("[消息处理异常] %v\n", r)}}()fmt.Printf("Received message on topic: %s\n", msg.Topic())fmt.Printf("Payload: %s\n", msg.Payload())// 使用InlineClient 进行发布，实际上就是对于mqtt broker 的消息，转发给mochi-mqtt/server 的clientif err := server.Publish("dalongdemo", msg.Payload(), false, 0); err != nil {fmt.Printf("Error publishing message: %v\n", err)}}opts.SetOnConnectHandler(func(client mqttclient.Client) {fmt.Println("✅ 重连成功，重新订阅主题")// 断线后需要手动恢复订阅if token := client.SubscribeMultiple(topicMap, messageHandler); token.Wait() && token.Error() != nil {fmt.Println("订阅失败:", token.Error())}})client := mqttclient.NewClient(opts)token := client.Connect()token.Wait()if err := token.Error(); err != nil {log.Printf("[连接MQTT服务器失败] %v", err)}if token := client.SubscribeMultiple(topicMap, messageHandler); token.Wait() {if err := token.Error(); err != nil {log.Printf("[订阅主题失败] %v", err)return}}log.Printf("📡 已订阅: %v", topicMap)}()<-done
}

说明

以上的实现比较简单，是一个单向的，如果需要讲mochi-mqtt/server 消息转发mqtt broker 的实现是类似的

参考资料

https://docs.nats.io/running-a-nats-service/configuration/mqtt

https://github.com/mochi-mqtt/server

https://github.com/nanomq/nanomq