wshub: add setCalibration/reloadConfig frames and config acks
Wire five new WebSocket frames into hub.go: setCalibration (client→hub), calibration (hub→client broadcast), reloadConfig (client→hub), configSaved and configReloaded (hub→client acks with ok/path/error). Extend hubCmd with cal field, add buildCalibrationMsg and buildConfigAckMsg builders, send calibration on client connect, and run Save/Reload on separate goroutines to avoid blocking the Run() select loop. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
dfd257cfd9
commit
ffe7cb1cc5
@@ -107,6 +107,27 @@ func (c *wsClient) readPump() {
|
|||||||
case c.hub.commandCh <- hubCmd{op: "wsSaveSources"}:
|
case c.hub.commandCh <- hubCmd{op: "wsSaveSources"}:
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
|
case "setCalibration":
|
||||||
|
source, _ := env["source"].(string)
|
||||||
|
signal, _ := env["signal"].(string)
|
||||||
|
scale, hasScale := env["scale"].(float64)
|
||||||
|
if !hasScale {
|
||||||
|
scale = 1
|
||||||
|
}
|
||||||
|
offset, _ := env["offset"].(float64)
|
||||||
|
unit, _ := env["unit"].(string)
|
||||||
|
select {
|
||||||
|
case c.hub.commandCh <- hubCmd{op: "wsSetCalibration", cal: CalConfig{
|
||||||
|
Source: source, Signal: signal,
|
||||||
|
Scale: scale, Offset: offset, Unit: unit,
|
||||||
|
}}:
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
case "reloadConfig":
|
||||||
|
select {
|
||||||
|
case c.hub.commandCh <- hubCmd{op: "wsReloadConfig"}:
|
||||||
|
default:
|
||||||
|
}
|
||||||
case "setMonotonic":
|
case "setMonotonic":
|
||||||
enabled, _ := env["enabled"].(bool)
|
enabled, _ := env["enabled"].(bool)
|
||||||
select {
|
select {
|
||||||
@@ -212,7 +233,8 @@ type taggedSample struct {
|
|||||||
// hubCmd carries a command to the Run() goroutine.
|
// hubCmd carries a command to the Run() goroutine.
|
||||||
type hubCmd struct {
|
type hubCmd struct {
|
||||||
op string // "addSource","removeSource","setSourceState","updateConfig",
|
op string // "addSource","removeSource","setSourceState","updateConfig",
|
||||||
// "wsAddSource","wsRemoveSource","wsSaveSources"
|
// "wsAddSource","wsRemoveSource","wsSaveSources",
|
||||||
|
// "wsSetCalibration","wsReloadConfig"
|
||||||
sourceID string
|
sourceID string
|
||||||
label string
|
label string
|
||||||
addr string
|
addr string
|
||||||
@@ -221,6 +243,7 @@ type hubCmd struct {
|
|||||||
multicastGroup string
|
multicastGroup string
|
||||||
dataPort int
|
dataPort int
|
||||||
enabled bool // "setMonotonic" toggle
|
enabled bool // "setMonotonic" toggle
|
||||||
|
cal CalConfig // "wsSetCalibration" payload
|
||||||
}
|
}
|
||||||
|
|
||||||
// Hub is the central broker between UDP clients and WebSocket clients.
|
// Hub is the central broker between UDP clients and WebSocket clients.
|
||||||
@@ -475,6 +498,26 @@ func buildSourcesMsg(sm map[string]*sourceHubState) []byte {
|
|||||||
return msg
|
return msg
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// buildCalibrationMsg serialises the calibration table as a "calibration"
|
||||||
|
// message. It is its own frame rather than a field on "sources" because the
|
||||||
|
// C++ BroadcastSources serialises into a fixed 4096-byte buffer that a
|
||||||
|
// calibration table would overflow.
|
||||||
|
func buildCalibrationMsg(t *calTable) []byte {
|
||||||
|
list := t.List() // never nil: the SPA replaces its table wholesale on receipt
|
||||||
|
msg, _ := json.Marshal(map[string]any{"type": "calibration", "cal": list})
|
||||||
|
return msg
|
||||||
|
}
|
||||||
|
|
||||||
|
// buildConfigAckMsg serialises a configSaved / configReloaded acknowledgement.
|
||||||
|
func buildConfigAckMsg(msgType, path string, err error) []byte {
|
||||||
|
m := map[string]any{"type": msgType, "ok": err == nil, "path": path}
|
||||||
|
if err != nil {
|
||||||
|
m["error"] = err.Error()
|
||||||
|
}
|
||||||
|
msg, _ := json.Marshal(m)
|
||||||
|
return msg
|
||||||
|
}
|
||||||
|
|
||||||
// Run is the hub's main goroutine. Must be started with go hub.Run().
|
// Run is the hub's main goroutine. Must be started with go hub.Run().
|
||||||
func (h *Hub) Run() {
|
func (h *Hub) Run() {
|
||||||
ticker := time.NewTicker(time.Second / 30)
|
ticker := time.NewTicker(time.Second / 30)
|
||||||
@@ -522,6 +565,11 @@ func (h *Hub) Run() {
|
|||||||
case c.send <- wsMessage{websocket.TextMessage, monoMsg}:
|
case c.send <- wsMessage{websocket.TextMessage, monoMsg}:
|
||||||
default:
|
default:
|
||||||
}
|
}
|
||||||
|
calMsg := buildCalibrationMsg(h.cal)
|
||||||
|
select {
|
||||||
|
case c.send <- wsMessage{websocket.TextMessage, calMsg}:
|
||||||
|
default:
|
||||||
|
}
|
||||||
// Notify the application layer so it can replay any persistent state
|
// Notify the application layer so it can replay any persistent state
|
||||||
// (e.g., MARTe2 connection status, forced/traced signals).
|
// (e.g., MARTe2 connection status, forced/traced signals).
|
||||||
h.onClientConnectMu.RLock()
|
h.onClientConnectMu.RLock()
|
||||||
@@ -647,10 +695,44 @@ func (h *Hub) Run() {
|
|||||||
|
|
||||||
case "wsSaveSources":
|
case "wsSaveSources":
|
||||||
if h.sm != nil {
|
if h.sm != nil {
|
||||||
if err := h.sm.Save(); err != nil {
|
// Save writes to disk; run it off the Run() goroutine so a
|
||||||
log.Printf("hub: save sources: %v", err)
|
// slow filesystem can never stall the hub loop.
|
||||||
|
go func(sm *SourceManager) {
|
||||||
|
err := sm.Save()
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("hub: save config: %v", err)
|
||||||
}
|
}
|
||||||
|
h.broadcast(buildConfigAckMsg("configSaved", sm.Path(), err))
|
||||||
|
}(h.sm)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
case "wsSetCalibration":
|
||||||
|
if h.cal.Set(cmd.cal) {
|
||||||
|
h.broadcast(buildCalibrationMsg(h.cal))
|
||||||
|
} else {
|
||||||
|
// No broadcast: the offending client reverts to the last
|
||||||
|
// value it was sent.
|
||||||
|
log.Printf("hub: rejected calibration %q/%q (scale=%v offset=%v)",
|
||||||
|
cmd.cal.Source, cmd.cal.Signal, cmd.cal.Scale, cmd.cal.Offset)
|
||||||
|
}
|
||||||
|
|
||||||
|
case "wsReloadConfig":
|
||||||
|
if h.sm != nil {
|
||||||
|
// Reload calls sm.Add(), which sends on commandCh; from the
|
||||||
|
// Run() goroutine that send would hit the non-blocking
|
||||||
|
// default and be dropped, so it must run elsewhere.
|
||||||
|
go func(sm *SourceManager) {
|
||||||
|
err := sm.Reload()
|
||||||
|
if err != nil {
|
||||||
|
log.Printf("hub: reload config: %v", err)
|
||||||
|
}
|
||||||
|
h.broadcast(buildConfigAckMsg("configReloaded", sm.Path(), err))
|
||||||
|
if err == nil {
|
||||||
|
h.broadcast(buildCalibrationMsg(h.cal))
|
||||||
|
}
|
||||||
|
}(h.sm)
|
||||||
|
}
|
||||||
|
|
||||||
case "setMonotonic":
|
case "setMonotonic":
|
||||||
h.monotonicTS = cmd.enabled
|
h.monotonicTS = cmd.enabled
|
||||||
monoMsg, _ := json.Marshal(map[string]any{"type": "monotonicState", "enabled": h.monotonicTS})
|
monoMsg, _ := json.Marshal(map[string]any{"type": "monotonicState", "enabled": h.monotonicTS})
|
||||||
|
|||||||
@@ -0,0 +1,136 @@
|
|||||||
|
package wshub
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestBuildCalibrationMsg(t *testing.T) {
|
||||||
|
tab := newCalTable()
|
||||||
|
tab.Set(CalConfig{Source: "wave", Signal: "Adc", Scale: 0.5, Offset: -1.25, Unit: "V"})
|
||||||
|
|
||||||
|
var got struct {
|
||||||
|
Type string `json:"type"`
|
||||||
|
Cal []CalConfig `json:"cal"`
|
||||||
|
}
|
||||||
|
raw := buildCalibrationMsg(tab)
|
||||||
|
if err := json.Unmarshal(raw, &got); err != nil {
|
||||||
|
t.Fatalf("unmarshal %s: %v", raw, err)
|
||||||
|
}
|
||||||
|
if got.Type != "calibration" {
|
||||||
|
t.Errorf("type = %q, want calibration", got.Type)
|
||||||
|
}
|
||||||
|
if len(got.Cal) != 1 || got.Cal[0] != (CalConfig{
|
||||||
|
Source: "wave", Signal: "Adc", Scale: 0.5, Offset: -1.25, Unit: "V"}) {
|
||||||
|
t.Errorf("cal = %+v", got.Cal)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBuildCalibrationMsgEmptyTableIsEmptyArray(t *testing.T) {
|
||||||
|
// The SPA replaces its table wholesale on every calibration message, so an
|
||||||
|
// empty table must serialise as [] and not as null.
|
||||||
|
raw := buildCalibrationMsg(newCalTable())
|
||||||
|
var got struct {
|
||||||
|
Cal []CalConfig `json:"cal"`
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(raw, &got); err != nil {
|
||||||
|
t.Fatalf("unmarshal %s: %v", raw, err)
|
||||||
|
}
|
||||||
|
if got.Cal == nil {
|
||||||
|
t.Errorf("cal = null, want []; raw = %s", raw)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestBuildConfigAckMsg(t *testing.T) {
|
||||||
|
ok := buildConfigAckMsg("configSaved", "/tmp/x.json", nil)
|
||||||
|
var m map[string]any
|
||||||
|
if err := json.Unmarshal(ok, &m); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if m["type"] != "configSaved" || m["ok"] != true || m["path"] != "/tmp/x.json" {
|
||||||
|
t.Errorf("success ack = %s", ok)
|
||||||
|
}
|
||||||
|
if _, has := m["error"]; has {
|
||||||
|
t.Errorf("success ack carries an error field: %s", ok)
|
||||||
|
}
|
||||||
|
|
||||||
|
bad := buildConfigAckMsg("configReloaded", "", errors.New("boom"))
|
||||||
|
m = nil
|
||||||
|
if err := json.Unmarshal(bad, &m); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if m["type"] != "configReloaded" || m["ok"] != false || m["error"] != "boom" {
|
||||||
|
t.Errorf("failure ack = %s", bad)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHubSetCalibrationCommand(t *testing.T) {
|
||||||
|
h := NewHub()
|
||||||
|
go h.Run()
|
||||||
|
|
||||||
|
// Register a client before sending commands so broadcasts are observable.
|
||||||
|
sendCh := make(chan wsMessage, 64)
|
||||||
|
c := &wsClient{hub: h, send: sendCh}
|
||||||
|
h.register <- c
|
||||||
|
sleepMillis(20) // let Run() process the register and flush initial state msgs
|
||||||
|
drainSendCh(sendCh) // discard state-sync messages (sources, trigger, cal, ...)
|
||||||
|
|
||||||
|
h.commandCh <- hubCmd{op: "wsSetCalibration", cal: CalConfig{
|
||||||
|
Source: "wave", Signal: "Adc", Scale: 4, Offset: 1, Unit: "V"}}
|
||||||
|
if raw := waitMsg(t, sendCh, "calibration"); raw == nil {
|
||||||
|
t.Fatal("no calibration broadcast after a valid setCalibration")
|
||||||
|
}
|
||||||
|
if got := h.cal.List(); len(got) != 1 || got[0].Scale != 4 {
|
||||||
|
t.Fatalf("table = %+v, want one entry with scale 4", got)
|
||||||
|
}
|
||||||
|
|
||||||
|
// An invalid entry is rejected and emits no broadcast at all.
|
||||||
|
h.commandCh <- hubCmd{op: "wsSetCalibration", cal: CalConfig{
|
||||||
|
Source: "wave", Signal: "Adc", Scale: 0}}
|
||||||
|
if raw := waitMsg(t, sendCh, "calibration"); raw != nil {
|
||||||
|
t.Errorf("invalid setCalibration broadcast %s", raw)
|
||||||
|
}
|
||||||
|
if got := h.cal.List(); len(got) != 1 || got[0].Scale != 4 {
|
||||||
|
t.Errorf("table changed after a rejected setCalibration: %+v", got)
|
||||||
|
}
|
||||||
|
|
||||||
|
h.unregister <- c
|
||||||
|
}
|
||||||
|
|
||||||
|
// drainSendCh reads all currently buffered messages from the channel.
|
||||||
|
func drainSendCh(ch chan wsMessage) {
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ch:
|
||||||
|
default:
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// waitMsg waits up to ~250 ms for a message of the given type on sendCh.
|
||||||
|
func waitMsg(t *testing.T, sendCh chan wsMessage, msgType string) []byte {
|
||||||
|
t.Helper()
|
||||||
|
deadline := time.After(250 * time.Millisecond)
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case msg := <-sendCh:
|
||||||
|
var env struct {
|
||||||
|
Type string `json:"type"`
|
||||||
|
}
|
||||||
|
if json.Unmarshal(msg.data, &env) == nil && env.Type == msgType {
|
||||||
|
return msg.data
|
||||||
|
}
|
||||||
|
case <-deadline:
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// waitBroadcast is kept for compatibility with the test helper interface;
|
||||||
|
// it delegates to waitMsg using a pre-registered client send channel.
|
||||||
|
// Callers that need it should register a client and use waitMsg directly.
|
||||||
|
|
||||||
|
func sleepMillis(n int) { time.Sleep(time.Duration(n) * time.Millisecond) }
|
||||||
Reference in New Issue
Block a user