Repinoid
c3106d6e20
bridge пишет, consumer парсит RFC3339Nano (субсекундная точность). Раньше RFC3339 (секунды) + non-FIFO очередь → показания в одну секунду могли встать не в том порядке. Parse RFC3339Nano совместим со старыми RFC3339-сообщениями (доля секунды опциональна).
62 lines
2.3 KiB
Go
62 lines
2.3 KiB
Go
// handler.go — обработчик MQTT-сообщений: envelope → канал диспатчера SQS.
|
|
//
|
|
// Колбэк НЕ делает сетевых вызовов: только парсинг топика и неблокирующий
|
|
// enqueue. Отправка в SQS — worker-пул sqsDispatcher (см. sender.go).
|
|
package bridge
|
|
|
|
import (
|
|
"encoding/json"
|
|
"log/slog"
|
|
"strings"
|
|
"time"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// 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"`
|
|
// MessageID — уникальный id сообщения. Нужен для идемпотентности: SQS
|
|
// at-least-once может доставить дубликат — consumer дедуплицирует по нему.
|
|
MessageID string `json:"message_id"`
|
|
}
|
|
|
|
// newMessageHandler возвращает обработчик MQTT-сообщений:
|
|
// валидация топика "{namespace}/telemetry/{deviceId}" → enqueue в диспатчер.
|
|
func newMessageHandler(dispatcher *sqsDispatcher, 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 || parts[0] == "" || parts[2] == "" || parts[1] != "telemetry" {
|
|
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)
|
|
}
|
|
|
|
dispatcher.enqueue(TelemetryEnvelope{
|
|
Namespace: ns,
|
|
DeviceID: deviceID,
|
|
Topic: topic,
|
|
Payload: rawPayload,
|
|
ReceivedAt: time.Now().UTC().Format(time.RFC3339Nano),
|
|
MessageID: uuid.NewString(),
|
|
})
|
|
}
|
|
}
|