major improvements for big apps
This commit is contained in:
Martino Ferrari
+188
-21
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/binary"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"math"
|
||||
"net"
|
||||
"strings"
|
||||
@@ -67,8 +68,13 @@ type MarteClient struct {
|
||||
|
||||
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
|
||||
@@ -144,6 +150,7 @@ func (m *MarteClient) Disconnect() {
|
||||
m.basesMu.Lock()
|
||||
m.baseTsSet = false
|
||||
m.basesMu.Unlock()
|
||||
m.discoverAcc = nil
|
||||
}
|
||||
|
||||
func (m *MarteClient) stopped() bool {
|
||||
@@ -180,6 +187,8 @@ func (m *MarteClient) runTCP(host string, port int) {
|
||||
// Send SERVICE_INFO to auto-discover ports
|
||||
m.writeCmd("SERVICE_INFO")
|
||||
|
||||
go m.runKeepalive()
|
||||
|
||||
m.readLoop(conn)
|
||||
|
||||
atomic.StoreInt32(&m.connected, 0)
|
||||
@@ -205,8 +214,35 @@ func (m *MarteClient) writeCmd(cmd string) {
|
||||
if w == nil {
|
||||
return
|
||||
}
|
||||
log.Printf("[→MARTe] %s", cmd)
|
||||
broadcast(m.hub, 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 no other command has been written
|
||||
// in the last 20 s, keeping the DebugService 30 s idle timer from firing.
|
||||
func (m *MarteClient) 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")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *MarteClient) SendCommand(cmd string) {
|
||||
@@ -215,7 +251,7 @@ func (m *MarteClient) SendCommand(cmd string) {
|
||||
|
||||
func (m *MarteClient) readLoop(conn net.Conn) {
|
||||
scanner := bufio.NewScanner(conn)
|
||||
scanner.Buffer(make([]byte, 1*1024*1024), 1*1024*1024)
|
||||
scanner.Buffer(make([]byte, 8*1024*1024), 8*1024*1024)
|
||||
|
||||
var jsonAcc strings.Builder
|
||||
inJSON := false
|
||||
@@ -256,15 +292,18 @@ func (m *MarteClient) readLoop(conn net.Conn) {
|
||||
}
|
||||
}
|
||||
|
||||
// detectJSONDone returns (tag, true) when the line contains an "OK <TAG>" sentinel.
|
||||
// detectJSONDone returns (tag, true) when the line is exactly "OK <TAG>".
|
||||
// DISCOVER_PART must appear before DISCOVER so partial chunks are not mistaken
|
||||
// for the final chunk (though with exact matching this is not strictly
|
||||
// necessary; kept for clarity).
|
||||
func detectJSONDone(line string) (string, bool) {
|
||||
tags := []string{
|
||||
"DISCOVER", "TREE", "INFO", "CONFIG", "STEP_STATUS",
|
||||
"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 strings.Contains(line, "OK "+t) {
|
||||
if line == "OK "+t {
|
||||
return t, true
|
||||
}
|
||||
}
|
||||
@@ -272,11 +311,47 @@ func detectJSONDone(line string) (string, bool) {
|
||||
}
|
||||
|
||||
func (m *MarteClient) handleJSONResponse(tag, data string) {
|
||||
log.Printf("[←MARTe] %s %d bytes", tag, len(data))
|
||||
broadcast(m.hub, 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":
|
||||
// Accumulate this chunk; do NOT broadcast — we wait for the final chunk.
|
||||
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":
|
||||
m.parseDiscover(data)
|
||||
// Merge any previously accumulated part-chunks with this final chunk.
|
||||
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)
|
||||
// Re-marshal the merged list so the browser gets a single consistent blob.
|
||||
merged, _ := json.Marshal(discoverResp{Signals: all})
|
||||
broadcast(m.hub, map[string]any{
|
||||
"type": "response", "tag": "DISCOVER", "data": string(merged),
|
||||
})
|
||||
return
|
||||
|
||||
case "TREE":
|
||||
// Parse server-side and stream pre-digested flat-node batches to the
|
||||
// browser so it never has to JSON.parse a multi-MB string.
|
||||
go m.sendTreeBatches(data)
|
||||
return
|
||||
}
|
||||
// Forward raw JSON to all browsers.
|
||||
broadcast(m.hub, map[string]any{
|
||||
"type": "response",
|
||||
"tag": tag,
|
||||
@@ -284,6 +359,91 @@ func (m *MarteClient) handleJSONResponse(tag, data string) {
|
||||
})
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// TREE streaming
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// TreeNode mirrors the DebugService TREE JSON structure.
|
||||
type TreeNode struct {
|
||||
Name string `json:"Name"`
|
||||
Class string `json:"Class"`
|
||||
IsTraceable bool `json:"IsTraceable"`
|
||||
IsForcable bool `json:"IsForcable"`
|
||||
Elements uint32 `json:"Elements"`
|
||||
Children []*TreeNode `json:"Children"`
|
||||
}
|
||||
|
||||
// FlatNode is a pre-digested, compact representation sent to the browser.
|
||||
type FlatNode struct {
|
||||
Path string `json:"p"`
|
||||
Name string `json:"n"`
|
||||
Class string `json:"c"`
|
||||
Parent string `json:"par"`
|
||||
Tr bool `json:"tr,omitempty"`
|
||||
Fo bool `json:"fo,omitempty"`
|
||||
El uint32 `json:"el,omitempty"`
|
||||
HasCh bool `json:"ch,omitempty"`
|
||||
}
|
||||
|
||||
func flattenTree(node *TreeNode, parentPath string, out *[]FlatNode) {
|
||||
path := node.Name
|
||||
if parentPath != "" {
|
||||
path = parentPath + "." + node.Name
|
||||
}
|
||||
hasCh := len(node.Children) > 0
|
||||
*out = append(*out, FlatNode{
|
||||
Path: path,
|
||||
Name: node.Name,
|
||||
Class: node.Class,
|
||||
Parent: parentPath,
|
||||
Tr: node.IsTraceable,
|
||||
Fo: node.IsForcable,
|
||||
El: node.Elements,
|
||||
HasCh: hasCh,
|
||||
})
|
||||
for _, c := range node.Children {
|
||||
flattenTree(c, path, out)
|
||||
}
|
||||
}
|
||||
|
||||
// sendTreeBatches parses the raw TREE JSON and streams compact FlatNode
|
||||
// batches to all browsers. Each browser processes one batch per animation
|
||||
// frame, keeping the UI responsive during large tree loads.
|
||||
func (m *MarteClient) sendTreeBatches(raw string) {
|
||||
var root TreeNode
|
||||
if err := json.Unmarshal([]byte(raw), &root); err != nil {
|
||||
log.Printf("[TREE] parse error: %v — falling back to raw JSON", err)
|
||||
broadcast(m.hub, map[string]any{"type": "response", "tag": "TREE", "data": raw})
|
||||
return
|
||||
}
|
||||
|
||||
var nodes []FlatNode
|
||||
if root.Name == "Root" && len(root.Children) > 0 {
|
||||
for _, c := range root.Children {
|
||||
flattenTree(c, "", &nodes)
|
||||
}
|
||||
} else {
|
||||
flattenTree(&root, "", &nodes)
|
||||
}
|
||||
|
||||
total := len(nodes)
|
||||
log.Printf("[TREE] streaming %d flat nodes to browser", total)
|
||||
broadcast(m.hub, map[string]any{"type": "tree_clear", "total": total})
|
||||
|
||||
const batchSize = 50
|
||||
for i := 0; i < total; i += batchSize {
|
||||
end := i + batchSize
|
||||
if end > total {
|
||||
end = total
|
||||
}
|
||||
broadcast(m.hub, map[string]any{
|
||||
"type": "tree_batch",
|
||||
"nodes": nodes[i:end],
|
||||
"done": end == total,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (m *MarteClient) handleTextLine(line string) {
|
||||
if strings.HasPrefix(line, "OK SERVICE_INFO") {
|
||||
// OK SERVICE_INFO TCP_CTRL:8110 UDP_STREAM:8111 TCP_LOG:8082 STATE:RUNNING
|
||||
@@ -340,29 +500,35 @@ func (m *MarteClient) handleTextLine(line string) {
|
||||
// DISCOVER parsing
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
type discoverResp struct {
|
||||
Signals []struct {
|
||||
Name string `json:"name"`
|
||||
ID uint32 `json:"id"`
|
||||
Type string `json:"type"`
|
||||
Dimensions uint8 `json:"dimensions"`
|
||||
Elements uint32 `json:"elements"`
|
||||
} `json:"Signals"`
|
||||
type discoverSignalJSON struct {
|
||||
Name string `json:"name"`
|
||||
ID uint32 `json:"id"`
|
||||
Type string `json:"type"`
|
||||
Dimensions uint8 `json:"dimensions"`
|
||||
Elements uint32 `json:"elements"`
|
||||
}
|
||||
|
||||
func (m *MarteClient) parseDiscover(data string) {
|
||||
var resp discoverResp
|
||||
if err := json.Unmarshal([]byte(data), &resp); err != nil {
|
||||
return
|
||||
}
|
||||
type discoverResp struct {
|
||||
Signals []discoverSignalJSON `json:"Signals"`
|
||||
}
|
||||
|
||||
func (m *MarteClient) parseDiscoverSignals(sigs []discoverSignalJSON) {
|
||||
m.sigMu.Lock()
|
||||
defer m.sigMu.Unlock()
|
||||
m.signals = make(map[uint32]*SignalMeta, len(resp.Signals))
|
||||
for _, s := range resp.Signals {
|
||||
m.signals = make(map[uint32]*SignalMeta, len(sigs))
|
||||
for _, s := range sigs {
|
||||
el := s.Elements
|
||||
if el == 0 {
|
||||
el = 1
|
||||
}
|
||||
// Multiple DISCOVER entries can share the same ID (canonical DataSource
|
||||
// path + one GAM alias per broker). Merge them into one SignalMeta so
|
||||
// that Names carries every alias. This is required for the telemetry
|
||||
// receiver to match whichever name the user traced.
|
||||
if existing, ok := m.signals[s.ID]; ok {
|
||||
existing.Names = append(existing.Names, s.Name)
|
||||
continue
|
||||
}
|
||||
meta := &SignalMeta{
|
||||
Name: s.Name,
|
||||
ID: s.ID,
|
||||
@@ -373,6 +539,7 @@ func (m *MarteClient) parseDiscover(data string) {
|
||||
}
|
||||
m.signals[s.ID] = meta
|
||||
}
|
||||
log.Printf("[DISCOVER] registered %d unique signals from %d entries", len(m.signals), len(sigs))
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user