cactoandClaude Sonnet 5
938700cf86
Subscribes to devices/+/telemetry, validates payloads, buffers writes into ClickHouse (flush on size or interval), and emits a per-reading event on RabbitMQ for rule-engine-service to consume later. Adds the initial Postgres and ClickHouse schema migrations, a Dockerfile, and wires the service into docker-compose. Verified end-to-end via docker compose + mosquitto_pub: row lands in ClickHouse and the event reaches the telemetry.new_reading queue. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
61 lines
1.7 KiB
Go
61 lines
1.7 KiB
Go
// Package mqttclient subscribes to device telemetry topics over MQTT.
|
|
package mqttclient
|
|
|
|
import (
|
|
"fmt"
|
|
"log/slog"
|
|
"time"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
)
|
|
|
|
// Handler receives one message's raw payload and the time it arrived.
|
|
type Handler func(payload []byte, receivedAt time.Time)
|
|
|
|
type Config struct {
|
|
BrokerURL string // e.g. tcp://mosquitto:1883
|
|
ClientID string
|
|
Topic string // e.g. devices/+/telemetry
|
|
QoS byte
|
|
}
|
|
|
|
type Subscriber struct {
|
|
client mqtt.Client
|
|
}
|
|
|
|
// Connect dials the broker and (re-)subscribes to cfg.Topic on every
|
|
// successful connection, so a dropped connection resubscribes automatically
|
|
// once auto-reconnect succeeds.
|
|
func Connect(cfg Config, handler Handler, logger *slog.Logger) (*Subscriber, error) {
|
|
opts := mqtt.NewClientOptions().
|
|
AddBroker(cfg.BrokerURL).
|
|
SetClientID(cfg.ClientID).
|
|
SetAutoReconnect(true).
|
|
SetConnectRetry(true).
|
|
SetOnConnectHandler(func(c mqtt.Client) {
|
|
token := c.Subscribe(cfg.Topic, cfg.QoS, func(_ mqtt.Client, msg mqtt.Message) {
|
|
handler(msg.Payload(), time.Now())
|
|
})
|
|
token.Wait()
|
|
if err := token.Error(); err != nil {
|
|
logger.Error("mqtt subscribe failed", "topic", cfg.Topic, "error", err)
|
|
}
|
|
}).
|
|
SetConnectionLostHandler(func(_ mqtt.Client, err error) {
|
|
logger.Warn("mqtt connection lost", "error", err)
|
|
})
|
|
|
|
client := mqtt.NewClient(opts)
|
|
token := client.Connect()
|
|
token.Wait()
|
|
if err := token.Error(); err != nil {
|
|
return nil, fmt.Errorf("connect to mqtt broker %s: %w", cfg.BrokerURL, err)
|
|
}
|
|
return &Subscriber{client: client}, nil
|
|
}
|
|
|
|
// Close disconnects, waiting up to quiesceMS for in-flight work to settle.
|
|
func (s *Subscriber) Close() {
|
|
s.client.Disconnect(250)
|
|
}
|