99 lines
3.3 KiB
Go
99 lines
3.3 KiB
Go
// Package bridge — MQTT-подписка на телеметрию устройств → shared-SQS.
|
|
//
|
|
// Роль старого mqtt-bridge, но внутри монолита: подключается к EMQX как
|
|
// MQTT-клиент, подписывается на "+/telemetry/+" и шлёт envelope в SQS.
|
|
package bridge
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"github.com/aws/aws-sdk-go-v2/service/sqs"
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
|
|
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/config"
|
|
"gitea.services.ngcloud.ru/Nail/IoT/internal/service/sqsclient"
|
|
)
|
|
|
|
// telemetryTopicFilter — все namespace, все устройства.
|
|
const telemetryTopicFilter = "+/telemetry/+"
|
|
|
|
// Run — бесконечный цикл bridge: подключение → подписка → ожидание сигнала завершения.
|
|
func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, log *slog.Logger) error {
|
|
log.Info("bridge: starting", "broker", cfg.MQTTBrokerURL, "client_id", cfg.MQTTClientID)
|
|
|
|
// URL очереди получаем один раз при старте (как старый consumer).
|
|
queueURL, err := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSQueueName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
log.Info("bridge: SQS queue resolved", "queue", cfg.SQSQueueName)
|
|
|
|
client, err := connectMQTT(ctx, cfg, log)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer client.Disconnect(250)
|
|
|
|
handler := newMessageHandler(ctx, sqsClient, queueURL, log)
|
|
token := client.Subscribe(telemetryTopicFilter, 1, handler)
|
|
if !token.WaitTimeout(10 * time.Second) {
|
|
return errSubscribeTimeout
|
|
}
|
|
if token.Error() != nil {
|
|
return token.Error()
|
|
}
|
|
log.Info("bridge: subscribed", "filter", telemetryTopicFilter)
|
|
|
|
<-ctx.Done()
|
|
log.Info("bridge: shutting down")
|
|
return nil
|
|
}
|
|
|
|
// errSubscribeTimeout — не дождались подтверждения подписки.
|
|
var errSubscribeTimeout = errBridge("MQTT subscribe timeout")
|
|
|
|
// errBridge — ошибки bridge.
|
|
type errBridge string
|
|
|
|
func (e errBridge) Error() string { return string(e) }
|
|
|
|
// connectMQTT устанавливает подключение к EMQX с автореконнектом.
|
|
func connectMQTT(ctx context.Context, cfg *config.Config, log *slog.Logger) (mqtt.Client, error) {
|
|
opts := mqtt.NewClientOptions()
|
|
opts.AddBroker(cfg.MQTTBrokerURL)
|
|
opts.SetClientID(cfg.MQTTClientID)
|
|
opts.SetUsername(cfg.MQTTUsername)
|
|
opts.SetPassword(cfg.MQTTPassword)
|
|
opts.SetAutoReconnect(true)
|
|
opts.SetConnectRetry(true)
|
|
opts.SetConnectRetryInterval(5 * time.Second)
|
|
opts.SetKeepAlive(30 * time.Second)
|
|
opts.SetCleanSession(false)
|
|
|
|
opts.SetConnectionLostHandler(func(_ mqtt.Client, err error) {
|
|
log.Warn("bridge: MQTT connection lost, reconnecting...", "err", err)
|
|
})
|
|
opts.SetReconnectingHandler(func(_ mqtt.Client, _ *mqtt.ClientOptions) {
|
|
log.Info("bridge: MQTT reconnecting...")
|
|
})
|
|
opts.SetOnConnectHandler(func(_ mqtt.Client) {
|
|
log.Info("bridge: MQTT connected")
|
|
})
|
|
|
|
client := mqtt.NewClient(opts)
|
|
token := client.Connect()
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
case <-time.After(30 * time.Second):
|
|
return nil, errBridge("bridge: MQTT connect timeout")
|
|
case <-token.Done():
|
|
}
|
|
if token.Error() != nil {
|
|
return nil, token.Error()
|
|
}
|
|
return client, nil
|
|
}
|