150 lines
3.5 KiB
Go
150 lines
3.5 KiB
Go
package mqttbridge
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"time"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
|
|
"git.i3omb.com/gronod/ha-n95-local-control/internal/robot"
|
|
)
|
|
|
|
// publish issues one QoS 0 publish and waits up to publishTimeout. Logs
|
|
// carry topic and error class only — never payloads.
|
|
func (b *Bridge) publish(s *slot, topic string, retain bool, payload []byte) error {
|
|
<-s.ready
|
|
tok := s.client.Publish(topic, 0, retain, payload)
|
|
if !tok.WaitTimeout(publishTimeout) {
|
|
slog.Error("mqtt publish failed", "topic", topic, "err", "timeout")
|
|
return errPublishTimeout
|
|
}
|
|
if err := tok.Error(); err != nil {
|
|
slog.Error("mqtt publish failed", "topic", topic, "err", err)
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b *Bridge) Publish(serial, suffix string, retain bool, payload []byte) error {
|
|
s := b.slotFor(serial)
|
|
if s == nil {
|
|
return fmt.Errorf("mqttbridge: unknown serial %q", serial)
|
|
}
|
|
return b.publish(s, b.root(serial)+"/"+suffix, retain, payload)
|
|
}
|
|
|
|
func (b *Bridge) Republisher(serial string) robot.RepublishFunc {
|
|
s := b.ensureSlot(serial)
|
|
if s == nil {
|
|
return func(context.Context, robot.Snapshot) {}
|
|
}
|
|
return func(_ context.Context, snap robot.Snapshot) {
|
|
if b.isClosing() {
|
|
return
|
|
}
|
|
s.stateMu.Lock()
|
|
defer s.stateMu.Unlock()
|
|
if b.isClosing() {
|
|
return
|
|
}
|
|
s.mu.Lock()
|
|
s.latest = snap
|
|
s.mu.Unlock()
|
|
stateJSON, err := json.Marshal(snap.State)
|
|
if err != nil {
|
|
return
|
|
}
|
|
attrJSON, err := json.Marshal(snap.Attributes)
|
|
if err != nil {
|
|
return
|
|
}
|
|
root := b.root(serial)
|
|
_ = b.publish(s, root+"/state", true, stateJSON)
|
|
_ = b.publish(s, root+"/json_attributes", true, attrJSON)
|
|
}
|
|
}
|
|
|
|
func (b *Bridge) DispatchSendCommand(ctx context.Context, serial string, payload []byte) (robot.Command, error) {
|
|
b.mu.Lock()
|
|
fn := b.SendCommand
|
|
b.mu.Unlock()
|
|
if fn == nil {
|
|
return robot.Command{}, errors.New("rejected:unsupported")
|
|
}
|
|
return fn(ctx, serial, payload)
|
|
}
|
|
|
|
// Shutdown marks the bridge closing, clears live on every slot, publishes
|
|
// retained offline for each, waits on all tokens bounded by ctx, then
|
|
// disconnects cleanly so brokers discard the wills.
|
|
func (b *Bridge) Shutdown(ctx context.Context) error {
|
|
b.mu.Lock()
|
|
b.closing = true
|
|
slots := make([]*slot, 0, len(b.slots))
|
|
for _, s := range b.slots {
|
|
slots = append(slots, s)
|
|
}
|
|
b.mu.Unlock()
|
|
|
|
var errs []error
|
|
var ready []*slot
|
|
toks := make([]mqtt.Token, 0, len(slots))
|
|
for _, s := range slots {
|
|
s.availabilityMu.Lock()
|
|
s.mu.Lock()
|
|
s.live = false
|
|
s.mu.Unlock()
|
|
select {
|
|
case <-s.ready:
|
|
case <-ctx.Done():
|
|
errs = append(errs, ctx.Err())
|
|
s.availabilityMu.Unlock()
|
|
continue
|
|
}
|
|
s.connectMu.Lock()
|
|
s.connectMu.Unlock()
|
|
toks = append(toks, s.client.Publish(b.root(s.serial)+"/availability", 0, true, []byte("offline")))
|
|
ready = append(ready, s)
|
|
s.availabilityMu.Unlock()
|
|
}
|
|
for _, tok := range toks {
|
|
select {
|
|
case <-tok.Done():
|
|
if err := tok.Error(); err != nil {
|
|
errs = append(errs, err)
|
|
}
|
|
case <-ctx.Done():
|
|
errs = append(errs, ctx.Err())
|
|
}
|
|
}
|
|
for _, s := range ready {
|
|
s.client.Disconnect(quiesceFor(ctx))
|
|
}
|
|
return errors.Join(errs...)
|
|
}
|
|
|
|
// quiesceFor bounds a Disconnect quiesce by the ctx deadline so many
|
|
// clients cannot extend shutdown past the caller's budget.
|
|
func quiesceFor(ctx context.Context) uint {
|
|
const max uint = 250
|
|
if dl, ok := ctx.Deadline(); ok {
|
|
rem := time.Until(dl)
|
|
if rem <= 0 {
|
|
return 0
|
|
}
|
|
ms := uint(rem / time.Millisecond)
|
|
if ms > max {
|
|
return max
|
|
}
|
|
return ms
|
|
}
|
|
if ctx.Err() != nil {
|
|
return 0
|
|
}
|
|
return max
|
|
}
|