d0137179e5
- neues Paket mqttheartbeat: Publisher mit paho, topic signage/screen/<id>/heartbeat,
payload {screen_id, ts, status, server_connectivity}, auto-reconnect bei Ausfall
- MORZ_INFOBOARD_MQTT_BROKER leer (Standard) -> MQTT komplett uebersprungen
- app.emitHeartbeat() publiziert bei jedem Tick per MQTT wenn Broker konfiguriert,
loggt Fehler und laeuft weiter (kein Stop bei MQTT-Ausfall)
- mqtt.Close() bei context.Done()
- MQTTBroker-Default von tcp://127.0.0.1:1883 auf "" geaendert
- erste externe Dep: github.com/eclipse/paho.mqtt.golang v1.5.1
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
78 lines
2.4 KiB
Go
78 lines
2.4 KiB
Go
package mqttheartbeat
|
|
|
|
import (
|
|
"encoding/json"
|
|
"time"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
)
|
|
|
|
// payload is the JSON structure published to the heartbeat topic.
|
|
type payload struct {
|
|
ScreenID string `json:"screen_id"`
|
|
Timestamp string `json:"ts"`
|
|
Status string `json:"status"`
|
|
ServerConnectivity string `json:"server_connectivity"`
|
|
}
|
|
|
|
// Publisher publishes MQTT heartbeats for a single screen.
|
|
// It connects with auto-reconnect enabled, so transient broker outages are
|
|
// handled transparently; individual Publish calls fail and should be logged
|
|
// by the caller.
|
|
type Publisher struct {
|
|
client mqtt.Client
|
|
screenID string
|
|
}
|
|
|
|
// New creates a Publisher and initiates a non-blocking connection to broker.
|
|
// paho retries the connection in the background if the broker is unreachable
|
|
// at startup, so New always succeeds. Publish calls will return errors until
|
|
// the connection is established.
|
|
func New(broker, screenID string) *Publisher {
|
|
opts := mqtt.NewClientOptions().
|
|
AddBroker(broker).
|
|
SetClientID("morz-agent-" + screenID).
|
|
SetCleanSession(true).
|
|
SetAutoReconnect(true).
|
|
SetConnectRetry(true).
|
|
SetConnectRetryInterval(10 * time.Second)
|
|
|
|
client := mqtt.NewClient(opts)
|
|
client.Connect() // non-blocking; paho retries in background
|
|
return &Publisher{client: client, screenID: screenID}
|
|
}
|
|
|
|
// Topic returns the MQTT topic for the screen's heartbeat.
|
|
func Topic(screenID string) string {
|
|
return "signage/screen/" + screenID + "/heartbeat"
|
|
}
|
|
|
|
// BuildPayload builds the JSON heartbeat payload. Exported for testing.
|
|
func BuildPayload(screenID, status, connectivity string, ts time.Time) ([]byte, error) {
|
|
return json.Marshal(payload{
|
|
ScreenID: screenID,
|
|
Timestamp: ts.UTC().Format(time.RFC3339),
|
|
Status: status,
|
|
ServerConnectivity: connectivity,
|
|
})
|
|
}
|
|
|
|
// SendHeartbeat publishes a heartbeat to the screen's topic.
|
|
// QoS 0, not retained. Returns an error if the broker is unreachable or
|
|
// the publish token fails; callers should log and continue.
|
|
func (p *Publisher) SendHeartbeat(status, connectivity string, ts time.Time) error {
|
|
data, err := BuildPayload(p.screenID, status, connectivity, ts)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
topic := Topic(p.screenID)
|
|
token := p.client.Publish(topic, 0, false, data)
|
|
token.WaitTimeout(3 * time.Second)
|
|
return token.Error()
|
|
}
|
|
|
|
// Close disconnects from the broker gracefully.
|
|
func (p *Publisher) Close() {
|
|
p.client.Disconnect(250)
|
|
}
|