Martino Ferrari
269b2c4d97
Add a sink func(v any) field so MarteController's event stream can be routed somewhere other than the browser WebSocket hub. NewMarteController now sets sink to broadcast through the hub as before; a new NewHeadlessMarteController(sink) constructor builds an instance with hub == nil for the upcoming debugclient E2E tool. Direct m.hub.* calls (SetSourceState/UpdateConfigForSource/PushDataForSource) are now guarded with nil checks so a headless controller doesn't panic.
1018 lines
27 KiB
Go
1018 lines
27 KiB
Go
package main
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"net"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"syscall"
|
|
"time"
|
|
|
|
"marte2/common/udpsprotocol"
|
|
"marte2/common/wshub"
|
|
)
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Command safety gate (CR-4)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
// dangerousCommandsEnabled gates commands that mutate the RT application state
|
|
// (FORCE, PAUSE, RESUME, STEP, BREAK, MSG). Set via --enable-dangerous-commands.
|
|
var dangerousCommandsEnabled = false
|
|
|
|
// dangerousCommands is the set of MARTe2 commands that can change signal values
|
|
// or alter execution flow. Without --enable-dangerous-commands these are blocked
|
|
// from the browser WebSocket path.
|
|
var dangerousCommands = map[string]bool{
|
|
"FORCE": true,
|
|
"UNFORCE": true,
|
|
"PAUSE": true,
|
|
"RESUME": true,
|
|
"STEP": true,
|
|
"BREAK": true,
|
|
"UNBREAK": true,
|
|
"MSG": true,
|
|
"LOAD": true,
|
|
"UNLOAD": true,
|
|
}
|
|
|
|
// isDangerousCommand returns true if the command's first word is in the
|
|
// dangerous set (case-insensitive).
|
|
func isDangerousCommand(cmd string) bool {
|
|
parts := strings.Fields(cmd)
|
|
if len(parts) == 0 {
|
|
return false
|
|
}
|
|
return dangerousCommands[strings.ToUpper(parts[0])]
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Signal metadata (populated by DISCOVER)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
type SignalMeta struct {
|
|
Name string `json:"name"`
|
|
ID uint32 `json:"id"`
|
|
Type string `json:"type"`
|
|
Dimensions uint8 `json:"dimensions"`
|
|
Elements uint32 `json:"elements"`
|
|
Names []string // canonical + alias names mapping to this ID
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Outbound broadcast helpers
|
|
// ---------------------------------------------------------------------------
|
|
|
|
func broadcastHub(hub *wshub.Hub, v any) {
|
|
b, err := json.Marshal(v)
|
|
if err != nil {
|
|
return
|
|
}
|
|
hub.Broadcast(b)
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// MarteController
|
|
// ---------------------------------------------------------------------------
|
|
|
|
type MarteController struct {
|
|
hub *wshub.Hub
|
|
sink func(v any)
|
|
|
|
mu sync.Mutex
|
|
tcpConn net.Conn
|
|
writer *bufio.Writer
|
|
|
|
cmdMu sync.Mutex // serialise TCP writes
|
|
|
|
sigMu sync.RWMutex
|
|
signals map[uint32]*SignalMeta // id -> meta
|
|
tracedMu sync.RWMutex
|
|
tracedNames map[string]bool // user-visible names currently being traced
|
|
|
|
// Persistent forced-signal state: replayed to new browser clients so the
|
|
// forced-signals panel stays correct across page reloads.
|
|
forcedMu sync.RWMutex
|
|
forcedState map[string]string // signal key (may include [i]) → forced value
|
|
|
|
baseTsSet bool
|
|
basesMu sync.Mutex
|
|
|
|
connected int32 // atomic bool
|
|
|
|
lastWriteMs int64 // atomic; updated on every TCP write, used by keepalive
|
|
|
|
stopCh chan struct{}
|
|
|
|
// accumulates signals across DISCOVER_PART chunks; merged on final DISCOVER
|
|
discoverAcc []discoverSignalJSON
|
|
|
|
// cached last-known ports (updated by SERVICE_INFO)
|
|
host string
|
|
cmdPort int
|
|
udpPort int
|
|
logPort int
|
|
|
|
// rawSigs holds the most-recently received UDP CONFIG signals before
|
|
// translation. After DISCOVER populates m.signals, we re-translate and
|
|
// push a corrected config to the hub so that src.signals names match the
|
|
// s.Values keys used by ParseData (which also re-translates rawSigs).
|
|
rawSigsMu sync.RWMutex
|
|
rawSigs []udpsprotocol.SignalInfo
|
|
rawPm uint8
|
|
}
|
|
|
|
// NewMarteController creates a MarteController bound to the given hub.
|
|
func NewMarteController(hub *wshub.Hub) *MarteController {
|
|
mc := &MarteController{
|
|
hub: hub,
|
|
signals: make(map[uint32]*SignalMeta),
|
|
tracedNames: make(map[string]bool),
|
|
forcedState: make(map[string]string),
|
|
stopCh: make(chan struct{}),
|
|
}
|
|
mc.sink = func(v any) { broadcastHub(mc.hub, v) }
|
|
// Register the new-client hook so connection + forced/traced state is
|
|
// replayed to any browser that connects (or reconnects) while the server
|
|
// already holds a live MARTe2 TCP session.
|
|
hub.SetOnClientConnect(mc.replayStateToClient)
|
|
return mc
|
|
}
|
|
|
|
// NewHeadlessMarteController creates a MarteController with no WebSocket hub,
|
|
// routing all events through sink instead (used by the debugclient E2E test tool).
|
|
func NewHeadlessMarteController(sink func(v any)) *MarteController {
|
|
mc := &MarteController{
|
|
hub: nil,
|
|
signals: make(map[uint32]*SignalMeta),
|
|
tracedNames: make(map[string]bool),
|
|
forcedState: make(map[string]string),
|
|
stopCh: make(chan struct{}),
|
|
}
|
|
mc.sink = sink
|
|
return mc
|
|
}
|
|
|
|
func (m *MarteController) IsConnected() bool {
|
|
return atomic.LoadInt32(&m.connected) == 1
|
|
}
|
|
|
|
// replayStateToClient is called by the hub whenever a new WebSocket client
|
|
// connects. It sends the current MARTe2 connection status and any persistent
|
|
// forced/traced signal state so the browser UI is always consistent.
|
|
func (m *MarteController) replayStateToClient(send func([]byte)) {
|
|
if !m.IsConnected() {
|
|
return
|
|
}
|
|
// Re-send "connected" so the browser updates its UI state.
|
|
if b, err := json.Marshal(map[string]any{"type": "connected"}); err == nil {
|
|
send(b)
|
|
}
|
|
|
|
// Replay forced signals.
|
|
m.forcedMu.RLock()
|
|
forced := make(map[string]string, len(m.forcedState))
|
|
for k, v := range m.forcedState {
|
|
forced[k] = v
|
|
}
|
|
m.forcedMu.RUnlock()
|
|
if len(forced) > 0 {
|
|
if b, err := json.Marshal(map[string]any{"type": "forced_state", "signals": forced}); err == nil {
|
|
send(b)
|
|
}
|
|
}
|
|
|
|
// Replay traced signal names.
|
|
m.tracedMu.RLock()
|
|
traced := make([]string, 0, len(m.tracedNames))
|
|
for n := range m.tracedNames {
|
|
traced = append(traced, n)
|
|
}
|
|
m.tracedMu.RUnlock()
|
|
if len(traced) > 0 {
|
|
if b, err := json.Marshal(map[string]any{"type": "traced_state", "signals": traced}); err == nil {
|
|
send(b)
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Connect / Disconnect
|
|
// ---------------------------------------------------------------------------
|
|
|
|
func (m *MarteController) Connect(host string, cmdPort, udpPort, logPort int) {
|
|
m.Disconnect()
|
|
|
|
m.mu.Lock()
|
|
m.host = host
|
|
m.cmdPort = cmdPort
|
|
m.udpPort = udpPort
|
|
m.logPort = logPort
|
|
m.stopCh = make(chan struct{})
|
|
m.mu.Unlock()
|
|
|
|
// Update source state so the browser shows "connecting". No-op headless
|
|
// (m.hub == nil for NewHeadlessMarteController instances).
|
|
if m.hub != nil {
|
|
m.hub.SetSourceState("debug", "connecting")
|
|
}
|
|
|
|
m.sink(map[string]any{
|
|
"type": "log", "time": time.Now().Format("15:04:05.000"),
|
|
"level": "INFO", "message": fmt.Sprintf("Connecting to %s cmd=%d udp=%d log=%d", host, cmdPort, udpPort, logPort),
|
|
})
|
|
|
|
go m.runTCP(host, cmdPort)
|
|
go m.runDebugUDP(host, udpPort)
|
|
go m.runLog(host, logPort)
|
|
}
|
|
|
|
func (m *MarteController) Disconnect() {
|
|
m.mu.Lock()
|
|
select {
|
|
case <-m.stopCh:
|
|
// already closed
|
|
default:
|
|
close(m.stopCh)
|
|
}
|
|
if m.tcpConn != nil {
|
|
m.tcpConn.Close()
|
|
m.tcpConn = nil
|
|
}
|
|
m.mu.Unlock()
|
|
|
|
atomic.StoreInt32(&m.connected, 0)
|
|
m.basesMu.Lock()
|
|
m.baseTsSet = false
|
|
m.basesMu.Unlock()
|
|
m.discoverAcc = nil
|
|
if m.hub != nil {
|
|
m.hub.SetSourceState("debug", "disconnected")
|
|
}
|
|
}
|
|
|
|
func (m *MarteController) stopped() bool {
|
|
select {
|
|
case <-m.stopCh:
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// HandleBrowserCommand — dispatch JSON commands from browser
|
|
// ---------------------------------------------------------------------------
|
|
|
|
func (m *MarteController) HandleBrowserCommand(msg []byte) {
|
|
var env map[string]interface{}
|
|
if err := json.Unmarshal(msg, &env); err != nil {
|
|
return
|
|
}
|
|
t, _ := env["type"].(string)
|
|
switch t {
|
|
case "connect":
|
|
data, _ := env["data"].(map[string]interface{})
|
|
if data == nil {
|
|
return
|
|
}
|
|
host, _ := data["host"].(string)
|
|
portF, _ := data["port"].(float64)
|
|
udpPortF, _ := data["udp_port"].(float64)
|
|
logPortF, _ := data["log_port"].(float64)
|
|
if host == "" {
|
|
host = "127.0.0.1"
|
|
}
|
|
port := int(portF)
|
|
if port == 0 {
|
|
port = 8080
|
|
}
|
|
udpPort := int(udpPortF)
|
|
if udpPort == 0 {
|
|
udpPort = port + 1
|
|
}
|
|
logPort := int(logPortF)
|
|
if logPort == 0 {
|
|
logPort = port + 2
|
|
}
|
|
m.Connect(host, port, udpPort, logPort)
|
|
|
|
case "disconnect":
|
|
m.Disconnect()
|
|
|
|
case "cmd":
|
|
data, _ := env["data"].(map[string]interface{})
|
|
if data == nil {
|
|
return
|
|
}
|
|
cmd, _ := data["cmd"].(string)
|
|
if cmd == "" {
|
|
return
|
|
}
|
|
// Gate dangerous commands (FORCE/UNFORCE/PAUSE/RESUME/STEP/BREAK/MSG)
|
|
// behind an explicit opt-in flag. Without it, only read-only commands
|
|
// (DISCOVER, TREE, INFO, LS, VALUE, TRACE, UNTRACE, STEP_STATUS) are
|
|
// forwarded to the MARTe2 TCP control connection.
|
|
if isDangerousCommand(cmd) {
|
|
if !dangerousCommandsEnabled {
|
|
m.sink(map[string]any{
|
|
"type": "log", "time": time.Now().Format("15:04:05.000"),
|
|
"level": "WARNING",
|
|
"message": fmt.Sprintf("Blocked dangerous command (requires --enable-dangerous-commands): %s", cmd),
|
|
})
|
|
return
|
|
}
|
|
}
|
|
m.trackForcedCmd(cmd)
|
|
m.SendCommand(cmd)
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// TCP command channel
|
|
// ---------------------------------------------------------------------------
|
|
|
|
func (m *MarteController) runTCP(host string, port int) {
|
|
addr := fmt.Sprintf("%s:%d", host, port)
|
|
for !m.stopped() {
|
|
conn, err := net.DialTimeout("tcp", addr, 5*time.Second)
|
|
if err != nil {
|
|
m.sink(map[string]any{
|
|
"type": "log", "time": time.Now().Format("15:04:05.000"),
|
|
"level": "WARNING", "message": fmt.Sprintf("TCP %s: %v — retrying…", addr, err),
|
|
})
|
|
time.Sleep(2 * time.Second)
|
|
continue
|
|
}
|
|
conn.(*net.TCPConn).SetNoDelay(true)
|
|
|
|
m.mu.Lock()
|
|
m.tcpConn = conn
|
|
m.writer = bufio.NewWriter(conn)
|
|
m.mu.Unlock()
|
|
|
|
atomic.StoreInt32(&m.connected, 1)
|
|
m.sink(map[string]any{"type": "connected"})
|
|
|
|
// Send SERVICE_INFO to auto-discover ports
|
|
m.writeCmd("SERVICE_INFO")
|
|
|
|
go m.runKeepalive()
|
|
|
|
m.readLoop(conn)
|
|
|
|
atomic.StoreInt32(&m.connected, 0)
|
|
m.sink(map[string]any{"type": "disconnected"})
|
|
|
|
m.mu.Lock()
|
|
m.tcpConn = nil
|
|
m.writer = nil
|
|
m.mu.Unlock()
|
|
|
|
if !m.stopped() {
|
|
time.Sleep(2 * time.Second)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (m *MarteController) writeCmd(cmd string) {
|
|
m.cmdMu.Lock()
|
|
defer m.cmdMu.Unlock()
|
|
m.mu.Lock()
|
|
w := m.writer
|
|
m.mu.Unlock()
|
|
if w == nil {
|
|
return
|
|
}
|
|
// Suppress high-frequency polling commands from both terminal and browser logs.
|
|
silent := cmd == "STEP_STATUS" || cmd == "INFO"
|
|
if !silent {
|
|
log.Printf("[→MARTe] %s", cmd)
|
|
m.sink(map[string]any{
|
|
"type": "log", "time": time.Now().Format("15:04:05.000"),
|
|
"level": "CMD", "message": fmt.Sprintf("→ %s", cmd),
|
|
})
|
|
}
|
|
w.WriteString(cmd + "\n")
|
|
w.Flush()
|
|
atomic.StoreInt64(&m.lastWriteMs, time.Now().UnixMilli())
|
|
}
|
|
|
|
// runKeepalive sends INFO every 20 s when idle.
|
|
func (m *MarteController) runKeepalive() {
|
|
ticker := time.NewTicker(20 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-m.stopCh:
|
|
return
|
|
case <-ticker.C:
|
|
if !m.IsConnected() {
|
|
continue
|
|
}
|
|
idleMs := time.Now().UnixMilli() - atomic.LoadInt64(&m.lastWriteMs)
|
|
if idleMs >= 20_000 {
|
|
m.writeCmd("INFO")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// trackForcedCmd intercepts FORCE/UNFORCE commands to maintain the persistent
|
|
// forced-signal state that is replayed to new browser clients.
|
|
func (m *MarteController) trackForcedCmd(cmd string) {
|
|
parts := strings.Fields(cmd)
|
|
if len(parts) < 2 {
|
|
return
|
|
}
|
|
switch strings.ToUpper(parts[0]) {
|
|
case "FORCE":
|
|
if len(parts) >= 3 {
|
|
key := parts[1]
|
|
val := strings.Join(parts[2:], " ")
|
|
m.forcedMu.Lock()
|
|
m.forcedState[key] = val
|
|
m.forcedMu.Unlock()
|
|
}
|
|
case "UNFORCE":
|
|
key := parts[1]
|
|
m.forcedMu.Lock()
|
|
// Remove the exact key and any element keys that share the same base name
|
|
// (e.g., UNFORCE Foo removes Foo, Foo[0], Foo[1], …).
|
|
delete(m.forcedState, key)
|
|
prefix := key + "["
|
|
for k := range m.forcedState {
|
|
if strings.HasPrefix(k, prefix) {
|
|
delete(m.forcedState, k)
|
|
}
|
|
}
|
|
m.forcedMu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (m *MarteController) SendCommand(cmd string) {
|
|
// Track TRACE enable/disable so translateSignalNames can pick the right alias.
|
|
if strings.HasPrefix(cmd, "TRACE ") {
|
|
parts := strings.Fields(cmd)
|
|
if len(parts) == 3 {
|
|
name := parts[1]
|
|
enable := parts[2] == "1"
|
|
m.tracedMu.Lock()
|
|
if enable {
|
|
m.tracedNames[name] = true
|
|
} else {
|
|
delete(m.tracedNames, name)
|
|
}
|
|
m.tracedMu.Unlock()
|
|
}
|
|
}
|
|
m.writeCmd(cmd)
|
|
}
|
|
|
|
func (m *MarteController) readLoop(conn net.Conn) {
|
|
scanner := bufio.NewScanner(conn)
|
|
scanner.Buffer(make([]byte, 8*1024*1024), 8*1024*1024)
|
|
|
|
var jsonAcc strings.Builder
|
|
inJSON := false
|
|
|
|
for scanner.Scan() {
|
|
if m.stopped() {
|
|
return
|
|
}
|
|
line := scanner.Text()
|
|
trimmed := strings.TrimSpace(line)
|
|
if trimmed == "" {
|
|
continue
|
|
}
|
|
|
|
// Detect start of JSON block
|
|
if !inJSON && strings.HasPrefix(trimmed, "{") {
|
|
inJSON = true
|
|
jsonAcc.Reset()
|
|
}
|
|
|
|
if inJSON {
|
|
jsonAcc.WriteString(trimmed)
|
|
tag, done := detectJSONDone(trimmed)
|
|
if done {
|
|
inJSON = false
|
|
raw := jsonAcc.String()
|
|
// Strip trailing sentinel
|
|
idx := strings.Index(raw, "OK "+tag)
|
|
if idx >= 0 {
|
|
raw = strings.TrimSpace(raw[:idx])
|
|
}
|
|
m.handleJSONResponse(tag, raw)
|
|
jsonAcc.Reset()
|
|
}
|
|
} else {
|
|
m.handleTextLine(trimmed)
|
|
}
|
|
}
|
|
}
|
|
|
|
func detectJSONDone(line string) (string, bool) {
|
|
tags := []string{
|
|
"DISCOVER_PART", "DISCOVER", "TREE", "INFO", "CONFIG", "STEP_STATUS",
|
|
"VALUE", "MSG", "TRACE", "FORCE", "UNFORCE", "BREAK",
|
|
"PAUSE", "RESUME", "STEP", "MONITOR", "UNMONITOR", "LS",
|
|
}
|
|
for _, t := range tags {
|
|
if line == "OK "+t {
|
|
return t, true
|
|
}
|
|
}
|
|
return "", false
|
|
}
|
|
|
|
func (m *MarteController) handleJSONResponse(tag, data string) {
|
|
silent := tag == "STEP_STATUS" || tag == "INFO"
|
|
if !silent {
|
|
log.Printf("[←MARTe] %s %d bytes", tag, len(data))
|
|
m.sink(map[string]any{
|
|
"type": "log", "time": time.Now().Format("15:04:05.000"),
|
|
"level": "RESP", "message": fmt.Sprintf("← %s (%d B)", tag, len(data)),
|
|
})
|
|
}
|
|
switch tag {
|
|
case "DISCOVER_PART":
|
|
var resp discoverResp
|
|
if err := json.Unmarshal([]byte(data), &resp); err != nil {
|
|
log.Printf("[DISCOVER_PART] parse error: %v", err)
|
|
return
|
|
}
|
|
m.discoverAcc = append(m.discoverAcc, resp.Signals...)
|
|
log.Printf("[DISCOVER_PART] accumulated %d signals (total so far: %d)",
|
|
len(resp.Signals), len(m.discoverAcc))
|
|
return
|
|
|
|
case "DISCOVER":
|
|
var resp discoverResp
|
|
if err := json.Unmarshal([]byte(data), &resp); err != nil {
|
|
log.Printf("[DISCOVER] parse error: %v", err)
|
|
return
|
|
}
|
|
all := append(m.discoverAcc, resp.Signals...)
|
|
m.discoverAcc = nil
|
|
m.parseDiscoverSignals(all)
|
|
// Update hub config now that m.signals is populated for name translation.
|
|
// If UDP CONFIG was already received (rawSigs non-empty), re-translate the
|
|
// real config so that src.signals uses the same names as s.Values produced
|
|
// by ParseData (which also re-translates rawSigs on every PktData).
|
|
// If UDP CONFIG has not arrived yet, fall back to synthesizing from DISCOVER.
|
|
m.rawSigsMu.RLock()
|
|
raw := m.rawSigs
|
|
m.rawSigsMu.RUnlock()
|
|
if len(raw) > 0 {
|
|
if m.hub != nil {
|
|
m.hub.UpdateConfigForSource("debug", m.translateSignalNames(raw))
|
|
}
|
|
} else {
|
|
m.synthesizeHubConfig(all)
|
|
}
|
|
// Re-marshal the merged list so the browser gets a single consistent blob.
|
|
merged, _ := json.Marshal(discoverResp{Signals: all})
|
|
m.sink(map[string]any{
|
|
"type": "response", "tag": "DISCOVER", "data": string(merged),
|
|
})
|
|
return
|
|
|
|
case "TREE":
|
|
m.sink(map[string]any{
|
|
"type": "tree_node",
|
|
"data": data,
|
|
})
|
|
return
|
|
}
|
|
m.sink(map[string]any{
|
|
"type": "response",
|
|
"tag": tag,
|
|
"data": data,
|
|
})
|
|
}
|
|
|
|
func (m *MarteController) handleTextLine(line string) {
|
|
if strings.HasPrefix(line, "OK SERVICE_INFO") {
|
|
parts := strings.Fields(line)
|
|
newUDP, newLog := 0, 0
|
|
for _, p := range parts {
|
|
if strings.HasPrefix(p, "UDP_STREAM:") {
|
|
fmt.Sscanf(p[11:], "%d", &newUDP)
|
|
}
|
|
if strings.HasPrefix(p, "TCP_LOG:") {
|
|
fmt.Sscanf(p[8:], "%d", &newLog)
|
|
}
|
|
}
|
|
m.sink(map[string]any{
|
|
"type": "response",
|
|
"tag": "SERVICE_INFO",
|
|
"data": line[len("OK SERVICE_INFO "):],
|
|
})
|
|
if newUDP > 0 || newLog > 0 {
|
|
m.sink(map[string]any{
|
|
"type": "service_config",
|
|
"udp_port": newUDP,
|
|
"log_port": newLog,
|
|
})
|
|
m.mu.Lock()
|
|
host := m.host
|
|
oldUDP := m.udpPort
|
|
oldLog := m.logPort
|
|
if newUDP > 0 {
|
|
m.udpPort = newUDP
|
|
}
|
|
if newLog > 0 {
|
|
m.logPort = newLog
|
|
}
|
|
m.mu.Unlock()
|
|
if newUDP > 0 && newUDP != oldUDP {
|
|
go m.runDebugUDP(host, newUDP)
|
|
}
|
|
if newLog > 0 && newLog != oldLog {
|
|
go m.runLog(host, newLog)
|
|
}
|
|
}
|
|
}
|
|
m.sink(map[string]any{
|
|
"type": "text_line",
|
|
"data": line,
|
|
})
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// DISCOVER parsing
|
|
// ---------------------------------------------------------------------------
|
|
|
|
type discoverSignalJSON struct {
|
|
Name string `json:"name"`
|
|
ID uint32 `json:"id"`
|
|
Type string `json:"type"`
|
|
Dimensions uint8 `json:"dimensions"`
|
|
Elements uint32 `json:"elements"`
|
|
}
|
|
|
|
type discoverResp struct {
|
|
Signals []discoverSignalJSON `json:"Signals"`
|
|
}
|
|
|
|
func (m *MarteController) parseDiscoverSignals(sigs []discoverSignalJSON) {
|
|
m.sigMu.Lock()
|
|
defer m.sigMu.Unlock()
|
|
m.signals = make(map[uint32]*SignalMeta, len(sigs))
|
|
for _, s := range sigs {
|
|
el := s.Elements
|
|
if el == 0 {
|
|
el = 1
|
|
}
|
|
if existing, ok := m.signals[s.ID]; ok {
|
|
existing.Names = append(existing.Names, s.Name)
|
|
continue
|
|
}
|
|
meta := &SignalMeta{
|
|
Name: s.Name,
|
|
ID: s.ID,
|
|
Type: s.Type,
|
|
Dimensions: s.Dimensions,
|
|
Elements: el,
|
|
Names: []string{s.Name},
|
|
}
|
|
m.signals[s.ID] = meta
|
|
}
|
|
log.Printf("[DISCOVER] registered %d unique signals from %d entries", len(m.signals), len(sigs))
|
|
}
|
|
|
|
// translateSignalNames maps UDPS signal names (DS canonical, e.g. "DDB1.Sine1")
|
|
// to the preferred GAM-path alias (e.g. "App.Functions.SineGAM1.Out.Sine1").
|
|
//
|
|
// DebugService always sets signals[i]->name to the first registered name, which
|
|
// is the DataSource canonical path. The user-facing name in the tree and the
|
|
// tracedSet is the GAM alias (contains ".Out." or ".In."). Without this
|
|
// translation the buffer key created from UDPS CONFIG ("debug:DDB1.Sine1")
|
|
// never matches the buffer key the tracedTab looks up ("debug:App…Out.Sine1").
|
|
func (m *MarteController) translateSignalNames(sigs []udpsprotocol.SignalInfo) []udpsprotocol.SignalInfo {
|
|
m.sigMu.RLock()
|
|
defer m.sigMu.RUnlock()
|
|
|
|
if len(m.signals) == 0 {
|
|
return sigs // no DISCOVER data yet — return as-is
|
|
}
|
|
|
|
// Build a reverse map: every known alias name → SignalMeta
|
|
nameToMeta := make(map[string]*SignalMeta, len(m.signals)*2)
|
|
for _, meta := range m.signals {
|
|
for _, n := range meta.Names {
|
|
nameToMeta[n] = meta
|
|
}
|
|
}
|
|
|
|
// Snapshot the user's currently-traced names for matching.
|
|
m.tracedMu.RLock()
|
|
traced := make(map[string]bool, len(m.tracedNames))
|
|
for k, v := range m.tracedNames {
|
|
traced[k] = v
|
|
}
|
|
m.tracedMu.RUnlock()
|
|
|
|
result := make([]udpsprotocol.SignalInfo, len(sigs))
|
|
for i, sig := range sigs {
|
|
result[i] = sig
|
|
meta, ok := nameToMeta[sig.Name]
|
|
if !ok {
|
|
continue
|
|
}
|
|
best := sig.Name
|
|
// Priority 1: find a traced name that alias-matches this signal (mirrors
|
|
// C++ AliasMatch: exact or suffix in either direction, dot-boundary).
|
|
outer:
|
|
for _, n := range meta.Names {
|
|
for tracedName := range traced {
|
|
if aliasMatch(n, tracedName) {
|
|
best = tracedName // use the name the user sees in the tree
|
|
break outer
|
|
}
|
|
}
|
|
}
|
|
// Priority 2 (fallback): longest alias with ".Out." or ".In." (GAM path).
|
|
if best == sig.Name {
|
|
for _, n := range meta.Names {
|
|
if (strings.Contains(n, ".Out.") || strings.Contains(n, ".In.")) && len(n) > len(best) {
|
|
best = n
|
|
}
|
|
}
|
|
}
|
|
if best != sig.Name {
|
|
log.Printf("[debug-udp] rename %q → %q", sig.Name, best)
|
|
result[i].Name = best
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
// aliasMatch mirrors C++ AliasMatch: returns true if a and b are equal or one
|
|
// is a dot-boundary suffix of the other.
|
|
func aliasMatch(a, b string) bool {
|
|
if a == b {
|
|
return true
|
|
}
|
|
return suffixMatchDot(a, b) || suffixMatchDot(b, a)
|
|
}
|
|
|
|
func suffixMatchDot(str, suffix string) bool {
|
|
if len(suffix) > len(str) {
|
|
return false
|
|
}
|
|
tail := str[len(str)-len(suffix):]
|
|
if tail != suffix {
|
|
return false
|
|
}
|
|
return len(str) == len(suffix) || str[len(str)-len(suffix)-1] == '.'
|
|
}
|
|
|
|
// typeCodeFromString maps a MARTe2 type string to a UDPS type code.
|
|
func typeCodeFromString(t string) uint8 {
|
|
switch strings.ToLower(t) {
|
|
case "uint8":
|
|
return 0
|
|
case "int8":
|
|
return 1
|
|
case "uint16":
|
|
return 2
|
|
case "int16":
|
|
return 3
|
|
case "uint32":
|
|
return 4
|
|
case "int32":
|
|
return 5
|
|
case "uint64":
|
|
return 6
|
|
case "int64":
|
|
return 7
|
|
case "float32":
|
|
return 8
|
|
case "float64":
|
|
return 9
|
|
default:
|
|
return 4 // default to uint32
|
|
}
|
|
}
|
|
|
|
// synthesizeHubConfig converts DISCOVER signals to udpsprotocol.SignalInfo and
|
|
// sends them to the hub so the signal list panel is populated before UDP data arrives.
|
|
func (m *MarteController) synthesizeHubConfig(sigs []discoverSignalJSON) {
|
|
seen := make(map[uint32]bool)
|
|
var sigInfos []udpsprotocol.SignalInfo
|
|
for _, s := range sigs {
|
|
if seen[s.ID] {
|
|
continue
|
|
}
|
|
seen[s.ID] = true
|
|
el := s.Elements
|
|
if el == 0 {
|
|
el = 1
|
|
}
|
|
numRows := el
|
|
numCols := uint32(1)
|
|
if s.Dimensions >= 2 {
|
|
numCols = el
|
|
numRows = 1
|
|
}
|
|
si := udpsprotocol.SignalInfo{
|
|
Name: s.Name,
|
|
TypeCode: typeCodeFromString(s.Type),
|
|
QuantType: 0,
|
|
NumDimensions: s.Dimensions,
|
|
NumRows: numRows,
|
|
NumCols: numCols,
|
|
RangeMin: 0,
|
|
RangeMax: 0,
|
|
TimeMode: udpsprotocol.TimeModePacket,
|
|
SamplingRate: 0,
|
|
TimeSignalIdx: udpsprotocol.NoTimeSignal,
|
|
Unit: "",
|
|
}
|
|
sigInfos = append(sigInfos, si)
|
|
}
|
|
// Translate to GAM-path names so the synthesised config matches the names
|
|
// that the real UDP CONFIG (and data frames) use. Without this translation,
|
|
// synthesizeHubConfig would overwrite the real config with DISCOVER-canonical
|
|
// names (e.g. "DDB1.Ch3"), causing buildBinaryDataMessageForSource to fail to
|
|
// find signals in s.Values (which uses translated names), starving the JS
|
|
// buffer and limiting live streaming to the fraction of a second that
|
|
// accumulated before the DISCOVER response arrived.
|
|
translated := m.translateSignalNames(sigInfos)
|
|
if m.hub != nil {
|
|
m.hub.UpdateConfigForSource("debug", translated)
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Debug UDP receiver — receives UDPS packets from DebugService
|
|
// ---------------------------------------------------------------------------
|
|
|
|
func (m *MarteController) runDebugUDP(host string, port int) {
|
|
addr := fmt.Sprintf("0.0.0.0:%d", port)
|
|
|
|
// Use SO_REUSEPORT so we can bind the same port that DebugService already
|
|
// holds open. Without this the second bind fails with EADDRINUSE and we
|
|
// receive nothing.
|
|
lc := net.ListenConfig{
|
|
Control: func(network, address string, c syscall.RawConn) error {
|
|
return c.Control(func(fd uintptr) {
|
|
const SO_REUSEPORT = 0xf // Linux SO_REUSEPORT = 15
|
|
if err := syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, SO_REUSEPORT, 1); err != nil {
|
|
log.Printf("[debug-udp] SO_REUSEPORT: %v", err)
|
|
}
|
|
})
|
|
},
|
|
}
|
|
pc, err := lc.ListenPacket(context.Background(), "udp4", addr)
|
|
if err != nil {
|
|
msg := fmt.Sprintf("UDP bind on %s failed: %v — rebuild DebugService C++ and restart", addr, err)
|
|
log.Printf("[debug-udp] %s", msg)
|
|
m.sink(map[string]any{
|
|
"type": "log", "time": time.Now().Format("15:04:05.000"),
|
|
"level": "ERROR", "message": msg,
|
|
})
|
|
return
|
|
}
|
|
conn := pc.(*net.UDPConn)
|
|
defer conn.Close()
|
|
conn.SetReadBuffer(10 * 1024 * 1024)
|
|
|
|
log.Printf("[debug-udp] listening on %s for UDPS packets", addr)
|
|
m.sink(map[string]any{
|
|
"type": "log", "time": time.Now().Format("15:04:05.000"),
|
|
"level": "INFO", "message": fmt.Sprintf("UDP listener bound on %s", addr),
|
|
})
|
|
|
|
reassembler := udpsprotocol.NewReassembler(2 * time.Second)
|
|
buf := make([]byte, 65535)
|
|
var currentSigs []udpsprotocol.SignalInfo
|
|
var currentPublishMode uint8
|
|
var pktCount int64
|
|
|
|
for !m.stopped() {
|
|
conn.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
|
|
n, _, err := conn.ReadFromUDP(buf)
|
|
arrivalTime := time.Now()
|
|
if err != nil {
|
|
continue
|
|
}
|
|
pktCount++
|
|
|
|
if n < udpsprotocol.HeaderSize {
|
|
continue
|
|
}
|
|
|
|
hdr, err := udpsprotocol.ParseHeader(buf[:n])
|
|
if err != nil {
|
|
continue
|
|
}
|
|
|
|
payload := make([]byte, n-udpsprotocol.HeaderSize)
|
|
copy(payload, buf[udpsprotocol.HeaderSize:n])
|
|
|
|
complete, ok := reassembler.AddFragment(hdr, payload)
|
|
if !ok {
|
|
continue
|
|
}
|
|
|
|
switch hdr.Type {
|
|
case udpsprotocol.PktConfig:
|
|
sigs, pm, err := udpsprotocol.ParseConfig(complete)
|
|
if err != nil {
|
|
log.Printf("[debug-udp] parse config: %v", err)
|
|
continue
|
|
}
|
|
// Store raw (untranslated) sigs so that after DISCOVER populates
|
|
// m.signals we can re-translate and push a corrected config to the hub.
|
|
m.rawSigsMu.Lock()
|
|
m.rawSigs = sigs
|
|
m.rawPm = pm
|
|
m.rawSigsMu.Unlock()
|
|
sigs = m.translateSignalNames(sigs)
|
|
currentSigs = sigs
|
|
currentPublishMode = pm
|
|
if m.hub != nil {
|
|
m.hub.UpdateConfigForSource("debug", sigs)
|
|
m.hub.SetSourceState("debug", "connected")
|
|
}
|
|
|
|
case udpsprotocol.PktData:
|
|
if len(currentSigs) == 0 {
|
|
continue
|
|
}
|
|
// Re-translate on every data packet: if DISCOVER ran since the last
|
|
// UDP CONFIG arrived (which is the common case — CONFIG is sent once
|
|
// at startup, before the browser connects and triggers DISCOVER),
|
|
// m.signals is now populated and we must use translated names so that
|
|
// s.Values keys match src.signals keys in buildBinaryDataMessageForSource.
|
|
m.rawSigsMu.RLock()
|
|
raw := m.rawSigs
|
|
m.rawSigsMu.RUnlock()
|
|
if len(raw) > 0 {
|
|
currentSigs = m.translateSignalNames(raw)
|
|
}
|
|
samples, err := udpsprotocol.ParseData(complete, currentSigs, currentPublishMode, arrivalTime)
|
|
if err != nil {
|
|
log.Printf("[debug-udp] parse data: %v", err)
|
|
continue
|
|
}
|
|
if m.hub != nil {
|
|
for _, s := range samples {
|
|
m.hub.PushDataForSource("debug", s)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
log.Printf("[debug-udp] stopped")
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// TCP log channel
|
|
// ---------------------------------------------------------------------------
|
|
|
|
func (m *MarteController) runLog(host string, port int) {
|
|
addr := fmt.Sprintf("%s:%d", host, port)
|
|
for !m.stopped() {
|
|
conn, err := net.DialTimeout("tcp", addr, 5*time.Second)
|
|
if err != nil {
|
|
time.Sleep(2 * time.Second)
|
|
continue
|
|
}
|
|
scanner := bufio.NewScanner(conn)
|
|
for scanner.Scan() {
|
|
if m.stopped() {
|
|
conn.Close()
|
|
return
|
|
}
|
|
line := strings.TrimSpace(scanner.Text())
|
|
if strings.HasPrefix(line, "LOG ") {
|
|
rest := line[4:]
|
|
idx := strings.Index(rest, " ")
|
|
if idx < 0 {
|
|
continue
|
|
}
|
|
level := rest[:idx]
|
|
msg := rest[idx+1:]
|
|
m.sink(map[string]any{
|
|
"type": "log",
|
|
"time": time.Now().Format("15:04:05.000"),
|
|
"level": level,
|
|
"message": msg,
|
|
})
|
|
}
|
|
}
|
|
conn.Close()
|
|
if !m.stopped() {
|
|
time.Sleep(2 * time.Second)
|
|
}
|
|
}
|
|
}
|