heqiapp/platforms
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: embed mochi mqtt broker

Browse Source
This commit is contained in:
yanweidong
2026-08-03 14:19:28 +08:00
parent fefdba470a
commit 5edca1524f
13 changed files with 148 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
backend/iot-server/internal/service/service.go
View File

@@ -1,4 +1,4 @@
// Package service 连接外部 MQTT Broker，并在内部 HTTP 边界接收待下发命令。
// Package service 内嵌 MQTT Broker，并在内部 HTTP 边界接收待下发命令。
package service
import (
@@ -19,15 +19,19 @@ import (
"git.apinb.com/heqiapp/platforms/backend/iot-server/internal/config"
"git.apinb.com/heqiapp/platforms/backend/iot-server/internal/protocol"
mqtt "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"
"github.com/mochi-mqtt/server/v2/packets"
)
type Service struct {
cfg config.Config
keys protocol.Keyring
mqtt mqtt.Client
broker *mqtt.Server
packet atomic.Uint32
http *http.Server
ready atomic.Bool
}
type Command struct {
Identity string `json:"identity"`
@@ -63,16 +67,37 @@ func New(cfg config.Config) (*Service, error) {
if err != nil {
return nil, err
}
options := mqtt.NewClientOptions().AddBroker(cfg.MQTT.Broker).SetClientID(cfg.MQTT.ClientID).SetUsername(cfg.MQTT.Username).SetPassword(cfg.MQTT.Password).SetAutoReconnect(true).SetConnectRetry(true)
broker := mqtt.New(&mqtt.Options{InlineClient: true})
ledger := &auth.Ledger{}
for _, device := range cfg.MQTT.Devices {
ledger.Auth = append(ledger.Auth, auth.AuthRule{
Client: auth.RString(device.ClientID), Username: auth.RString(device.Username),
Password: auth.RString(device.Password), Allow: true,
})
ledger.ACL = append(ledger.ACL, auth.ACLRule{
Client: auth.RString(device.ClientID), Username: auth.RString(device.Username),
Filters: auth.Filters{
auth.RString(strings.ReplaceAll(cfg.MQTT.UpTopic, "+", device.DeviceID)): auth.WriteOnly,
auth.RString(strings.ReplaceAll(cfg.MQTT.AckTopic, "+", device.DeviceID)): auth.WriteOnly,
auth.RString(strings.ReplaceAll(cfg.MQTT.DownTopic, "{deviceId}", device.DeviceID)): auth.ReadOnly,
},
})
}
if err = broker.AddHook(new(auth.Hook), &auth.Options{Ledger: ledger}); err != nil {
return nil, fmt.Errorf("配置 MQTT 认证: %w", err)
}
listenerConfig := listeners.Config{ID: cfg.MQTT.ListenerID, Address: cfg.MQTT.Address}
if cfg.MQTT.TLS {
tlsConfig, tlsErr := makeTLSConfig(cfg)
if tlsErr != nil {
return nil, tlsErr
}
options.SetTLSConfig(tlsConfig)
listenerConfig.TLSConfig = tlsConfig
}
client := mqtt.NewClient(options)
return &Service{cfg: cfg, keys: keys, mqtt: client}, nil
if err = broker.AddListener(listeners.NewTCP(listenerConfig)); err != nil {
return nil, fmt.Errorf("配置 MQTT 监听器: %w", err)
}
return &Service{cfg: cfg, keys: keys, broker: broker}, nil
}
func makeTLSConfig(cfg config.Config) (*tls.Config, error) {
@@ -89,28 +114,32 @@ func makeTLSConfig(cfg config.Config) (*tls.Config, error) {
return nil, fmt.Errorf("MQTT CA 证书无效")
}
}
result := &tls.Config{MinVersion: tls.VersionTLS12, RootCAs: roots}
if cfg.MQTT.CertificateFile != "" || cfg.MQTT.PrivateKeyFile != "" {
certificate, loadErr := tls.LoadX509KeyPair(cfg.MQTT.CertificateFile, cfg.MQTT.PrivateKeyFile)
if loadErr != nil {
return nil, loadErr
}
result.Certificates = []tls.Certificate{certificate}
if cfg.MQTT.CertificateFile == "" || cfg.MQTT.PrivateKeyFile == "" {
return nil, fmt.Errorf("启用 MQTT TLS 时 CertificateFile 和 PrivateKeyFile 必填")
}
certificate, err := tls.LoadX509KeyPair(cfg.MQTT.CertificateFile, cfg.MQTT.PrivateKeyFile)
if err != nil {
return nil, err
}
result := &tls.Config{MinVersion: tls.VersionTLS12, Certificates: []tls.Certificate{certificate}}
if cfg.MQTT.CAFile != "" {
result.ClientCAs = roots
result.ClientAuth = tls.RequireAndVerifyClientCert
}
return result, nil
}
func (s *Service) Run(ctx context.Context) error {
if token := s.mqtt.Connect(); !token.WaitTimeout(15 * time.Second) {
return fmt.Errorf("MQTT 连接超时")
} else if token.Error() != nil {
return token.Error()
if err := s.broker.Subscribe(s.cfg.MQTT.UpTopic, 1, s.onMessage); err != nil {
return fmt.Errorf("订阅设备上行 Topic: %w", err)
}
for _, topic := range []string{s.cfg.MQTT.UpTopic, s.cfg.MQTT.AckTopic} {
if token := s.mqtt.Subscribe(topic, s.cfg.MQTT.QoS, s.onMessage); token.Wait() && token.Error() != nil {
return token.Error()
}
if err := s.broker.Subscribe(s.cfg.MQTT.AckTopic, 2, s.onMessage); err != nil {
return fmt.Errorf("订阅设备回执 Topic: %w", err)
}
if err := s.broker.Serve(); err != nil {
return fmt.Errorf("启动 MQTT Broker: %w", err)
}
s.ready.Store(true)
mux := http.NewServeMux()
mux.HandleFunc("/health", s.health)
mux.HandleFunc("/internal/v1/commands", s.command)
@@ -119,8 +148,9 @@ func (s *Service) Run(ctx context.Context) error {
<-ctx.Done()
shutdown, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
s.ready.Store(false)
_ = s.http.Shutdown(shutdown)
s.mqtt.Disconnect(250)
_ = s.broker.Close()
}()
err := s.http.ListenAndServe()
if err == http.ErrServerClosed {
@@ -130,8 +160,8 @@ func (s *Service) Run(ctx context.Context) error {
}
func (s *Service) health(w http.ResponseWriter, _ *http.Request) {
if !s.mqtt.IsConnectionOpen() {
http.Error(w, "mqtt disconnected", http.StatusServiceUnavailable)
if !s.ready.Load() {
http.Error(w, "mqtt broker unavailable", http.StatusServiceUnavailable)
return
}
w.WriteHeader(http.StatusNoContent)
@@ -181,8 +211,7 @@ func (s *Service) command(w http.ResponseWriter, r *http.Request) {
return
}
topic := strings.ReplaceAll(s.cfg.MQTT.DownTopic, "{deviceId}", cmd.DeviceID)
token := s.mqtt.Publish(topic, s.cfg.MQTT.QoS, false, raw)
if !token.WaitTimeout(10*time.Second) || token.Error() != nil {
if err := s.broker.Publish(topic, raw, false, s.cfg.MQTT.QoS); err != nil {
http.Error(w, "mqtt publish failed", http.StatusBadGateway)
return
}
@@ -191,9 +220,9 @@ func (s *Service) command(w http.ResponseWriter, r *http.Request) {
_ = json.NewEncoder(w).Encode(map[string]any{"command_identity": cmd.Identity, "packet_number": frame.PacketNumber, "status": "dispatched"})
}
func (s *Service) onMessage(_ mqtt.Client, message mqtt.Message) {
raw := append([]byte(nil), message.Payload()...)
envelope := Envelope{Type: "device_message", Topic: message.Topic(), ReceivedAt: time.Now().UTC().Format(time.RFC3339Nano), PayloadHex: hex.EncodeToString(raw)}
func (s *Service) onMessage(_ *mqtt.Client, _ packets.Subscription, message packets.Packet) {
raw := append([]byte(nil), message.Payload...)
envelope := Envelope{Type: "device_message", Topic: message.TopicName, ReceivedAt: time.Now().UTC().Format(time.RFC3339Nano), PayloadHex: hex.EncodeToString(raw)}
if frame, err := protocol.Decode(raw, s.keys); err == nil {
deviceID := hex.EncodeToString(frame.DeviceID[:])
envelope.DeviceID = deviceID