72 lines
2.1 KiB
Go
72 lines
2.1 KiB
Go
// handler.go — обработчик MQTT-сообщений: envelope → SQS SendMessage.
|
|
package bridge
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/aws/aws-sdk-go-v2/aws"
|
|
"github.com/aws/aws-sdk-go-v2/service/sqs"
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
)
|
|
|
|
// TelemetryEnvelope — сообщение, перекладываемое в SQS (разбирается consumer).
|
|
type TelemetryEnvelope struct {
|
|
Namespace string `json:"namespace"`
|
|
DeviceID string `json:"device_id"`
|
|
Topic string `json:"topic"`
|
|
Payload json.RawMessage `json:"payload"`
|
|
ReceivedAt string `json:"received_at"`
|
|
}
|
|
|
|
// newMessageHandler возвращает обработчик MQTT-сообщений.
|
|
func newMessageHandler(ctx context.Context, sqsClient *sqs.Client, queueURL string, log *slog.Logger) mqtt.MessageHandler {
|
|
return func(_ mqtt.Client, msg mqtt.Message) {
|
|
topic := msg.Topic()
|
|
payload := msg.Payload()
|
|
|
|
// Топик: "{namespace}/telemetry/{deviceId}".
|
|
parts := strings.SplitN(topic, "/", 3)
|
|
if len(parts) != 3 {
|
|
log.Warn("bridge: unexpected topic format, skipping", "topic", topic)
|
|
return
|
|
}
|
|
ns := parts[0]
|
|
deviceID := parts[2]
|
|
|
|
// Не-JSON payload оборачиваем в строку.
|
|
rawPayload := json.RawMessage(payload)
|
|
if !json.Valid(payload) {
|
|
quoted, _ := json.Marshal(string(payload))
|
|
rawPayload = json.RawMessage(quoted)
|
|
}
|
|
|
|
envelope := TelemetryEnvelope{
|
|
Namespace: ns,
|
|
DeviceID: deviceID,
|
|
Topic: topic,
|
|
Payload: rawPayload,
|
|
ReceivedAt: time.Now().UTC().Format(time.RFC3339),
|
|
}
|
|
body, err := json.Marshal(envelope)
|
|
if err != nil {
|
|
log.Error("bridge: marshal envelope", "topic", topic, "err", err)
|
|
return
|
|
}
|
|
|
|
_, err = sqsClient.SendMessage(ctx, &sqs.SendMessageInput{
|
|
QueueUrl: aws.String(queueURL),
|
|
MessageBody: aws.String(string(body)),
|
|
})
|
|
if err != nil {
|
|
log.Error("bridge: SQS SendMessage failed", "topic", topic, "err", err)
|
|
return
|
|
}
|
|
log.Info("bridge: forwarded telemetry to SQS",
|
|
"mqtt_topic", topic, "namespace", ns, "device", deviceID)
|
|
}
|
|
}
|