重构：拆出 auth / blocking / runtimesettings / help / mqttforward 包
第二批：把根目录中纯逻辑领域文件（cache、service、admin route）按业务边界
迁到 internal/ 下的子包。各子包暴露 RegisterRoutes 给 web 包调用，根目录
只留下一行 bridge 文件保留旧名字别名。
新增包
- internal/auth/ SessionClaims / Manager / RequireAdmin / HashPassword /
VerifyPassword / AdminUserResponse 等。原 auth.go 中
被两个 admin route 依赖的 sessionClaims 现在以 auth.
SessionClaims 形式被它们 import；不再被锁在 main 包。
- internal/blocking/ Cache + RegisterRoutes，以前散在 blocking_cache.go
和 admin_blocking_routes.go 里。
- internal/runtimesettings/ Cache + RegisterRoutes。
- internal/help/ RenderMarkdown / RegisterPublicRoutes /
RegisterAdminRoutes（拆分原来的 registerHelpRoutes
和 registerAdminHelpRoutes 两条入口）。
- internal/mqttforward/ Manager / Reloader / Stats / RegisterRoutes。
forwarder runner、循环抑制 cache 等运行时逻辑随之迁入。
- internal/webutil/ ParseListOptions / WriteListResponse[WithTotal] /
ParseMapReportListOptions / ParseMapReportViewportOptions
以及 PtrString/PtrInt64/... 等指针解引用 helper。
以前散在 web.go 中，现在被各 admin route 子包共享，
避免 internal/blocking → internal/web → internal/blocking
的循环依赖。
- internal/store/testutil/ OpenStore(t) helper，让其它包测试零样板拿到 store。
根目录新增 bridge 文件
- blocking_bridge.go / runtime_settings_bridge.go / help_bridge.go /
mqttforward_bridge.go：用 type alias + thin wrapper 把上述子包的导出
名映射到旧的小写名（blockingCache、registerAdminBlockingRoutes 等），
让 main.go / web.go 等仍未迁出的文件无须改动。
修改
- auth.go 改为对 internal/auth 的 bridge；web.go 中 sessions.newCookie /
clearCookie 改为 NewCookie / ClearCookie。
- main_test.go 中 BlockingViolationForRecord* 测试不再直接构造未导出字段，
改成走 store.CreateNodeBlocking → newBlockingCache 的真实路径。
- internal/mqttforward 把以前 *_store.go 中没有方法依赖的运行时类型
（forwarder runner、loop cache）和 admin route 一并归位；mqtt_status.go
暂时仍留在根目录（依赖 main 中的 mqttClientInfoFromClient）。
go build ./... / go test ./... 全部通过；测试数量未变。
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
@@ -0,0 +1,284 @@
|
||||
package mqttforward
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"gorm.io/gorm"
|
||||
|
||||
storepkg "meshtastic_mqtt_server/internal/store"
|
||||
"meshtastic_mqtt_server/internal/webutil"
|
||||
)
|
||||
|
||||
type mqttForwarderRequest struct {
|
||||
Name string `json:"name"`
|
||||
Enabled bool `json:"enabled"`
|
||||
SourceHost string `json:"source_host"`
|
||||
SourcePort int `json:"source_port"`
|
||||
SourceUsername string `json:"source_username"`
|
||||
SourcePassword *string `json:"source_password"`
|
||||
SourcePasswordClear bool `json:"source_password_clear"`
|
||||
SourceClientID string `json:"source_client_id"`
|
||||
SourceTLS bool `json:"source_tls"`
|
||||
TargetHost string `json:"target_host"`
|
||||
TargetPort int `json:"target_port"`
|
||||
TargetUsername string `json:"target_username"`
|
||||
TargetPassword *string `json:"target_password"`
|
||||
TargetPasswordClear bool `json:"target_password_clear"`
|
||||
TargetClientID string `json:"target_client_id"`
|
||||
TargetTLS bool `json:"target_tls"`
|
||||
}
|
||||
|
||||
type mqttForwardTopicRequest struct {
|
||||
Topic string `json:"topic"`
|
||||
Enabled bool `json:"enabled"`
|
||||
Direction string `json:"direction"`
|
||||
SourcePrefix string `json:"source_prefix"`
|
||||
TargetPrefix string `json:"target_prefix"`
|
||||
QoS int `json:"qos"`
|
||||
Retain bool `json:"retain"`
|
||||
}
|
||||
|
||||
func RegisterRoutes(r gin.IRouter, store *storepkg.Store, forwarder Reloader) {
|
||||
r.GET("/mqtt-forward/forwarders", func(c *gin.Context) {
|
||||
opts, ok := webutil.ParseListOptions(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
rows, err := store.ListMQTTForwarders(opts)
|
||||
if err != nil {
|
||||
webutil.WriteListResponse(c, rows, opts, err, mqttForwarderDTO)
|
||||
return
|
||||
}
|
||||
total, err := store.CountMQTTForwarders(opts)
|
||||
webutil.WriteListResponseWithTotal(c, rows, opts, total, err, mqttForwarderDTO)
|
||||
})
|
||||
r.POST("/mqtt-forward/forwarders", func(c *gin.Context) {
|
||||
var req mqttForwarderRequest
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid mqtt forwarder request"})
|
||||
return
|
||||
}
|
||||
input := mqttForwarderInputFromRequest(req)
|
||||
row, err := store.CreateMQTTForwarder(input)
|
||||
writeMQTTForwardMutationResponse(c, http.StatusCreated, row, err, func() error {
|
||||
return reloadMQTTForwarder(forwarder, row.ID)
|
||||
})
|
||||
})
|
||||
r.PUT("/mqtt-forward/forwarders/:id", func(c *gin.Context) {
|
||||
id, ok := parseMQTTForwardID(c, "invalid mqtt forwarder id")
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var req mqttForwarderRequest
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid mqtt forwarder request"})
|
||||
return
|
||||
}
|
||||
input := mqttForwarderInputFromRequest(req)
|
||||
row, err := store.UpdateMQTTForwarder(id, input)
|
||||
writeMQTTForwardMutationResponse(c, http.StatusOK, row, err, func() error {
|
||||
return reloadMQTTForwarder(forwarder, id)
|
||||
})
|
||||
})
|
||||
r.DELETE("/mqtt-forward/forwarders/:id", func(c *gin.Context) {
|
||||
id, ok := parseMQTTForwardID(c, "invalid mqtt forwarder id")
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if forwarder != nil {
|
||||
forwarder.StopForwarder(id)
|
||||
}
|
||||
writeMQTTForwardDeleteResponse(c, store.DeleteMQTTForwarder(id), nil)
|
||||
})
|
||||
r.POST("/mqtt-forward/forwarders/:id/restart", func(c *gin.Context) {
|
||||
id, ok := parseMQTTForwardID(c, "invalid mqtt forwarder id")
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if err := reloadMQTTForwarder(forwarder, id); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"status": "ok"})
|
||||
})
|
||||
r.GET("/mqtt-forward/forwarders/:id/topics", func(c *gin.Context) {
|
||||
id, ok := parseMQTTForwardID(c, "invalid mqtt forwarder id")
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
opts, ok := webutil.ParseListOptions(c)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
rows, err := store.ListMQTTForwardTopics(id, opts)
|
||||
if err != nil {
|
||||
webutil.WriteListResponse(c, rows, opts, err, mqttForwardTopicDTO)
|
||||
return
|
||||
}
|
||||
total, err := store.CountMQTTForwardTopics(id)
|
||||
webutil.WriteListResponseWithTotal(c, rows, opts, total, err, mqttForwardTopicDTO)
|
||||
})
|
||||
r.POST("/mqtt-forward/forwarders/:id/topics", func(c *gin.Context) {
|
||||
id, ok := parseMQTTForwardID(c, "invalid mqtt forwarder id")
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var req mqttForwardTopicRequest
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid mqtt forward topic request"})
|
||||
return
|
||||
}
|
||||
row, err := store.CreateMQTTForwardTopic(id, mqttForwardTopicInputFromRequest(req))
|
||||
writeMQTTForwardTopicMutationResponse(c, http.StatusCreated, row, err, func() error {
|
||||
return reloadMQTTForwarder(forwarder, id)
|
||||
})
|
||||
})
|
||||
r.PUT("/mqtt-forward/topics/:id", func(c *gin.Context) {
|
||||
id, ok := parseMQTTForwardID(c, "invalid mqtt forward topic id")
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var req mqttForwardTopicRequest
|
||||
if err := c.ShouldBindJSON(&req); err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "invalid mqtt forward topic request"})
|
||||
return
|
||||
}
|
||||
row, err := store.UpdateMQTTForwardTopic(id, mqttForwardTopicInputFromRequest(req))
|
||||
writeMQTTForwardTopicMutationResponse(c, http.StatusOK, row, err, func() error {
|
||||
return reloadMQTTForwarder(forwarder, row.ForwarderID)
|
||||
})
|
||||
})
|
||||
r.DELETE("/mqtt-forward/topics/:id", func(c *gin.Context) {
|
||||
id, ok := parseMQTTForwardID(c, "invalid mqtt forward topic id")
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
row, err := store.GetMQTTForwardTopic(id)
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "mqtt forward topic not found"})
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
parentID := row.ForwarderID
|
||||
writeMQTTForwardDeleteResponse(c, store.DeleteMQTTForwardTopic(id), func() error {
|
||||
return reloadMQTTForwarder(forwarder, parentID)
|
||||
})
|
||||
})
|
||||
r.GET("/mqtt-forward/status", func(c *gin.Context) {
|
||||
items := []RuntimeStatus{}
|
||||
if forwarder != nil {
|
||||
items = forwarder.Status()
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"items": items})
|
||||
})
|
||||
}
|
||||
|
||||
func mqttForwarderInputFromRequest(req mqttForwarderRequest) storepkg.MQTTForwarderInput {
|
||||
sourcePassword := req.SourcePassword
|
||||
if req.SourcePasswordClear {
|
||||
empty := ""
|
||||
sourcePassword = &empty
|
||||
}
|
||||
targetPassword := req.TargetPassword
|
||||
if req.TargetPasswordClear {
|
||||
empty := ""
|
||||
targetPassword = &empty
|
||||
}
|
||||
return storepkg.MQTTForwarderInput{Name: req.Name, Enabled: req.Enabled, SourceHost: req.SourceHost, SourcePort: req.SourcePort, SourceUsername: req.SourceUsername, SourcePassword: sourcePassword, SourceClientID: req.SourceClientID, SourceTLS: req.SourceTLS, TargetHost: req.TargetHost, TargetPort: req.TargetPort, TargetUsername: req.TargetUsername, TargetPassword: targetPassword, TargetClientID: req.TargetClientID, TargetTLS: req.TargetTLS}
|
||||
}
|
||||
|
||||
func mqttForwardTopicInputFromRequest(req mqttForwardTopicRequest) storepkg.MQTTForwardTopicInput {
|
||||
return storepkg.MQTTForwardTopicInput{Topic: req.Topic, Enabled: req.Enabled, Direction: req.Direction, SourcePrefix: req.SourcePrefix, TargetPrefix: req.TargetPrefix, QoS: req.QoS, Retain: req.Retain}
|
||||
}
|
||||
|
||||
func parseMQTTForwardID(c *gin.Context, message string) (uint64, bool) {
|
||||
id, err := strconv.ParseUint(c.Param("id"), 10, 64)
|
||||
if err != nil || id == 0 {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": message})
|
||||
return 0, false
|
||||
}
|
||||
return id, true
|
||||
}
|
||||
|
||||
func reloadMQTTForwarder(forwarder Reloader, id uint64) error {
|
||||
if forwarder == nil {
|
||||
return nil
|
||||
}
|
||||
return forwarder.ReloadForwarder(id)
|
||||
}
|
||||
|
||||
func writeMQTTForwardMutationResponse(c *gin.Context, status int, row *storepkg.MQTTForwarderRecord, err error, afterSuccess func() error) {
|
||||
if errors.Is(err, storepkg.ErrMQTTForwarderAlreadyExists) {
|
||||
c.JSON(http.StatusConflict, gin.H{"error": "mqtt forwarder already exists"})
|
||||
return
|
||||
}
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "mqtt forwarder not found"})
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
if afterSuccess != nil {
|
||||
if err := afterSuccess(); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "mqtt forwarder saved but reload failed: " + err.Error()})
|
||||
return
|
||||
}
|
||||
}
|
||||
c.JSON(status, gin.H{"item": mqttForwarderDTO(*row)})
|
||||
}
|
||||
|
||||
func writeMQTTForwardTopicMutationResponse(c *gin.Context, status int, row *storepkg.MQTTForwardTopicRecord, err error, afterSuccess func() error) {
|
||||
if errors.Is(err, storepkg.ErrMQTTForwardTopicAlreadyExists) {
|
||||
c.JSON(http.StatusConflict, gin.H{"error": "mqtt forward topic already exists"})
|
||||
return
|
||||
}
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "mqtt forward topic not found"})
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
if afterSuccess != nil {
|
||||
if err := afterSuccess(); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "mqtt forward topic saved but reload failed: " + err.Error()})
|
||||
return
|
||||
}
|
||||
}
|
||||
c.JSON(status, gin.H{"item": mqttForwardTopicDTO(*row)})
|
||||
}
|
||||
|
||||
func writeMQTTForwardDeleteResponse(c *gin.Context, err error, afterSuccess func() error) {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": "mqtt forward item not found"})
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
|
||||
return
|
||||
}
|
||||
if afterSuccess != nil {
|
||||
if err := afterSuccess(); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "mqtt forward item deleted but reload failed: " + err.Error()})
|
||||
return
|
||||
}
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"status": "ok"})
|
||||
}
|
||||
|
||||
func mqttForwarderDTO(row storepkg.MQTTForwarderRecord) gin.H {
|
||||
return gin.H{"id": row.ID, "name": row.Name, "enabled": row.Enabled, "source_host": row.SourceHost, "source_port": row.SourcePort, "source_username": row.SourceUsername, "source_password_set": row.SourcePassword != "", "source_client_id": row.SourceClientID, "source_tls": row.SourceTLS, "target_host": row.TargetHost, "target_port": row.TargetPort, "target_username": row.TargetUsername, "target_password_set": row.TargetPassword != "", "target_client_id": row.TargetClientID, "target_tls": row.TargetTLS, "created_at": row.CreatedAt, "updated_at": row.UpdatedAt}
|
||||
}
|
||||
|
||||
func mqttForwardTopicDTO(row storepkg.MQTTForwardTopicRecord) gin.H {
|
||||
return gin.H{"id": row.ID, "forwarder_id": row.ForwarderID, "topic": row.Topic, "enabled": row.Enabled, "direction": row.Direction, "source_prefix": row.SourcePrefix, "target_prefix": row.TargetPrefix, "qos": row.QoS, "retain": row.Retain, "created_at": row.CreatedAt, "updated_at": row.UpdatedAt}
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
package mqttforward
|
||||
|
||||
import "sync/atomic"
|
||||
|
||||
type Stats struct {
|
||||
forwarded atomic.Int64
|
||||
dropped atomic.Int64
|
||||
}
|
||||
|
||||
func (s *Stats) IncForwarded() {
|
||||
if s != nil {
|
||||
s.forwarded.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Stats) IncDropped() {
|
||||
if s != nil {
|
||||
s.dropped.Add(1)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Stats) Forwarded() int64 {
|
||||
if s == nil {
|
||||
return 0
|
||||
}
|
||||
return s.forwarded.Load()
|
||||
}
|
||||
|
||||
func (s *Stats) Dropped() int64 {
|
||||
if s == nil {
|
||||
return 0
|
||||
}
|
||||
return s.dropped.Load()
|
||||
}
|
||||
@@ -0,0 +1,409 @@
|
||||
package mqttforward
|
||||
|
||||
import (
|
||||
storepkg "meshtastic_mqtt_server/internal/store"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"crypto/tls"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
pahomqtt "github.com/eclipse/paho.mqtt.golang"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
mqttForwardDirectionTargetToSource = "target_to_source"
|
||||
mqttForwardLoopTTL = 15 * time.Second
|
||||
mqttForwardLoopMaxEntries = 10000
|
||||
)
|
||||
|
||||
type Reloader interface {
|
||||
ReloadForwarder(id uint64) error
|
||||
StopForwarder(id uint64)
|
||||
Status() []RuntimeStatus
|
||||
}
|
||||
|
||||
type Manager struct {
|
||||
store *storepkg.Store
|
||||
mu sync.Mutex
|
||||
runners map[uint64]*runner
|
||||
}
|
||||
|
||||
type RuntimeStatus struct {
|
||||
ForwarderID uint64 `json:"forwarder_id"`
|
||||
Running bool `json:"running"`
|
||||
SourceConnected bool `json:"source_connected"`
|
||||
TargetConnected bool `json:"target_connected"`
|
||||
LastError string `json:"last_error"`
|
||||
StartedAt *time.Time `json:"started_at"`
|
||||
MessagesForwarded uint64 `json:"messages_forwarded"`
|
||||
MessagesDropped uint64 `json:"messages_dropped"`
|
||||
}
|
||||
|
||||
type runner struct {
|
||||
config storepkg.MQTTForwarderConfig
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
source pahomqtt.Client
|
||||
target pahomqtt.Client
|
||||
|
||||
mu sync.Mutex
|
||||
lastError string
|
||||
startedAt time.Time
|
||||
sourceConnected bool
|
||||
targetConnected bool
|
||||
messagesForwarded uint64
|
||||
messagesDropped uint64
|
||||
loopCache map[string]time.Time
|
||||
}
|
||||
|
||||
func NewManager(store *storepkg.Store) *Manager {
|
||||
return &Manager{store: store, runners: make(map[uint64]*runner)}
|
||||
}
|
||||
|
||||
func (m *Manager) StartFromStore() error {
|
||||
configs, err := m.store.ListEnabledMQTTForwarderConfigs()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, cfg := range configs {
|
||||
if len(cfg.Topics) == 0 {
|
||||
continue
|
||||
}
|
||||
runner := newRunner(cfg)
|
||||
runner.Start()
|
||||
m.mu.Lock()
|
||||
m.runners[cfg.Forwarder.ID] = runner
|
||||
m.mu.Unlock()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) ReloadForwarder(id uint64) error {
|
||||
m.StopForwarder(id)
|
||||
cfg, err := m.store.GetMQTTForwarderConfig(id)
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !cfg.Forwarder.Enabled || len(cfg.Topics) == 0 {
|
||||
return nil
|
||||
}
|
||||
runner := newRunner(*cfg)
|
||||
runner.Start()
|
||||
m.mu.Lock()
|
||||
m.runners[id] = runner
|
||||
m.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) StopForwarder(id uint64) {
|
||||
m.mu.Lock()
|
||||
runner := m.runners[id]
|
||||
delete(m.runners, id)
|
||||
m.mu.Unlock()
|
||||
if runner != nil {
|
||||
runner.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) StopAll() {
|
||||
m.mu.Lock()
|
||||
runners := make([]*runner, 0, len(m.runners))
|
||||
for id, runner := range m.runners {
|
||||
runners = append(runners, runner)
|
||||
delete(m.runners, id)
|
||||
}
|
||||
m.mu.Unlock()
|
||||
for _, runner := range runners {
|
||||
runner.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) Status() []RuntimeStatus {
|
||||
m.mu.Lock()
|
||||
runners := make([]*runner, 0, len(m.runners))
|
||||
for _, runner := range m.runners {
|
||||
runners = append(runners, runner)
|
||||
}
|
||||
m.mu.Unlock()
|
||||
items := make([]RuntimeStatus, 0, len(runners))
|
||||
for _, runner := range runners {
|
||||
items = append(items, runner.Status())
|
||||
}
|
||||
sort.Slice(items, func(i, j int) bool { return items[i].ForwarderID < items[j].ForwarderID })
|
||||
return items
|
||||
}
|
||||
|
||||
func newRunner(config storepkg.MQTTForwarderConfig) *runner {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &runner{config: config, ctx: ctx, cancel: cancel, startedAt: time.Now(), loopCache: make(map[string]time.Time)}
|
||||
}
|
||||
|
||||
func (r *runner) Start() {
|
||||
r.source = r.newClient(true)
|
||||
r.target = r.newClient(false)
|
||||
r.connectClient(r.target, "target")
|
||||
r.connectClient(r.source, "source")
|
||||
}
|
||||
|
||||
func (r *runner) Stop() {
|
||||
r.cancel()
|
||||
if r.source != nil && r.source.IsConnected() {
|
||||
r.source.Disconnect(250)
|
||||
}
|
||||
if r.target != nil && r.target.IsConnected() {
|
||||
r.target.Disconnect(250)
|
||||
}
|
||||
}
|
||||
|
||||
func (r *runner) Status() RuntimeStatus {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
started := r.startedAt
|
||||
return RuntimeStatus{
|
||||
ForwarderID: r.config.Forwarder.ID,
|
||||
Running: true,
|
||||
SourceConnected: r.sourceConnected,
|
||||
TargetConnected: r.targetConnected,
|
||||
LastError: r.lastError,
|
||||
StartedAt: &started,
|
||||
MessagesForwarded: r.messagesForwarded,
|
||||
MessagesDropped: r.messagesDropped,
|
||||
}
|
||||
}
|
||||
|
||||
func (r *runner) newClient(source bool) pahomqtt.Client {
|
||||
forwarder := r.config.Forwarder
|
||||
host, port, username, password, clientID, useTLS := forwarder.SourceHost, forwarder.SourcePort, forwarder.SourceUsername, forwarder.SourcePassword, forwarder.SourceClientID, forwarder.SourceTLS
|
||||
role := "source"
|
||||
if !source {
|
||||
host, port, username, password, clientID, useTLS = forwarder.TargetHost, forwarder.TargetPort, forwarder.TargetUsername, forwarder.TargetPassword, forwarder.TargetClientID, forwarder.TargetTLS
|
||||
role = "target"
|
||||
}
|
||||
if clientID == "" {
|
||||
clientID = fmt.Sprintf("mesh-forward-%d-%s", forwarder.ID, role)
|
||||
}
|
||||
scheme := "tcp"
|
||||
if useTLS {
|
||||
scheme = "ssl"
|
||||
}
|
||||
opts := pahomqtt.NewClientOptions().
|
||||
AddBroker(fmt.Sprintf("%s://%s", scheme, net.JoinHostPort(host, fmt.Sprint(port)))).
|
||||
SetClientID(clientID).
|
||||
SetAutoReconnect(true).
|
||||
SetConnectRetry(true).
|
||||
SetKeepAlive(60 * time.Second).
|
||||
SetConnectionLostHandler(func(_ pahomqtt.Client, err error) {
|
||||
r.setConnected(source, false)
|
||||
r.setError(fmt.Sprintf("%s connection lost: %v", role, err))
|
||||
}).
|
||||
SetOnConnectHandler(func(client pahomqtt.Client) {
|
||||
r.setConnected(source, true)
|
||||
r.subscribe(client, source)
|
||||
})
|
||||
if username != "" {
|
||||
opts.SetUsername(username)
|
||||
}
|
||||
if password != "" {
|
||||
opts.SetPassword(password)
|
||||
}
|
||||
if useTLS {
|
||||
opts.SetTLSConfig(&tls.Config{MinVersion: tls.VersionTLS12})
|
||||
}
|
||||
return pahomqtt.NewClient(opts)
|
||||
}
|
||||
|
||||
func (r *runner) connectClient(client pahomqtt.Client, label string) {
|
||||
token := client.Connect()
|
||||
if !token.WaitTimeout(2 * time.Second) {
|
||||
r.setError(label + " connect pending")
|
||||
return
|
||||
}
|
||||
if err := token.Error(); err != nil {
|
||||
r.setError(fmt.Sprintf("%s connect failed: %v", label, err))
|
||||
}
|
||||
}
|
||||
|
||||
func (r *runner) subscribe(client pahomqtt.Client, source bool) {
|
||||
for _, topic := range r.config.Topics {
|
||||
filter := topic.Topic
|
||||
if !source {
|
||||
if topic.Direction != storepkg.MQTTForwardDirectionBidirectional {
|
||||
continue
|
||||
}
|
||||
filter = mapMQTTForwardTopic(topic.Topic, topic.SourcePrefix, topic.TargetPrefix)
|
||||
}
|
||||
topicRule := topic
|
||||
token := client.Subscribe(filter, byte(topic.QoS), func(_ pahomqtt.Client, msg pahomqtt.Message) {
|
||||
r.forwardMessage(source, topicRule, msg)
|
||||
})
|
||||
if !token.WaitTimeout(2 * time.Second) {
|
||||
r.setError("subscribe pending: " + filter)
|
||||
continue
|
||||
}
|
||||
if err := token.Error(); err != nil {
|
||||
r.setError(fmt.Sprintf("subscribe %s failed: %v", filter, err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *runner) forwardMessage(fromSource bool, rule storepkg.MQTTForwardTopicRecord, msg pahomqtt.Message) {
|
||||
if r.ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
fromTopic := msg.Topic()
|
||||
if fromSource {
|
||||
if !mqttTopicFilterMatches(rule.Topic, fromTopic) {
|
||||
return
|
||||
}
|
||||
} else if !mqttTopicFilterMatches(mapMQTTForwardTopic(rule.Topic, rule.SourcePrefix, rule.TargetPrefix), fromTopic) {
|
||||
return
|
||||
}
|
||||
toTopic := fromTopic
|
||||
forwardDirection := storepkg.MQTTForwardDirectionSourceToTarget
|
||||
if fromSource {
|
||||
toTopic = mapMQTTForwardTopic(fromTopic, rule.SourcePrefix, rule.TargetPrefix)
|
||||
} else {
|
||||
forwardDirection = mqttForwardDirectionTargetToSource
|
||||
toTopic = mapMQTTForwardTopic(fromTopic, rule.TargetPrefix, rule.SourcePrefix)
|
||||
}
|
||||
if r.isSuppressed(forwardDirection, fromTopic, toTopic, msg.Payload(), rule.QoS, rule.Retain) {
|
||||
r.incDropped()
|
||||
return
|
||||
}
|
||||
target := r.target
|
||||
reverseDirection := mqttForwardDirectionTargetToSource
|
||||
if !fromSource {
|
||||
target = r.source
|
||||
reverseDirection = storepkg.MQTTForwardDirectionSourceToTarget
|
||||
}
|
||||
r.markSuppressed(reverseDirection, toTopic, fromTopic, msg.Payload(), rule.QoS, rule.Retain)
|
||||
token := target.Publish(toTopic, byte(rule.QoS), rule.Retain, msg.Payload())
|
||||
if !token.WaitTimeout(2 * time.Second) {
|
||||
r.setError("publish pending: " + toTopic)
|
||||
r.incDropped()
|
||||
return
|
||||
}
|
||||
if err := token.Error(); err != nil {
|
||||
r.setError(fmt.Sprintf("publish %s failed: %v", toTopic, err))
|
||||
r.incDropped()
|
||||
return
|
||||
}
|
||||
r.incForwarded()
|
||||
}
|
||||
|
||||
func (r *runner) setConnected(source bool, connected bool) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if source {
|
||||
r.sourceConnected = connected
|
||||
} else {
|
||||
r.targetConnected = connected
|
||||
}
|
||||
}
|
||||
|
||||
func (r *runner) setError(message string) {
|
||||
r.mu.Lock()
|
||||
r.lastError = message
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
func (r *runner) incForwarded() {
|
||||
r.mu.Lock()
|
||||
r.messagesForwarded++
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
func (r *runner) incDropped() {
|
||||
r.mu.Lock()
|
||||
r.messagesDropped++
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
func (r *runner) isSuppressed(direction, fromTopic, toTopic string, payload []byte, qos int, retain bool) bool {
|
||||
key := mqttForwardLoopKey(direction, fromTopic, toTopic, payload, qos, retain)
|
||||
now := time.Now()
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
expires, ok := r.loopCache[key]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
if now.After(expires) {
|
||||
delete(r.loopCache, key)
|
||||
return false
|
||||
}
|
||||
delete(r.loopCache, key)
|
||||
return true
|
||||
}
|
||||
|
||||
func (r *runner) markSuppressed(direction, fromTopic, toTopic string, payload []byte, qos int, retain bool) {
|
||||
key := mqttForwardLoopKey(direction, fromTopic, toTopic, payload, qos, retain)
|
||||
now := time.Now()
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if len(r.loopCache) >= mqttForwardLoopMaxEntries {
|
||||
for existing, expires := range r.loopCache {
|
||||
if now.After(expires) || len(r.loopCache) >= mqttForwardLoopMaxEntries {
|
||||
delete(r.loopCache, existing)
|
||||
}
|
||||
if len(r.loopCache) < mqttForwardLoopMaxEntries {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
r.loopCache[key] = now.Add(mqttForwardLoopTTL)
|
||||
}
|
||||
|
||||
func mqttForwardLoopKey(direction, fromTopic, toTopic string, payload []byte, qos int, retain bool) string {
|
||||
sum := sha256.Sum256(payload)
|
||||
return fmt.Sprintf("%s\x00%s\x00%s\x00%d\x00%t\x00%s", direction, fromTopic, toTopic, qos, retain, hex.EncodeToString(sum[:]))
|
||||
}
|
||||
|
||||
func mapMQTTForwardTopic(topic, fromPrefix, toPrefix string) string {
|
||||
fromPrefix = strings.Trim(fromPrefix, "/")
|
||||
toPrefix = strings.Trim(toPrefix, "/")
|
||||
if fromPrefix == "" {
|
||||
return topic
|
||||
}
|
||||
if topic == fromPrefix {
|
||||
return toPrefix
|
||||
}
|
||||
if strings.HasPrefix(topic, fromPrefix+"/") {
|
||||
if toPrefix == "" {
|
||||
return strings.TrimPrefix(topic, fromPrefix+"/")
|
||||
}
|
||||
return toPrefix + strings.TrimPrefix(topic, fromPrefix)
|
||||
}
|
||||
return topic
|
||||
}
|
||||
|
||||
func mqttTopicFilterMatches(filter, topic string) bool {
|
||||
filterParts := strings.Split(filter, "/")
|
||||
topicParts := strings.Split(topic, "/")
|
||||
for i, filterPart := range filterParts {
|
||||
if filterPart == "#" {
|
||||
return i == len(filterParts)-1
|
||||
}
|
||||
if i >= len(topicParts) {
|
||||
return false
|
||||
}
|
||||
if filterPart == "+" {
|
||||
continue
|
||||
}
|
||||
if filterPart != topicParts[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return len(filterParts) == len(topicParts)
|
||||
}
|
||||
Reference in New Issue
Block a user