From ffe7cb1cc5a5cc617c595eb1d463ea0e85c37a07 Mon Sep 17 00:00:00 2001 From: Martino Ferrari Date: Sun, 16 Aug 2026 19:36:09 +0200 Subject: [PATCH] wshub: add setCalibration/reloadConfig frames and config acks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- Common/Client/go/wshub/hub.go | 92 +++++++++++- .../Client/go/wshub/hub_calibration_test.go | 136 ++++++++++++++++++ 2 files changed, 223 insertions(+), 5 deletions(-) create mode 100644 Common/Client/go/wshub/hub_calibration_test.go diff --git a/Common/Client/go/wshub/hub.go b/Common/Client/go/wshub/hub.go index a67f609..a986772 100644 --- a/Common/Client/go/wshub/hub.go +++ b/Common/Client/go/wshub/hub.go @@ -107,6 +107,27 @@ func (c *wsClient) readPump() { case c.hub.commandCh <- hubCmd{op: "wsSaveSources"}: 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": enabled, _ := env["enabled"].(bool) select { @@ -212,7 +233,8 @@ type taggedSample struct { // hubCmd carries a command to the Run() goroutine. type hubCmd struct { op string // "addSource","removeSource","setSourceState","updateConfig", - // "wsAddSource","wsRemoveSource","wsSaveSources" + // "wsAddSource","wsRemoveSource","wsSaveSources", + // "wsSetCalibration","wsReloadConfig" sourceID string label string addr string @@ -220,7 +242,8 @@ type hubCmd struct { sigs []udpsprotocol.SignalInfo multicastGroup string 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. @@ -475,6 +498,26 @@ func buildSourcesMsg(sm map[string]*sourceHubState) []byte { 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(). func (h *Hub) Run() { ticker := time.NewTicker(time.Second / 30) @@ -522,6 +565,11 @@ func (h *Hub) Run() { case c.send <- wsMessage{websocket.TextMessage, monoMsg}: 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 // (e.g., MARTe2 connection status, forced/traced signals). h.onClientConnectMu.RLock() @@ -647,10 +695,44 @@ func (h *Hub) Run() { case "wsSaveSources": if h.sm != nil { - if err := h.sm.Save(); err != nil { - log.Printf("hub: save sources: %v", err) - } + // Save writes to disk; run it off the Run() goroutine so a + // 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": h.monotonicTS = cmd.enabled monoMsg, _ := json.Marshal(map[string]any{"type": "monotonicState", "enabled": h.monotonicTS}) diff --git a/Common/Client/go/wshub/hub_calibration_test.go b/Common/Client/go/wshub/hub_calibration_test.go new file mode 100644 index 0000000..d9ccf4f --- /dev/null +++ b/Common/Client/go/wshub/hub_calibration_test.go @@ -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) }