Implemented backend hr resolution data, splitted test gam

Reorder Execute() main stage so the background thread waits on dataSem
(sleeping until the RT thread posts) before doing the socket select().
The socket poll is now non-blocking (timeout=0) since command latency
bounded by UDPS_DATA_WAIT_MS is acceptable for CONNECT/DISCONNECT.

Also force-remove old udpstreamer-webui binary in run.sh before rebuild
so stale embedded assets are never served.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,41 @@
+# ── Build output ─────────────────────────────────────────────────────────────
+Build/
+*.o
+*.a
+*.so
+*.ex
+*.ex.dbg
+
+# ── Symlinks created by run.sh for MARTe2 auto-loader ────────────────────────
+# (WaveformSin.so, SineArrayGAM.so, etc. → underlying .so)
+Source/**/WaveformSin.so
+Source/**/WaveformChirp.so
+Source/**/WaveformPointsDef.so
+Source/**/SineArrayGAM.so
+
+# ── Generated dependency files ────────────────────────────────────────────────
+depends.*
+dependsRaw.*
+
+# ── Code coverage ─────────────────────────────────────────────────────────────
+*.gcov
+*.gcno
+*.gcda
+Makefile.cov
+lcov_output/
+coverage/
+
+# ── Go binaries ───────────────────────────────────────────────────────────────
+Client/WebUI/udpstreamer-webui
+
+# ── IDE / editor ──────────────────────────────────────────────────────────────
+.idea/
+.vscode/
+*.swp
+*~
+*.orig
+*.bak
+
+# ── OS ────────────────────────────────────────────────────────────────────────
+.DS_Store
+Thumbs.db

Client/WebUI/hub.go

@@ -3,7 +3,10 @@ package main
 import (
 	"encoding/json"
 	"log"
+	"math"
 	"net/http"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -59,7 +62,6 @@ func (c *wsClient) readPump() {
 		if err != nil {
 			break
 		}
-		// Handle client messages (ping, setWindow, etc.) – currently just log.
 		var env map[string]interface{}
 		if json.Unmarshal(msg, &env) == nil {
 			if t, ok := env["type"].(string); ok {
@@ -70,6 +72,28 @@ func (c *wsClient) readPump() {
 					case c.send <- resp:
 					default:
 					}
+				case "addSource":
+					label, _ := env["label"].(string)
+					addr, _ := env["addr"].(string)
+					if addr != "" {
+						select {
+						case c.hub.commandCh <- hubCmd{op: "wsAddSource", label: label, addr: addr}:
+						default:
+						}
+					}
+				case "removeSource":
+					id, _ := env["id"].(string)
+					if id != "" {
+						select {
+						case c.hub.commandCh <- hubCmd{op: "wsRemoveSource", sourceID: id}:
+						default:
+						}
+					}
+				case "saveSources":
+					select {
+					case c.hub.commandCh <- hubCmd{op: "wsSaveSources"}:
+					default:
+					}
 				}
 			}
 		}
@@ -82,21 +106,56 @@ func (c *wsClient) readPump() {
 var upgrader = websocket.Upgrader{
 	ReadBufferSize:  4096,
 	WriteBufferSize: 64 * 1024,
-	CheckOrigin:    func(r *http.Request) bool { return true },
+	CheckOrigin:     func(r *http.Request) bool { return true },
 }
 
-// Hub is the central data broker between the UDP client and WebSocket clients.
+// sourceHubState holds all data for one active data source.
-type Hub struct {
+// Only accessed from the Run() goroutine.
-	mu       sync.RWMutex
+type sourceHubState struct {
-	signals  []SignalInfo
+	id, label, addr, connState string
-	configJS []byte // cached JSON config message
+	signals                    []SignalInfo
+	configJS                   []byte
 
+	// Time-signal calibration — only accessed from Run() goroutine.
+	timeSigCalib     map[string]float64
+	configSeq        uint64
+	configSeqAtCalib uint64
+}
+
+// taggedSample is a DataSample annotated with its source ID.
+type taggedSample struct {
+	sourceID string
+	sample   DataSample
+}
+
+// hubCmd carries a command to the Run() goroutine.
+type hubCmd struct {
+	op       string // "addSource","removeSource","setSourceState","updateConfig",
+	//                "wsAddSource","wsRemoveSource","wsSaveSources"
+	sourceID string
+	label    string
+	addr     string
+	state    string
+	sigs     []SignalInfo
+}
+
+// Hub is the central broker between UDP clients and WebSocket clients.
+// All map state is accessed exclusively from the Run() goroutine, except
+// ringsMu/rings which are also read by HTTP handler goroutines.
+type Hub struct {
 	clients     map[*wsClient]bool
 	register    chan *wsClient
 	unregister  chan *wsClient
-	broadcastCh chan []byte // all sends go through Run() to avoid races on c.send
+	broadcastCh chan []byte
+	dataCh      chan taggedSample
+	commandCh   chan hubCmd
 
-	dataCh chan DataSample // incoming samples from UDP goroutine
+	sm *SourceManager // set after construction; used for WS-initiated source changes
+
+	// Ring buffers for hi-res zoom data.
+	// ringsMu protects the map structure; each sigRing has its own RWMutex for data.
+	ringsMu sync.RWMutex
+	rings   map[string]*sigRing // "sourceId:signalKey" → ring
 }
 
 // NewHub creates an initialised Hub.
@@ -106,44 +165,125 @@ func NewHub() *Hub {
 		register:    make(chan *wsClient, 8),
 		unregister:  make(chan *wsClient, 8),
 		broadcastCh: make(chan []byte, 64),
-		dataCh:      make(chan DataSample, 256),
+		dataCh:      make(chan taggedSample, 256),
+		commandCh:   make(chan hubCmd, 64),
+		rings:       make(map[string]*sigRing),
 	}
 }
 
-// UpdateConfig stores a new signal config and broadcasts it to all WS clients.
+// getRing returns the ring buffer for a fully-prefixed signal key, or nil.
-func (h *Hub) UpdateConfig(sigs []SignalInfo) {
+func (h *Hub) getRing(key string) *sigRing {
-	msg, err := json.Marshal(map[string]interface{}{
+	h.ringsMu.RLock()
-		"type":    "config",
+	rb := h.rings[key]
-		"signals": sigs,
+	h.ringsMu.RUnlock()
-	})
+	return rb
-	if err != nil {
+}
-		log.Printf("hub: marshal config: %v", err)
+
+// HandleZoom serves GET /api/zoom?t0=...&t1=...&n=...&signals=key1,key2
+// It reads from the ring buffers (safe for concurrent access) and returns
+// LTTB-decimated signal data for the requested time range.
+func (h *Hub) HandleZoom(w http.ResponseWriter, r *http.Request) {
+	q := r.URL.Query()
+	t0, err0 := strconv.ParseFloat(q.Get("t0"), 64)
+	t1, err1 := strconv.ParseFloat(q.Get("t1"), 64)
+	if err0 != nil || err1 != nil || t1 <= t0 {
+		http.Error(w, "invalid t0/t1", http.StatusBadRequest)
 		return
 	}
-	h.mu.Lock()
+	// n=0 (or explicit "0") means no LTTB decimation — return all ring data in range.
-	h.signals = sigs
+	// n omitted / invalid → default 2400 (display quality).
-	h.configJS = msg
+	var n int
-	h.mu.Unlock()
+	if nStr := q.Get("n"); nStr == "" {
+		n = 2400
+	} else {
+		n, _ = strconv.Atoi(nStr)
+		if n <= 0 {
+			n = 1 << 30 // no decimation
+		} else if n < 10 {
+			n = 2400
+		}
+	}
 
-	h.broadcast(msg)
+	keys := strings.Split(q.Get("signals"), ",")
+
+	// Collect ring references under a brief RLock.
+	h.ringsMu.RLock()
+	refs := make(map[string]*sigRing, len(keys))
+	for _, k := range keys {
+		k = strings.TrimSpace(k)
+		if k == "" {
+			continue
+		}
+		if rb, ok := h.rings[k]; ok {
+			refs[k] = rb
+		}
+	}
+	h.ringsMu.RUnlock()
+
+	result := make(map[string]sigData, len(refs))
+	for k, rb := range refs {
+		rt, rv := rb.slice(t0, t1)
+		if len(rt) == 0 {
+			continue
+		}
+		dt, dv := lttbDecimate(rt, rv, n)
+		result[k] = sigData{T: dt, V: dv}
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	if err := json.NewEncoder(w).Encode(map[string]any{
+		"type":    "zoom",
+		"signals": result,
+	}); err != nil {
+		log.Printf("hub: zoom encode: %v", err)
+	}
 }
 
-// PushData enqueues a data sample for broadcasting to WebSocket clients.
+// AddSource notifies the Hub that a new source has been registered.
-func (h *Hub) PushData(s DataSample) {
+func (h *Hub) AddSource(id, label, addr string) {
 	select {
-	case h.dataCh <- s:
+	case h.commandCh <- hubCmd{op: "addSource", sourceID: id, label: label, addr: addr}:
+	default:
+	}
+}
+
+// RemoveSource notifies the Hub that a source has been removed.
+func (h *Hub) RemoveSource(id string) {
+	select {
+	case h.commandCh <- hubCmd{op: "removeSource", sourceID: id}:
+	default:
+	}
+}
+
+// SetSourceState updates the connection state of a source.
+func (h *Hub) SetSourceState(id, state string) {
+	select {
+	case h.commandCh <- hubCmd{op: "setSourceState", sourceID: id, state: state}:
+	default:
+	}
+}
+
+// UpdateConfigForSource stores a new signal config for a source and broadcasts it.
+func (h *Hub) UpdateConfigForSource(sourceID string, sigs []SignalInfo) {
+	select {
+	case h.commandCh <- hubCmd{op: "updateConfig", sourceID: sourceID, sigs: sigs}:
+	default:
+	}
+}
+
+// PushDataForSource enqueues a data sample from a specific source.
+func (h *Hub) PushDataForSource(sourceID string, s DataSample) {
+	select {
+	case h.dataCh <- taggedSample{sourceID: sourceID, sample: s}:
 	default:
-		// Drop if buffer full to avoid blocking the UDP goroutine.
 	}
 }
 
 // broadcast enqueues a message for delivery to all WebSocket clients.
-// All actual sends happen inside Run() to avoid concurrent access to c.send.
 func (h *Hub) broadcast(msg []byte) {
 	select {
 	case h.broadcastCh <- msg:
 	default:
-		// Drop if the broadcast queue is full.
 	}
 }
 
@@ -154,173 +294,348 @@ func (h *Hub) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
 		log.Printf("ws upgrade: %v", err)
 		return
 	}
-	c := &wsClient{
+	c := &wsClient{hub: h, conn: conn, send: make(chan []byte, 64)}
-		hub:  h,
-		conn: conn,
-		send: make(chan []byte, 64),
-	}
 	h.register <- c
-
 	go c.writePump()
 	go c.readPump()
 }
 
-// Run is the hub's main goroutine. It must be started with go hub.Run().
+// buildSourcesMsg serialises the current source list as a JSON "sources" message.
+func buildSourcesMsg(sm map[string]*sourceHubState) []byte {
+	type srcInfo struct {
+		ID    string `json:"id"`
+		Label string `json:"label"`
+		Addr  string `json:"addr"`
+		State string `json:"state"`
+	}
+	list := make([]srcInfo, 0, len(sm))
+	for _, src := range sm {
+		list = append(list, srcInfo{ID: src.id, Label: src.label, Addr: src.addr, State: src.connState})
+	}
+	msg, _ := json.Marshal(map[string]interface{}{"type": "sources", "sources": list})
+	return msg
+}
+
+// Run is the hub's main goroutine. Must be started with go hub.Run().
 func (h *Hub) Run() {
-	// Batch data at ≤30 Hz.
 	ticker := time.NewTicker(time.Second / 30)
 	defer ticker.Stop()
 
-	// Accumulate samples between ticks.
+	sourcesMap := make(map[string]*sourceHubState)
-	pending := make([]DataSample, 0, 64)
+	var sourcesMsg []byte
+
+	// pending[sourceID] accumulates samples between 30 Hz ticks.
+	pending := make(map[string][]DataSample)
+
+	rebuildSources := func() {
+		sourcesMsg = buildSourcesMsg(sourcesMap)
+		h.broadcast(sourcesMsg)
+	}
 
 	for {
 		select {
 		case c := <-h.register:
-			h.mu.Lock()
 			h.clients[c] = true
-			cfg := h.configJS
+			// Send current state to the new client.
-			h.mu.Unlock()
+			if sourcesMsg != nil {
-			// Send current config immediately if we have one.
+				select { case c.send <- sourcesMsg: default: }
-			if cfg != nil {
+			}
-				select {
+			for _, src := range sourcesMap {
-				case c.send <- cfg:
+				if src.configJS != nil {
-				default:
+					select { case c.send <- src.configJS: default: }
 				}
 			}
 
 		case c := <-h.unregister:
-			h.mu.Lock()
 			if _, ok := h.clients[c]; ok {
 				delete(h.clients, c)
 				close(c.send)
 			}
-			h.mu.Unlock()
 
 		case msg := <-h.broadcastCh:
-			h.mu.RLock()
 			for c := range h.clients {
-				select {
+				select { case c.send <- msg: default: }
-				case c.send <- msg:
+			}
-				default:
+
+		case cmd := <-h.commandCh:
+			switch cmd.op {
+			case "addSource":
+				sourcesMap[cmd.sourceID] = &sourceHubState{
+					id:           cmd.sourceID,
+					label:        cmd.label,
+					addr:         cmd.addr,
+					connState:    "connecting",
+					timeSigCalib: make(map[string]float64),
+				}
+				rebuildSources()
+
+			case "removeSource":
+				delete(sourcesMap, cmd.sourceID)
+				delete(pending, cmd.sourceID)
+				pfxDel := cmd.sourceID + ":"
+				h.ringsMu.Lock()
+				for k := range h.rings {
+					if strings.HasPrefix(k, pfxDel) {
+						delete(h.rings, k)
+					}
+				}
+				h.ringsMu.Unlock()
+				rebuildSources()
+
+			case "setSourceState":
+				if src, ok := sourcesMap[cmd.sourceID]; ok {
+					src.connState = cmd.state
+					rebuildSources()
+				}
+
+			case "updateConfig":
+				src, ok := sourcesMap[cmd.sourceID]
+				if !ok {
+					continue
+				}
+				src.signals = cmd.sigs
+				src.configSeq++
+				cfgMsg, err := json.Marshal(map[string]any{
+					"type":     "config",
+					"sourceId": cmd.sourceID,
+					"signals":  cmd.sigs,
+				})
+				if err != nil {
+					log.Printf("hub: marshal config: %v", err)
+					continue
+				}
+				src.configJS = cfgMsg
+				h.broadcast(cfgMsg)
+				// Rebuild ring buffers for this source.
+				pfxUpd := cmd.sourceID + ":"
+				h.ringsMu.Lock()
+				for k := range h.rings {
+					if strings.HasPrefix(k, pfxUpd) {
+						delete(h.rings, k)
+					}
+				}
+				for _, sig := range cmd.sigs {
+					ne := sig.NumElements()
+					isTemporal := ne > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample)
+					if isTemporal {
+						h.rings[pfxUpd+sig.Name] = newSigRing(ringCapTemporal)
+					} else if ne == 1 {
+						h.rings[pfxUpd+sig.Name] = newSigRing(ringCapScalar)
+					} else {
+						for i := 0; i < ne; i++ {
+							h.rings[pfxUpd+arrayKey(sig.Name, i)] = newSigRing(ringCapScalar)
+						}
+					}
+				}
+				h.ringsMu.Unlock()
+
+			case "wsAddSource":
+				if h.sm != nil {
+					go func(label, addr string) { h.sm.Add(label, addr) }(cmd.label, cmd.addr)
+				}
+
+			case "wsRemoveSource":
+				if h.sm != nil {
+					go func(id string) { h.sm.Remove(id) }(cmd.sourceID)
+				}
+
+			case "wsSaveSources":
+				if h.sm != nil {
+					if err := h.sm.Save(); err != nil {
+						log.Printf("hub: save sources: %v", err)
+					}
 				}
 			}
-			h.mu.RUnlock()
 
-		case s := <-h.dataCh:
+		case ts := <-h.dataCh:
-			pending = append(pending, s)
+			pending[ts.sourceID] = append(pending[ts.sourceID], ts.sample)
 
 		case <-ticker.C:
-			if len(pending) == 0 {
+			for srcID, samples := range pending {
-				continue
+				if len(samples) == 0 {
-			}
+					continue
-			h.mu.RLock()
+				}
-			sigs := h.signals
+				src, ok := sourcesMap[srcID]
-			noClients := len(h.clients) == 0
+				if !ok || len(src.signals) == 0 || len(h.clients) == 0 {
-			h.mu.RUnlock()
+					pending[srcID] = pending[srcID][:0]
-
+					continue
-			if noClients || len(sigs) == 0 {
+				}
-				pending = pending[:0]
+				msg := h.buildDataMessageForSource(src, samples)
-				continue
+				pending[srcID] = pending[srcID][:0]
-			}
+				if msg != nil {
-
+					h.broadcast(msg)
-			msg := h.buildDataMessage(pending, sigs)
+				}
-			pending = pending[:0]
-			if msg != nil {
-				h.broadcast(msg)
 			}
 		}
 	}
 }
 
-// maxBatchPoints is the maximum number of points sent per signal per 30 Hz display tick.
+// ─── Data serialisation ───────────────────────────────────────────────────────
-// For temporal-array signals (e.g. 1 MSps packed as 1000 samples/packet), the expanded
-// sample stream is decimated to this limit before transmission to WebSocket clients.
-const maxBatchPoints = 2000
 
-// sigData carries one signal's worth of time+value pairs in a single batch message.
+// maxPushPoints is the LTTB target for data pushed over WebSocket per 30 Hz tick.
+// At 2 k pts/tick × 30 Hz = 60 k pts/s; the 600 k frontend buffer covers ~10 s.
+// Kept deliberately low so the rolling window shows plenty of history even for
+// multi-MHz signals — zoom resolution comes from the ring buffer instead.
+const maxPushPoints = 2_000
+
+// maxRingPoints is the LTTB target written into the ring buffer per tick.
+// At 5 Msps / 5 kHz packet rate ≈ 167 k raw samples/tick → LTTB to 20 k →
+// min Δt ≈ 33 ms / 20 k ≈ 1.65 µs, sufficient for sub-10 µs zoom resolution.
+const maxRingPoints = 20_000
+
+// ringCapTemporal is the ring buffer capacity for temporal-array signals.
+// At 20 k pts/tick × 30 Hz = 600 k pts/s → 6 M cap gives ~10 s of hi-res
+// history — the same temporal coverage as the frontend push buffer.
+const ringCapTemporal = 6_000_000
+
+// ringCapScalar is the ring buffer capacity for scalar / spatial-array signals.
+// At ≤10 kHz → ~333 pts/tick × 30 Hz ≈ 10 k pts/s → ~10 s of history.
+const ringCapScalar = 100_000
+
+// lttbDecimate reduces (tIn, vIn) to at most threshold representative points
+// using the Largest-Triangle-Three-Buckets algorithm.
+// Returns the original slices unchanged when len(tIn) ≤ threshold.
+func lttbDecimate(tIn, vIn []float64, threshold int) ([]float64, []float64) {
+	n := len(tIn)
+	if n <= threshold || threshold < 3 {
+		return tIn, vIn
+	}
+	outT := make([]float64, threshold)
+	outV := make([]float64, threshold)
+	outT[0], outV[0] = tIn[0], vIn[0]
+	outT[threshold-1], outV[threshold-1] = tIn[n-1], vIn[n-1]
+
+	every := float64(n-2) / float64(threshold-2)
+	a := 0
+	for i := 0; i < threshold-2; i++ {
+		// Centroid of the next bucket
+		avgS := int(float64(i+1)*every) + 1
+		avgE := int(float64(i+2)*every) + 1
+		if avgE > n {
+			avgE = n
+		}
+		avgT, avgV, cnt := 0.0, 0.0, 0
+		for j := avgS; j < avgE; j++ {
+			avgT += tIn[j]; avgV += vIn[j]; cnt++
+		}
+		if cnt > 0 {
+			avgT /= float64(cnt); avgV /= float64(cnt)
+		}
+		// Pick the point in the current bucket that forms the largest triangle
+		rS := int(float64(i)*every) + 1
+		rE := int(float64(i+1)*every) + 1
+		if rE > n {
+			rE = n
+		}
+		maxArea, next := -1.0, rS
+		aT, aV := tIn[a], vIn[a]
+		for j := rS; j < rE; j++ {
+			area := math.Abs((aT-avgT)*(vIn[j]-aV) - (aT-tIn[j])*(avgV-aV))
+			if area > maxArea {
+				maxArea = area; next = j
+			}
+		}
+		outT[i+1], outV[i+1] = tIn[next], vIn[next]
+		a = next
+	}
+	return outT, outV
+}
+
+// sigData carries one signal's worth of time+value pairs.
 type sigData struct {
-	T []float64 `json:"t"` // unix seconds
+	T []float64 `json:"t"`
-	V []float64 `json:"v"` // physical values
+	V []float64 `json:"v"`
 }
 
-// dataMsg is the JSON envelope for batched data sent to WebSocket clients.
+// dataMsg is the JSON envelope sent to WebSocket clients.
-// Each signal carries its own time axis so that temporal arrays (packed sample
-// bursts with a known sampling rate) and scalar signals can coexist cleanly.
 type dataMsg struct {
-	Type    string             `json:"type"`
+	Type     string             `json:"type"`
-	Signals map[string]sigData `json:"signals"`
+	SourceID string             `json:"sourceId"`
+	Signals  map[string]sigData `json:"signals"`
 }
 
-// buildDataMessage merges a batch of DataSamples into one JSON message.
+// buildDataMessageForSource serialises a batch of samples for one source.
-//
+// Signal keys in the output are prefixed with "sourceId:" so the browser can
-// Three cases are handled:
+// store them in a single flat buffer map without collision.
-//
+func (h *Hub) buildDataMessageForSource(src *sourceHubState, batch []DataSample) []byte {
-//  1. Temporal array (NumElements > 1, TimeMode != PacketTime):
-//     The N samples in each packet represent a contiguous time burst at SamplingRate.
-//     Wall arrival time is treated as the timestamp of the last sample; earlier
-//     samples are reconstructed as t[k] = wallT - (N-1-k)/SamplingRate.
-//     The full expanded stream is decimated to maxBatchPoints if needed.
-//
-//  2. Scalar signal (NumElements == 1):
-//     One {t, v} pair per packet – wall arrival time used as timestamp.
-//
-//  3. Spatial / PacketTime array (NumElements > 1, TimeMode == 0):
-//     Each element is tracked as a separate stream keyed "sig[i]", with wall
-//     arrival time as the shared timestamp.
-func (h *Hub) buildDataMessage(batch []DataSample, sigs []SignalInfo) []byte {
 	if len(batch) == 0 {
 		return nil
 	}
 
+	// Reset time-signal calibration whenever the signal config changed.
+	if src.configSeq != src.configSeqAtCalib {
+		src.configSeqAtCalib = src.configSeq
+		src.timeSigCalib = make(map[string]float64)
+	}
+
+	sigs := src.signals
+	pfx := src.id + ":"
 	out := make(map[string]sigData, len(sigs)*2)
 
 	for _, sig := range sigs {
 		n := sig.NumElements()
-		isTemporal := n > 1 && sig.TimeMode != 0
+		isTemporal := n > 1 && (sig.TimeMode == TimeModeFirstSample || sig.TimeMode == TimeModeLastSample)
 
 		switch {
 		case isTemporal:
-			// Expand each packet's N samples into individual time-stamped points.
+			hasTimeSig := sig.TimeSignalIdx != NoTimeSignal && int(sig.TimeSignalIdx) < len(sigs)
-			allT := make([]float64, 0, len(batch)*n)
+			var timeSigName string
-			allV := make([]float64, 0, len(batch)*n)
+			if hasTimeSig {
-
+				timeSigName = sigs[sig.TimeSignalIdx].Name
+			}
 			dt := 0.0
 			if sig.SamplingRate > 0 {
 				dt = 1.0 / sig.SamplingRate
 			}
+			allT := make([]float64, 0, len(batch)*n)
+			allV := make([]float64, 0, len(batch)*n)
 
 			for _, s := range batch {
 				vals, ok := s.Values[sig.Name]
 				if !ok || len(vals) < n {
 					continue
 				}
-				// wallT ≈ arrival time of the packet ≈ timestamp of the last sample.
+				var anchorTime float64
-				wallT := float64(s.WallTime.UnixNano()) / 1e9
+				anchorIsFirstSample := sig.TimeMode == TimeModeFirstSample
+				if hasTimeSig {
+					tVals, tOk := s.Values[timeSigName]
+					if tOk && len(tVals) >= 1 {
+						timerS := tVals[0] * 1e-6
+						wallT := float64(s.WallTime.UnixNano()) / 1e9
+						if _, exists := src.timeSigCalib[timeSigName]; !exists {
+							src.timeSigCalib[timeSigName] = wallT - timerS
+						}
+						anchorTime = src.timeSigCalib[timeSigName] + timerS
+					} else {
+						anchorTime = float64(s.WallTime.UnixNano()) / 1e9
+						anchorIsFirstSample = false
+					}
+				} else {
+					anchorTime = float64(s.WallTime.UnixNano()) / 1e9
+					anchorIsFirstSample = false
+				}
 				for k := 0; k < n; k++ {
-					allT = append(allT, wallT-float64(n-1-k)*dt)
+					var t float64
+					if anchorIsFirstSample {
+						t = anchorTime + float64(k)*dt
+					} else {
+						t = anchorTime - float64(n-1-k)*dt
+					}
+					allT = append(allT, t)
 					allV = append(allV, vals[k])
 				}
 			}
 
-			// Decimate to maxBatchPoints if necessary.
+			// Write hi-res LTTB data to ring for on-demand zoom queries.
-			step := 1
+			ringT, ringV := lttbDecimate(allT, allV, maxRingPoints)
-			if len(allT) > maxBatchPoints {
+			if rb := h.getRing(pfx + sig.Name); rb != nil {
-				step = len(allT) / maxBatchPoints
+				rb.write(ringT, ringV)
-				if step < 1 {
-					step = 1
-				}
 			}
-			decimT := make([]float64, 0, len(allT)/step+1)
+			// Decimate further for WebSocket push (rolling window).
-			decimV := make([]float64, 0, len(allV)/step+1)
+			decimT, decimV := lttbDecimate(allT, allV, maxPushPoints)
-			for i := 0; i < len(allT); i += step {
+			out[pfx+sig.Name] = sigData{T: decimT, V: decimV}
-				decimT = append(decimT, allT[i])
-				decimV = append(decimV, allV[i])
-			}
-			out[sig.Name] = sigData{T: decimT, V: decimV}
 
 		case n == 1:
-			// Scalar signal: one sample per packet.
 			ts := make([]float64, 0, len(batch))
 			vs := make([]float64, 0, len(batch))
 			for _, s := range batch {
@@ -331,12 +646,15 @@ func (h *Hub) buildDataMessage(batch []DataSample, sigs []SignalInfo) []byte {
 				ts = append(ts, float64(s.WallTime.UnixNano())/1e9)
 				vs = append(vs, vals[0])
 			}
-			out[sig.Name] = sigData{T: ts, V: vs}
+			if rb := h.getRing(pfx + sig.Name); rb != nil {
+				rb.write(ts, vs)
+			}
+			out[pfx+sig.Name] = sigData{T: ts, V: vs}
 
 		default:
-			// Spatial / PacketTime array: one stream per element, keyed "sig[i]".
+			// Spatial / PacketTime array: one stream per element.
 			for i := 0; i < n; i++ {
-				key := arrayKey(sig.Name, i)
+				key := pfx + arrayKey(sig.Name, i)
 				ts := make([]float64, 0, len(batch))
 				vs := make([]float64, 0, len(batch))
 				for _, s := range batch {
@@ -347,15 +665,15 @@ func (h *Hub) buildDataMessage(batch []DataSample, sigs []SignalInfo) []byte {
 					ts = append(ts, float64(s.WallTime.UnixNano())/1e9)
 					vs = append(vs, vals[i])
 				}
+				if rb := h.getRing(key); rb != nil {
+					rb.write(ts, vs)
+				}
 				out[key] = sigData{T: ts, V: vs}
 			}
 		}
 	}
 
-	result, err := json.Marshal(dataMsg{
+	result, err := json.Marshal(dataMsg{Type: "data", SourceID: src.id, Signals: out})
-		Type:    "data",
-		Signals: out,
-	})
 	if err != nil {
 		log.Printf("hub: marshal data: %v", err)
 		return nil

Client/WebUI/main.go

@@ -2,45 +2,63 @@ package main
 
 import (
 	"embed"
+	"errors"
 	"flag"
+	"fmt"
+	"io/fs"
 	"log"
 	"net/http"
+	"os"
 )
 
-//go:embed static/index.html
+// version is set at build time via -ldflags "-X main.version=..."
+var version = "dev"
+
+//go:embed static
 var staticFiles embed.FS
 
+// multiFlag allows a flag to be repeated: --source a --source b
+type multiFlag []string
+
+func (f *multiFlag) String() string  { return fmt.Sprintf("%v", []string(*f)) }
+func (f *multiFlag) Set(v string) error { *f = append(*f, v); return nil }
+
 func main() {
-	streamerAddr := flag.String("streamer", "127.0.0.1:44500", "MARTe2 UDP streamer address (host:port)")
+	var sourceArgs multiFlag
+	flag.Var(&sourceArgs, "source", `Data source in the form [label@]host:port (repeatable)`)
+	sourcesFile := flag.String("sources-file", "", "JSON file for persistent source list (load on start, save target)")
 	listenAddr := flag.String("listen", ":8080", "HTTP listen address")
-	clientPort := flag.Int("clientport", 44900, "Local UDP port to bind for streamer data")
 	flag.Parse()
 
 	hub := NewHub()
+	sm := NewSourceManager(hub, *sourcesFile)
+	hub.sm = sm
+
 	go hub.Run()
 
-	udpClient := NewUDPClient(*streamerAddr, *clientPort, hub)
+	// Load sources from file first (if specified), then add any CLI --source flags.
-	go udpClient.Run()
+	if *sourcesFile != "" {
+		if err := sm.Load(*sourcesFile); err != nil && !errors.Is(err, os.ErrNotExist) {
+			log.Printf("sources-file load: %v", err)
+		}
+	}
+	for _, arg := range sourceArgs {
+		label, addr := ParseSourceArg(arg)
+		sm.Add(label, addr)
+	}
 
-	// Serve the embedded index.html at /.
+	sub, err := fs.Sub(staticFiles, "static")
-	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+	if err != nil {
-		if r.URL.Path != "/" {
+		log.Fatalf("static sub-fs: %v", err)
-			http.NotFound(w, r)
+	}
-			return
+	http.Handle("/", http.FileServer(http.FS(sub)))
-		}
+	http.HandleFunc("/ws", hub.HandleWebSocket)
-		data, err := staticFiles.ReadFile("static/index.html")
+	http.HandleFunc("/api/zoom", hub.HandleZoom)
-		if err != nil {
+	http.HandleFunc("/version", func(w http.ResponseWriter, r *http.Request) {
-			http.Error(w, "index.html not found", http.StatusInternalServerError)
+		fmt.Fprint(w, version)
-			return
-		}
-		w.Header().Set("Content-Type", "text/html; charset=utf-8")
-		w.Write(data)
 	})
 
-	http.HandleFunc("/ws", hub.HandleWebSocket)
+	log.Printf("WebUI listening on %s (version=%s)", *listenAddr, version)
-
-	log.Printf("WebUI listening on %s  (streamer=%s, local UDP port=%d)",
-		*listenAddr, *streamerAddr, *clientPort)
 	if err := http.ListenAndServe(*listenAddr, nil); err != nil {
 		log.Fatalf("http: %v", err)
 	}

Client/WebUI/protocol.go

@@ -19,15 +19,21 @@ const (
 	PktConnect    uint8 = 3
 	PktDisconnect uint8 = 4
 
-	HeaderSize     = 17
+	HeaderSize   = 17
-	SigDescSize    = 136
+	SigDescSize  = 136
-	NoTimeSignal   = uint32(0xFFFFFFFF)
+	NoTimeSignal = uint32(0xFFFFFFFF)
 
 	QuantNone   uint8 = 0
 	QuantUint8  uint8 = 1
 	QuantInt8   uint8 = 2
 	QuantUint16 uint8 = 3
 	QuantInt16  uint8 = 4
+
+	// TimeMode values – must match UDPStreamerTimeMode enum in UDPStreamer.h
+	TimeModePacket      uint8 = 0 // use wall-clock packet arrival time
+	TimeModeFullArray   uint8 = 1 // TimeSignal has same N elements; not expanded here
+	TimeModeFirstSample uint8 = 2 // TimeSignal scalar = time of element [0]
+	TimeModeLastSample  uint8 = 3 // TimeSignal scalar = time of element [N-1]
 )
 
 // ─── Packet header (17 bytes, little-endian, packed) ─────────────────────────

Client/WebUI/ringbuf.go

@@ -0,0 +1,88 @@
+package main
+
+import "sync"
+
+// sigRing is a fixed-capacity circular buffer storing (time, value) pairs.
+// Writes come from the Hub.Run() goroutine; reads come from HTTP handler goroutines.
+// The embedded RWMutex protects concurrent access.
+type sigRing struct {
+	mu         sync.RWMutex
+	t, v       []float64
+	cap        int
+	head, size int // next write position; current fill
+}
+
+func newSigRing(capacity int) *sigRing {
+	return &sigRing{
+		t:   make([]float64, capacity),
+		v:   make([]float64, capacity),
+		cap: capacity,
+	}
+}
+
+// write appends (tArr[i], vArr[i]) pairs, overwriting oldest entries when full.
+func (rb *sigRing) write(tArr, vArr []float64) {
+	rb.mu.Lock()
+	defer rb.mu.Unlock()
+	for i := 0; i < len(tArr); i++ {
+		rb.t[rb.head] = tArr[i]
+		rb.v[rb.head] = vArr[i]
+		rb.head = (rb.head + 1) % rb.cap
+		if rb.size < rb.cap {
+			rb.size++
+		}
+	}
+}
+
+// slice returns copies of all (t, v) pairs whose timestamp falls in [t0, t1].
+// The returned slices are safe to use after the call without holding any lock.
+func (rb *sigRing) slice(t0, t1 float64) ([]float64, []float64) {
+	rb.mu.RLock()
+	defer rb.mu.RUnlock()
+
+	if rb.size == 0 {
+		return nil, nil
+	}
+	start := 0
+	if rb.size == rb.cap {
+		start = rb.head
+	}
+	physAt := func(k int) int { return (start + k) % rb.cap }
+
+	// Binary search for t0
+	lo, hi := 0, rb.size
+	for lo < hi {
+		m := (lo + hi) >> 1
+		if rb.t[physAt(m)] < t0 {
+			lo = m + 1
+		} else {
+			hi = m
+		}
+	}
+	kStart := lo
+
+	// Binary search for t1
+	lo, hi = kStart, rb.size
+	for lo < hi {
+		m := (lo + hi) >> 1
+		if rb.t[physAt(m)] <= t1 {
+			lo = m + 1
+		} else {
+			hi = m
+		}
+	}
+	kEnd := lo
+
+	n := kEnd - kStart
+	if n <= 0 {
+		return nil, nil
+	}
+	outT := make([]float64, n)
+	outV := make([]float64, n)
+	for i := 0; i < n; i++ {
+		p := physAt(kStart + i)
+		outT[i] = rb.t[p]
+		outV[i] = rb.v[p]
+	}
+	return outT, outV
+}

Client/WebUI/run.sh

@@ -0,0 +1,13 @@
+#!/usr/bin/env bash
+set -e
+cd "$(dirname "$0")"
+
+COMMIT="$(git describe --always --dirty 2>/dev/null || echo 'unknown')"
+BUILD_TIME="$(date -u +%Y%m%dT%H%M%SZ)"
+VERSION="${COMMIT}-${BUILD_TIME}"
+
+echo "Building udpstreamer-webui ${VERSION}..."
+go build -ldflags "-X main.version=${VERSION}" -o udpstreamer-webui .
+
+echo "Starting udpstreamer-webui ${VERSION}..."
+exec ./udpstreamer-webui "$@"

Client/WebUI/sources.go

@@ -0,0 +1,128 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+)
+
+// SourceConfig is the serialisable description of one data source (for file save/load).
+type SourceConfig struct {
+	Label string `json:"label"`
+	Addr  string `json:"addr"`
+}
+
+// managedSource is the SourceManager's view of one running source.
+type managedSource struct {
+	id     string
+	label  string
+	addr   string
+	client *UDPClient
+}
+
+// SourceManager owns the lifecycle of all active data sources.
+// It is the only place where UDPClient instances are created and stopped.
+type SourceManager struct {
+	mu       sync.RWMutex
+	sources  map[string]*managedSource
+	hub      *Hub
+	filePath string // empty = save not configured
+	nextID   atomic.Int32
+}
+
+// NewSourceManager creates a SourceManager bound to the given hub.
+// filePath is the default path for Save(); may be empty.
+func NewSourceManager(hub *Hub, filePath string) *SourceManager {
+	return &SourceManager{
+		sources:  make(map[string]*managedSource),
+		hub:      hub,
+		filePath: filePath,
+	}
+}
+
+func (sm *SourceManager) genID() string {
+	return fmt.Sprintf("s%d", sm.nextID.Add(1))
+}
+
+// Add creates a new source, registers it in the hub, and starts connecting.
+// Returns the source ID.
+func (sm *SourceManager) Add(label, addr string) string {
+	if label == "" {
+		label = addr
+	}
+	id := sm.genID()
+	c := NewUDPClient(addr, id, sm.hub)
+	ms := &managedSource{id: id, label: label, addr: addr, client: c}
+
+	sm.mu.Lock()
+	sm.sources[id] = ms
+	sm.mu.Unlock()
+
+	// Register in hub (creates hub-side state, broadcasts sources list).
+	sm.hub.AddSource(id, label, addr)
+	go c.Run()
+	return id
+}
+
+// Remove stops the source's client and removes it from the hub.
+func (sm *SourceManager) Remove(id string) {
+	sm.mu.Lock()
+	ms, ok := sm.sources[id]
+	if ok {
+		delete(sm.sources, id)
+	}
+	sm.mu.Unlock()
+
+	if ok {
+		ms.client.Stop()
+		sm.hub.RemoveSource(id)
+	}
+}
+
+// Save writes the current source list to filePath.
+func (sm *SourceManager) Save() error {
+	if sm.filePath == "" {
+		return fmt.Errorf("no sources-file configured (start with --sources-file path)")
+	}
+	sm.mu.RLock()
+	cfgs := make([]SourceConfig, 0, len(sm.sources))
+	for _, ms := range sm.sources {
+		cfgs = append(cfgs, SourceConfig{Label: ms.label, Addr: ms.addr})
+	}
+	sm.mu.RUnlock()
+
+	data, err := json.MarshalIndent(cfgs, "", "  ")
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(sm.filePath, data, 0644)
+}
+
+// Load reads sources from path and adds them. Sets filePath for future saves.
+func (sm *SourceManager) Load(path string) error {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return err
+	}
+	var cfgs []SourceConfig
+	if err := json.Unmarshal(data, &cfgs); err != nil {
+		return err
+	}
+	sm.filePath = path
+	for _, cfg := range cfgs {
+		sm.Add(cfg.Label, cfg.Addr)
+	}
+	return nil
+}
+
+// ParseSourceArg parses "label@host:port" or "host:port".
+func ParseSourceArg(s string) (label, addr string) {
+	s = strings.TrimSpace(s)
+	if idx := strings.Index(s, "@"); idx >= 0 {
+		return strings.TrimSpace(s[:idx]), strings.TrimSpace(s[idx+1:])
+	}
+	return "", s
+}

Client/WebUI/static/favicon.svg

@@ -0,0 +1,5 @@
+<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 32 32" width="32" height="32">
+  <rect width="32" height="32" rx="6" fill="#1e1e2e"/>
+  <polyline points="2,16 7,16 9,8 11,24 14,6 17,26 20,12 23,20 25,16 30,16"
+            fill="none" stroke="#89b4fa" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"/>
+</svg>

Client/WebUI/static/style.css

@@ -0,0 +1,389 @@
+/* ── Catppuccin Mocha palette ──────────────────────────────── */
+:root {
+  --bg:       #1e1e2e; --mantle:   #181825; --crust:    #11111b;
+  --surface0: #313244; --surface1: #45475a; --surface2: #585b70;
+  --overlay0: #6c7086; --overlay1: #7f849c; --text:     #cdd6f4;
+  --subtext0: #a6adc8; --subtext1: #bac2de;
+  --accent:   #89b4fa; --green:    #a6e3a1; --red:      #f38ba8;
+  --yellow:   #f9e2af; --peach:    #fab387; --mauve:    #cba6f7;
+  --teal:     #94e2d5; --sky:      #89dceb; --lavender: #b4befe;
+  --pink:     #f5c2e7;
+  --radius:   8px; --sidebar-w: 280px; --topbar-h: 52px;
+  --trigbar-h: 0px; --statusbar-h: 20px; --transition: 0.18s ease;
+}
+*, *::before, *::after { box-sizing: border-box; margin: 0; padding: 0; }
+html, body { height:100%; background:var(--bg); color:var(--text);
+  font-family:'Segoe UI',system-ui,sans-serif; font-size:14px; overflow:hidden; }
+::-webkit-scrollbar { width:6px; }
+::-webkit-scrollbar-track { background:var(--mantle); }
+::-webkit-scrollbar-thumb { background:var(--surface1); border-radius:3px; }
+
+/* ── uPlot overrides ─────────────────────────────────────────── */
+.uplot { background:transparent !important; }
+.uplot .u-over { cursor: crosshair; }
+.uplot .u-cursor-x, .uplot .u-cursor-y { border-color: #585b70 !important; }
+.uplot .u-select { background: rgba(137,180,250,0.1) !important; border: 1px solid rgba(137,180,250,0.4) !important; }
+
+/* ── Top bar ─────────────────────────────────────────────────── */
+#topbar {
+  position:fixed; top:0; left:0; right:0; height:var(--topbar-h);
+  background:var(--mantle); border-bottom:1px solid var(--surface0);
+  display:flex; align-items:center; gap:10px; padding:0 14px;
+  z-index:100; box-shadow:0 2px 8px rgba(0,0,0,0.4); overflow:hidden;
+}
+#app-title { font-weight:700; font-size:15px; color:var(--accent);
+  white-space:nowrap; letter-spacing:0.3px; flex-shrink:0; }
+.topbar-sep { flex:1; min-width:4px; }
+#status-led { width:8px; height:8px; border-radius:50%;
+  background:var(--red); flex-shrink:0; transition:background var(--transition); }
+#status-led.green  { background:var(--green);  animation:pulse-green  2s infinite; }
+#status-led.orange { background:var(--yellow); animation:pulse-orange 1.5s infinite; }
+@keyframes pulse-green  { 0%,100%{box-shadow:0 0 0 0 rgba(166,227,161,0.4)} 50%{box-shadow:0 0 0 4px rgba(166,227,161,0)} }
+@keyframes pulse-orange { 0%,100%{box-shadow:0 0 0 0 rgba(249,226,175,0.4)} 50%{box-shadow:0 0 0 4px rgba(249,226,175,0)} }
+#status-text { font-size:11px; color:var(--subtext0); white-space:nowrap; }
+#sb-tsage    { font-size:11px; color:var(--overlay0);  white-space:nowrap; font-family:monospace; }
+
+#cursor-readout {
+  display:none; align-items:center; gap:8px;
+  font-size:11px; font-family:monospace;
+  background:var(--surface0); border:1px solid var(--surface1);
+  border-radius:5px; padding:3px 8px; white-space:nowrap; flex-shrink:0;
+}
+#cursor-readout.visible { display:flex; }
+#cur-ta { color:var(--sky); } #cur-tb { color:var(--yellow); }
+#cur-dt { color:var(--subtext1); } .cur-sep { color:var(--surface2); }
+
+.topbar-vsep { width:1px; height:22px; background:var(--surface0); flex-shrink:0; margin:0 2px; }
+#layout-btns { display:flex; gap:2px; align-items:center; flex-shrink:0; }
+.ctrl-label { font-size:12px; color:var(--subtext0); white-space:nowrap; flex-shrink:0; }
+select.ctrl-select {
+  background:var(--surface0); color:var(--text);
+  border:1px solid var(--surface1); border-radius:var(--radius);
+  padding:4px 6px; font-size:12px; cursor:pointer; outline:none; flex-shrink:0;
+}
+select.ctrl-select:hover { border-color:var(--accent); }
+button.ctrl-btn {
+  background:var(--surface0); color:var(--text);
+  border:1px solid var(--surface1); border-radius:var(--radius);
+  padding:4px 12px; font-size:12px; cursor:pointer; white-space:nowrap; flex-shrink:0;
+  transition:background var(--transition),border-color var(--transition);
+}
+button.ctrl-btn:hover  { background:var(--surface1); border-color:var(--accent); }
+button.ctrl-btn.active { background:var(--surface1); border-color:var(--accent); color:var(--accent); }
+button.ctrl-btn.trig-active { background:rgba(203,166,247,0.15); border-color:var(--mauve); color:var(--mauve); }
+button.ctrl-btn.cursor-a    { border-color:var(--sky);   color:var(--sky); }
+button.ctrl-btn.cursor-b    { border-color:var(--yellow); color:var(--yellow); }
+button.ctrl-btn.resume-btn  { border-color:var(--teal);  color:var(--teal); }
+
+/* ── Trigger bar ──────────────────────────────────────────────── */
+#trigbar {
+  position:fixed; top:var(--topbar-h); left:0; right:0;
+  background:var(--crust); border-bottom:1px solid var(--surface0);
+  display:flex; align-items:center; flex-wrap:wrap; gap:10px;
+  padding:0 16px; z-index:99;
+  height:0; overflow:hidden;
+  transition:height var(--transition),padding var(--transition);
+}
+#trigbar.open { height:48px; padding:0 16px; }
+.trig-group { display:flex; align-items:center; gap:6px; }
+.trig-sep  { width:1px; height:24px; background:var(--surface0); flex-shrink:0; }
+.trig-label { font-size:11px; color:var(--subtext0); white-space:nowrap; }
+select.trig-select {
+  background:var(--surface0); color:var(--text);
+  border:1px solid var(--surface1); border-radius:5px;
+  padding:3px 6px; font-size:12px; cursor:pointer; outline:none;
+}
+select.trig-select:hover { border-color:var(--mauve); }
+input.trig-input {
+  background:var(--surface0); color:var(--text);
+  border:1px solid var(--surface1); border-radius:5px;
+  padding:3px 6px; font-size:12px; outline:none; width:80px;
+}
+input.trig-input:focus { border-color:var(--mauve); }
+input[type=range].trig-range {
+  -webkit-appearance:none; width:90px; height:4px;
+  background:var(--surface1); border-radius:2px; outline:none; cursor:pointer;
+}
+input[type=range].trig-range::-webkit-slider-thumb {
+  -webkit-appearance:none; width:12px; height:12px;
+  border-radius:50%; background:var(--mauve); cursor:pointer;
+}
+.trig-range-val { font-size:11px; color:var(--mauve); min-width:28px; }
+#trig-status-badge {
+  font-size:11px; font-weight:700; letter-spacing:0.8px;
+  padding:3px 10px; border-radius:12px;
+  border:1px solid var(--surface1); background:var(--surface0); color:var(--subtext0);
+  min-width:80px; text-align:center; white-space:nowrap;
+}
+#trig-status-badge.armed     { background:rgba(166,227,161,0.12); border-color:var(--green);  color:var(--green); }
+#trig-status-badge.waiting   { background:rgba(249,226,175,0.12); border-color:var(--yellow); color:var(--yellow); }
+#trig-status-badge.triggered { background:rgba(203,166,247,0.15); border-color:var(--mauve);  color:var(--mauve); }
+#btn-trig-rearm, #btn-trig-stop {
+  border:none; border-radius:5px;
+  padding:4px 12px; font-size:12px; font-weight:600; cursor:pointer; display:none;
+}
+#btn-trig-rearm { background:var(--mauve); color:var(--crust); }
+#btn-trig-stop  { background:var(--surface1); color:var(--yellow); border:1px solid var(--yellow); }
+#btn-trig-rearm:hover, #btn-trig-stop:hover { opacity:0.85; }
+
+/* ── Status bar ───────────────────────────────────────────────── */
+#statusbar {
+  position:fixed; bottom:0; left:0; right:0; height:var(--statusbar-h);
+  background:var(--crust); border-top:1px solid var(--surface0);
+  display:flex; align-items:center; justify-content:space-between;
+  padding:0 10px; z-index:100;
+}
+#sb-left { display:flex; align-items:center; gap:6px; }
+#build-version { font-size:10px; color:var(--overlay0); font-family:monospace; white-space:nowrap; }
+
+/* ── Body ─────────────────────────────────────────────────────── */
+#body {
+  position:fixed; top:calc(var(--topbar-h) + var(--trigbar-h));
+  left:0; right:0; bottom:var(--statusbar-h); display:flex; overflow:hidden;
+  transition:top var(--transition);
+}
+
+/* ── Sidebar ──────────────────────────────────────────────────── */
+#sidebar {
+  width:var(--sidebar-w); min-width:var(--sidebar-w);
+  background:var(--mantle); border-right:1px solid var(--surface0);
+  display:flex; flex-direction:column;
+  transition:width var(--transition),min-width var(--transition); overflow:hidden;
+}
+#sidebar.collapsed { width:0; min-width:0; }
+#sidebar-header {
+  display:flex; align-items:center; justify-content:space-between;
+  padding:10px 14px; border-bottom:1px solid var(--surface0);
+  font-weight:600; color:var(--subtext1); font-size:12px;
+  text-transform:uppercase; letter-spacing:0.8px; flex-shrink:0;
+}
+#signal-list { flex:1; overflow-y:auto; padding:8px 0; }
+.sig-item {
+  padding:6px 14px; cursor:grab; border-radius:6px; margin:1px 6px;
+  transition:background var(--transition); display:flex; align-items:center; gap:8px;
+  user-select:none;
+}
+.sig-item:hover  { background:var(--surface0); }
+.sig-item:active { cursor:grabbing; }
+.sig-item.dragging { opacity:0.4; }
+.sig-name  { flex:1; font-size:13px; color:var(--text); overflow:hidden; text-overflow:ellipsis; white-space:nowrap; }
+.sig-unit  { font-size:11px; color:var(--subtext0); font-style:italic; }
+.type-badge { font-size:10px; background:var(--surface1); color:var(--subtext1); padding:1px 5px; border-radius:3px; white-space:nowrap; }
+.array-group {}
+.array-header {
+  padding:6px 14px 6px 10px; cursor:pointer; border-radius:6px; margin:1px 6px;
+  transition:background var(--transition); display:flex; align-items:center; gap:6px; user-select:none;
+}
+.array-header:hover { background:var(--surface0); }
+.array-arrow { font-size:10px; color:var(--subtext0); transition:transform var(--transition); display:inline-block; }
+.array-header.open .array-arrow { transform:rotate(90deg); }
+.array-children { display:none; padding-left:16px; }
+.array-header.open + .array-children { display:block; }
+.array-child {
+  padding:4px 14px 4px 8px; cursor:grab; border-radius:6px; margin:1px 6px;
+  transition:background var(--transition); display:flex; align-items:center; gap:8px;
+  user-select:none; color:var(--subtext1); font-size:12px;
+}
+.array-child:hover  { background:var(--surface0); }
+.array-child:active { cursor:grabbing; }
+
+/* ── Main area ────────────────────────────────────────────────── */
+#main { flex:1; display:flex; flex-direction:column; overflow:hidden; min-width:0; }
+/* Layout toggle button in topbar */
+#btn-layout {
+  display:flex; align-items:center; gap:4px;
+  font-size:11px; padding:3px 8px;
+}
+#btn-layout svg { flex-shrink:0; }
+#btn-layout span { flex-shrink:0; }
+
+/* Layout dropdown menu */
+#layout-menu {
+  position:fixed; z-index:200;
+  background:var(--mantle); border:1px solid var(--surface1); border-radius:var(--radius);
+  box-shadow:0 6px 20px rgba(0,0,0,0.5);
+  display:none; grid-template-columns:1fr 1fr; gap:4px; padding:6px;
+}
+#layout-menu.open { display:grid; }
+.layout-menu-item {
+  display:flex; flex-direction:column; align-items:center; gap:3px;
+  padding:5px 10px; cursor:pointer;
+  background:var(--surface0); border:1px solid var(--surface1); border-radius:6px;
+  color:var(--subtext1); font-size:10px; font-family:monospace;
+  transition:background var(--transition),border-color var(--transition),color var(--transition);
+}
+.layout-menu-item:hover  { background:var(--surface1); border-color:var(--accent); color:var(--accent); }
+.layout-menu-item.active { background:var(--surface1); border-color:var(--accent); color:var(--accent); }
+
+/* ── Plot grid ────────────────────────────────────────────────── */
+#plot-grid {
+  flex:1; min-height:0; display:grid; gap:0; padding:0; overflow:hidden;
+  border-top:1px solid var(--surface0); border-left:1px solid var(--surface0);
+}
+#plot-grid.l1x1 { grid-template-columns:1fr;         grid-template-rows:1fr; }
+#plot-grid.l2x1 { grid-template-columns:1fr 1fr;     grid-template-rows:1fr; }
+#plot-grid.l1x2 { grid-template-columns:1fr;         grid-template-rows:1fr 1fr; }
+#plot-grid.l2x2 { grid-template-columns:1fr 1fr;     grid-template-rows:1fr 1fr; }
+#plot-grid.l3x1 { grid-template-columns:1fr 1fr 1fr; grid-template-rows:1fr; }
+#plot-grid.l1x3 { grid-template-columns:1fr;         grid-template-rows:1fr 1fr 1fr; }
+#plot-grid.l3x2 { grid-template-columns:1fr 1fr 1fr; grid-template-rows:1fr 1fr; }
+#plot-grid.l2x3 { grid-template-columns:1fr 1fr;     grid-template-rows:1fr 1fr 1fr; }
+#plot-grid.l1x4 { grid-template-columns:1fr;         grid-template-rows:1fr 1fr 1fr 1fr; }
+#plot-grid.l4x1 { grid-template-columns:1fr 1fr 1fr 1fr; grid-template-rows:1fr; }
+
+/* ── Plot card ────────────────────────────────────────────────── */
+.plot-card {
+  background:var(--bg);
+  border-right:1px solid var(--surface0); border-bottom:1px solid var(--surface0);
+  border-radius:0; display:flex; flex-direction:column;
+  min-height:0; position:relative; overflow:hidden;
+}
+.plot-card.drag-over { background:rgba(137,180,250,0.04); box-shadow:inset 0 0 0 2px var(--accent); }
+/* Header: always visible, part of normal card flex flow */
+.plot-card-header {
+  z-index:5; flex-shrink:0;
+  background:rgba(17,17,27,0.88); backdrop-filter:blur(6px);
+  border-bottom:1px solid var(--surface1);
+  display:flex; align-items:center; gap:5px;
+  padding:3px 8px; min-height:26px; overflow:hidden;
+}
+.plot-title {
+  font-size:11px; color:var(--subtext1); font-weight:600;
+  cursor:text; outline:none; border:1px solid transparent; border-radius:3px;
+  padding:1px 3px; background:transparent; white-space:nowrap;
+  flex-shrink:0; max-width:80px; overflow:hidden; text-overflow:ellipsis;
+}
+.plot-title:focus { border-color:var(--accent); background:var(--surface1); color:var(--text); }
+.sig-badges { display:flex; flex-wrap:nowrap; gap:3px; flex:1; overflow:hidden; min-width:0; }
+.sig-badge {
+  display:inline-flex; align-items:center; gap:3px;
+  background:rgba(69,71,90,0.6); color:var(--subtext1);
+  border-radius:10px; padding:1px 6px 1px 4px;
+  font-size:10px; white-space:nowrap; flex-shrink:0;
+}
+.trace-dot { width:7px; height:7px; border-radius:50%; flex-shrink:0; display:inline-block; }
+.sig-badge-x {
+  cursor:pointer; color:var(--overlay0); font-size:11px; line-height:1; margin-left:1px;
+  transition:color var(--transition);
+}
+.sig-badge-x:hover { color:var(--red); }
+.plot-body { flex:1; position:relative; min-height:0; overflow:hidden; }
+.drop-hint {
+  position:absolute; inset:0; display:flex; align-items:center; justify-content:center;
+  color:var(--overlay0); font-size:13px; pointer-events:none;
+}
+.trig-collect-overlay {
+  position:absolute; inset:0; display:none;
+  align-items:center; justify-content:center; pointer-events:none; z-index:10;
+}
+.plot-card.trig-collecting .trig-collect-overlay { display:flex; }
+.trig-collect-text {
+  background:rgba(49,50,68,0.82); color:var(--mauve);
+  font-size:11px; font-weight:600; padding:4px 12px; border-radius:20px;
+  border:1px solid var(--mauve);
+}
+
+/* ── Signal style context menu ────────────────────────────────── */
+#sig-ctx-menu {
+  position:fixed; z-index:300;
+  background:var(--mantle); border:1px solid var(--surface1); border-radius:var(--radius);
+  box-shadow:0 8px 24px rgba(0,0,0,0.6); padding:10px; min-width:210px;
+}
+.ctx-menu-header {
+  font-size:11px; color:var(--subtext0); margin-bottom:8px;
+  padding-bottom:6px; border-bottom:1px solid var(--surface0);
+  white-space:nowrap; overflow:hidden; text-overflow:ellipsis;
+}
+.ctx-menu-key { color:var(--accent); font-weight:600; }
+.ctx-row { display:flex; align-items:center; gap:8px; margin-bottom:6px; }
+.ctx-row label { font-size:11px; color:var(--subtext0); width:42px; flex-shrink:0; }
+.ctx-btns { display:flex; gap:3px; flex-wrap:wrap; }
+.ctx-btn {
+  background:var(--surface0); border:1px solid var(--surface1); border-radius:4px;
+  color:var(--subtext1); font-size:11px; padding:2px 7px; cursor:pointer;
+  transition:background var(--transition),border-color var(--transition),color var(--transition);
+}
+.ctx-btn:hover  { background:var(--surface1); border-color:var(--accent); }
+.ctx-btn.active { background:var(--surface1); border-color:var(--accent); color:var(--accent); }
+#ctx-color {
+  width:28px; height:22px; border:1px solid var(--surface1); border-radius:4px;
+  background:transparent; cursor:pointer; padding:1px;
+}
+.ctx-range { width:80px; }
+.ctx-range-val { font-size:11px; color:var(--mauve); min-width:26px; }
+
+/* ── Source groups ────────────────────────────────────────────── */
+.source-group { margin-bottom:2px; }
+.source-group-header {
+  display:flex; align-items:center; gap:5px;
+  padding:5px 8px 5px 10px; margin:4px 6px 2px;
+  background:var(--surface0); border-radius:6px;
+}
+.source-state-dot {
+  width:7px; height:7px; border-radius:50%; flex-shrink:0;
+  background:var(--overlay0); transition:background var(--transition);
+}
+.source-state-dot.connected    { background:var(--green); }
+.source-state-dot.connecting   { background:var(--yellow); animation:pulse-orange 1.5s infinite; }
+.source-state-dot.disconnected { background:var(--red); }
+.source-name {
+  font-size:11px; font-weight:600; color:var(--subtext1); flex:1;
+  overflow:hidden; text-overflow:ellipsis; white-space:nowrap;
+}
+.source-addr {
+  font-size:10px; color:var(--overlay0); font-family:monospace;
+  overflow:hidden; text-overflow:ellipsis; white-space:nowrap; max-width:90px;
+}
+.source-remove-btn {
+  background:none; border:none; color:var(--overlay0);
+  cursor:pointer; font-size:15px; line-height:1; padding:0 2px; flex-shrink:0;
+  transition:color var(--transition);
+}
+.source-remove-btn:hover { color:var(--red); }
+
+/* ── Add source section ───────────────────────────────────────── */
+.add-source-section {
+  border-top:1px solid var(--surface0); margin-top:4px;
+}
+.add-source-title {
+  display:flex; align-items:center; gap:5px;
+  padding:6px 10px; cursor:pointer; user-select:none;
+  font-size:11px; color:var(--overlay0);
+  transition:color var(--transition);
+}
+.add-source-title:hover { color:var(--subtext1); }
+.add-src-arrow {
+  font-size:9px; display:inline-block;
+  transition:transform var(--transition);
+}
+.add-source-body {
+  display:none; flex-direction:column; gap:5px;
+  padding:0 10px 10px;
+}
+.add-source-section.open .add-source-body { display:flex; }
+.add-src-input {
+  background:var(--surface0); color:var(--text);
+  border:1px solid var(--surface1); border-radius:5px;
+  padding:4px 8px; font-size:12px; outline:none; width:100%;
+}
+.add-src-input:focus { border-color:var(--accent); }
+.add-src-btn {
+  background:var(--surface0); color:var(--accent);
+  border:1px solid var(--surface1); border-radius:5px;
+  padding:4px 10px; font-size:12px; cursor:pointer; width:100%;
+  transition:background var(--transition),border-color var(--transition);
+}
+.add-src-btn:hover { background:rgba(137,180,250,0.15); border-color:var(--accent); }
+.save-src-btn { color:var(--green); }
+.save-src-btn:hover { background:rgba(166,227,161,0.1); border-color:var(--green); }
+
+/* ── Empty state ──────────────────────────────────────────────── */
+#empty-state {
+  position:absolute; top:50%; left:50%; transform:translate(-50%,-50%);
+  text-align:center; color:var(--subtext0); pointer-events:none; display:none;
+}
+#empty-state.visible { display:block; }
+#empty-state h2 { font-size:20px; margin-bottom:8px; color:var(--surface2); }
+#empty-state p   { font-size:13px; }
+
+@media (max-width:700px) { #sidebar { width:0; min-width:0; } :root { --sidebar-w:240px; } }

Client/WebUI/static/uPlot.min.css

@@ -0,0 +1 @@
+.uplot, .uplot *, .uplot *::before, .uplot *::after {box-sizing: border-box;}.uplot {font-family: system-ui, -apple-system, "Segoe UI", Roboto, "Helvetica Neue", Arial, "Noto Sans", sans-serif, "Apple Color Emoji", "Segoe UI Emoji", "Segoe UI Symbol", "Noto Color Emoji";line-height: 1.5;width: min-content;}.u-title {text-align: center;font-size: 18px;font-weight: bold;}.u-wrap {position: relative;user-select: none;}.u-over, .u-under {position: absolute;}.u-under {overflow: hidden;}.uplot canvas {display: block;position: relative;width: 100%;height: 100%;}.u-axis {position: absolute;}.u-legend {font-size: 14px;margin: auto;text-align: center;}.u-inline {display: block;}.u-inline * {display: inline-block;}.u-inline tr {margin-right: 16px;}.u-legend th {font-weight: 600;}.u-legend th > * {vertical-align: middle;display: inline-block;}.u-legend .u-marker {width: 1em;height: 1em;margin-right: 4px;background-clip: padding-box !important;}.u-inline.u-live th::after {content: ":";vertical-align: middle;}.u-inline:not(.u-live) .u-value {display: none;}.u-series > * {padding: 4px;}.u-series th {cursor: pointer;}.u-legend .u-off > * {opacity: 0.3;}.u-select {background: rgba(0,0,0,0.07);position: absolute;pointer-events: none;}.u-cursor-x, .u-cursor-y {position: absolute;left: 0;top: 0;pointer-events: none;will-change: transform;}.u-hz .u-cursor-x, .u-vt .u-cursor-y {height: 100%;border-right: 1px dashed #607D8B;}.u-hz .u-cursor-y, .u-vt .u-cursor-x {width: 100%;border-bottom: 1px dashed #607D8B;}.u-cursor-pt {position: absolute;top: 0;left: 0;border-radius: 50%;border: 0 solid;pointer-events: none;will-change: transform;/*this has to be !important since we set inline "background" shorthand */background-clip: padding-box !important;}.u-axis.u-off, .u-select.u-off, .u-cursor-x.u-off, .u-cursor-y.u-off, .u-cursor-pt.u-off {display: none;}

Client/WebUI/udpclient.go

@@ -12,20 +12,20 @@ const (
 	readBufSize    = 65536
 )
 
-// UDPClient manages the UDP connection to the MARTe2 streamer.
+// UDPClient manages the UDP connection to one MARTe2 streamer source.
 type UDPClient struct {
 	serverAddr string
-	localPort  int
+	sourceID   string
 	hub        *Hub
 
 	stopCh chan struct{}
 }
 
-// NewUDPClient creates a UDPClient. Call Run() in a goroutine.
+// NewUDPClient creates a UDPClient bound to a specific source ID. Call Run() in a goroutine.
-func NewUDPClient(serverAddr string, localPort int, hub *Hub) *UDPClient {
+func NewUDPClient(serverAddr, sourceID string, hub *Hub) *UDPClient {
 	return &UDPClient{
 		serverAddr: serverAddr,
-		localPort:  localPort,
+		sourceID:   sourceID,
 		hub:        hub,
 		stopCh:     make(chan struct{}),
 	}
@@ -45,13 +45,14 @@ func (u *UDPClient) Run() {
 		default:
 		}
 
-		log.Printf("udp: connecting to %s (local port %d)", u.serverAddr, u.localPort)
+		u.hub.SetSourceState(u.sourceID, "connecting")
+		log.Printf("[%s] udp: connecting to %s", u.sourceID, u.serverAddr)
 		err := u.runSession()
 		if err != nil {
-			log.Printf("udp: session ended: %v", err)
+			log.Printf("[%s] udp: session ended: %v", u.sourceID, err)
 		}
+		u.hub.SetSourceState(u.sourceID, "disconnected")
 
-		// Check stop before sleeping.
 		select {
 		case <-u.stopCh:
 			return
@@ -63,8 +64,8 @@ func (u *UDPClient) Run() {
 // runSession opens a UDP socket, sends CONNECT, reads data until the server
 // goes silent or an error occurs.
 func (u *UDPClient) runSession() error {
-	localAddr := &net.UDPAddr{Port: u.localPort}
+	// Port 0 lets the OS pick a free local port.
-	conn, err := net.ListenUDP("udp4", localAddr)
+	conn, err := net.ListenUDP("udp4", &net.UDPAddr{})
 	if err != nil {
 		return err
 	}
@@ -75,36 +76,33 @@ func (u *UDPClient) runSession() error {
 		return err
 	}
 
-	// Send CONNECT.
 	if _, err := conn.WriteToUDP(BuildConnectPacket(), serverAddr); err != nil {
 		return err
 	}
-	log.Printf("udp: sent CONNECT")
+	log.Printf("[%s] udp: sent CONNECT", u.sourceID)
 
 	reassembler := NewReassembler(2 * time.Second)
 	buf := make([]byte, readBufSize)
 	var currentSigs []SignalInfo
 
 	for {
-		// Silence timeout.
 		conn.SetReadDeadline(time.Now().Add(silenceTimeout))
 
 		n, _, err := conn.ReadFromUDP(buf)
 		arrivalTime := time.Now()
 		if err != nil {
-			// Send DISCONNECT best-effort before returning.
 			conn.WriteToUDP(BuildDisconnectPacket(), serverAddr)
 			return err
 		}
 
 		if n < HeaderSize {
-			log.Printf("udp: short datagram (%d bytes), skipping", n)
+			log.Printf("[%s] udp: short datagram (%d bytes), skipping", u.sourceID, n)
 			continue
 		}
 
 		hdr, err := ParseHeader(buf[:n])
 		if err != nil {
-			log.Printf("udp: parse header: %v", err)
+			log.Printf("[%s] udp: parse header: %v", u.sourceID, err)
 			continue
 		}
 
@@ -120,37 +118,36 @@ func (u *UDPClient) runSession() error {
 		case PktConfig:
 			sigs, err := ParseConfig(complete)
 			if err != nil {
-				log.Printf("udp: parse config: %v", err)
+				log.Printf("[%s] udp: parse config: %v", u.sourceID, err)
 				continue
 			}
 			currentSigs = sigs
-			log.Printf("udp: received CONFIG (%d signals)", len(sigs))
+			log.Printf("[%s] udp: received CONFIG (%d signals)", u.sourceID, len(sigs))
-			u.hub.UpdateConfig(sigs)
+			u.hub.SetSourceState(u.sourceID, "connected")
+			u.hub.UpdateConfigForSource(u.sourceID, sigs)
 
 		case PktData:
 			if len(currentSigs) == 0 {
-				// We haven't received a CONFIG yet – ignore.
 				continue
 			}
 			sample, err := ParseData(complete, currentSigs, arrivalTime)
 			if err != nil {
-				log.Printf("udp: parse data: %v", err)
+				log.Printf("[%s] udp: parse data: %v", u.sourceID, err)
 				continue
 			}
-			u.hub.PushData(sample)
+			u.hub.PushDataForSource(u.sourceID, sample)
 
 		case PktACK:
-			log.Printf("udp: received ACK (counter=%d)", hdr.Counter)
+			log.Printf("[%s] udp: received ACK (counter=%d)", u.sourceID, hdr.Counter)
 
 		case PktDisconnect:
-			log.Printf("udp: server sent DISCONNECT")
+			log.Printf("[%s] udp: server sent DISCONNECT", u.sourceID)
 			return nil
 
 		default:
-			log.Printf("udp: unknown packet type %d", hdr.Type)
+			log.Printf("[%s] udp: unknown packet type %d", u.sourceID, hdr.Type)
 		}
 
-		// Check stop request.
 		select {
 		case <-u.stopCh:
 			conn.WriteToUDP(BuildDisconnectPacket(), serverAddr)

Makefile.inc

@@ -27,8 +27,7 @@
 #If SPB is directly exported as an environment variable it will also be evaluated as part of the subprojects SPB, thus
 #potentially overriding its value
 #Main target subprojects. May be overridden by shell definition.
-SPBM?=Source/Components/DataSources.x 
+SPBM?=Source/Components/DataSources.x Source/Components/GAMs.x 
-#	Source/Components/GAMs.x \
 	
 	
 

Source/Components/DataSources/UDPStreamer/Makefile.inc

@@ -22,8 +22,7 @@
 #
 #############################################################
 
-OBJSX = UDPStreamer.x \
+OBJSX = UDPStreamer.x
-        SineArrayGAM.x
 
 PACKAGE=Components/DataSources
 ROOT_DIR=../../../../

Source/Components/DataSources/UDPStreamer/UDPStreamer.cpp

@@ -57,8 +57,11 @@ static const uint32 UDPS_DEFAULT_MAX_PAYLOAD = 1400u;
 /** Minimum MaxPayloadSize: header + at least 1 byte of payload. */
 static const uint32 UDPS_MIN_PAYLOAD = static_cast<uint32>(sizeof(UDPSPacketHeader)) + 1u;
 
-/** Server socket receive timeout in milliseconds. */
+/** Server socket receive timeout in milliseconds.
-static const uint32 UDPS_RECV_TIMEOUT_MS = 5u;
+ *  Set to 0 for a pure non-blocking poll so the send loop can keep pace with
+ *  the RT thread regardless of its frequency.  Client commands (CONNECT /
+ *  DISCONNECT) are still caught on the very next loop iteration. */
+static const uint32 UDPS_RECV_TIMEOUT_MS = 0u;
 
 /** EventSem wait timeout in milliseconds for the data loop. */
 static const uint32 UDPS_DATA_WAIT_MS = 10u;
@@ -615,18 +618,24 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
     }
 
     if (info.GetStage() == ExecutionInfo::MainStage) {
-        /* --- Poll server socket for incoming client commands ---
+        /* --- Wait for RT thread to post new data ---
-         * Use select() directly so we get a silent timeout with no log spam.
+         * ResetWait sleeps the background thread until the RT thread calls
-         * MARTe2's Read(buf, size, timeout) calls recvfrom() and logs an error
+         * Synchronise() and posts dataSem, or until the timeout expires.
-         * on every timeout (EAGAIN from SO_RCVTIMEO). */
+         * Doing this FIRST means the thread spends nearly all its time here
+         * instead of spinning on the non-blocking select() below.
+         * Command latency is bounded by UDPS_DATA_WAIT_MS (acceptable for
+         * CONNECT / DISCONNECT). */
+        ErrorManagement::ErrorType waitErr =
+            dataSem.ResetWait(TimeoutType(UDPS_DATA_WAIT_MS));
+        bool dataReady = (waitErr == ErrorManagement::NoError);
+
+        /* --- Poll server socket for incoming client commands (non-blocking) --- */
         uint8 cmdBuf[256u];
         Handle sockFd = serverSocket.GetReadHandle();
         fd_set rfds;
         FD_ZERO(&rfds);
         FD_SET(static_cast<int>(sockFd), &rfds);
-        struct timeval tv;
+        struct timeval tv = { 0, 0 };  /* non-blocking poll */
-        tv.tv_sec  = 0;
-        tv.tv_usec = static_cast<long>(UDPS_RECV_TIMEOUT_MS) * 1000L;
         int nReady = select(static_cast<int>(sockFd) + 1, &rfds, NULL_PTR(fd_set *), NULL_PTR(fd_set *), &tv);
         if (nReady > 0) {
             uint32 recvSize = static_cast<uint32>(sizeof(cmdBuf));
@@ -636,12 +645,6 @@ ErrorManagement::ErrorType UDPStreamer::Execute(ExecutionInfo &info) {
             }
         }
 
-        /* --- Wait for RT thread to post new data --- */
-        /* ResetWait atomically lowers the barrier then waits, preventing missed posts */
-        ErrorManagement::ErrorType waitErr =
-            dataSem.ResetWait(TimeoutType(UDPS_DATA_WAIT_MS));
-        bool dataReady = (waitErr == ErrorManagement::NoError);
-
         if (dataReady && clientConnected) {
             /* Copy readyBuffer → scratchBuffer under brief spinlock */
             uint64 ts = 0u;

Source/Components/GAMs/Makefile.gcc

@@ -0,0 +1,25 @@
+#############################################################
+#
+# Copyright 2015 F4E | European Joint Undertaking for ITER
+#  and the Development of Fusion Energy ('Fusion for Energy')
+#
+# Licensed under the EUPL, Version 1.1 or - as soon they
+# will be approved by the European Commission - subsequent
+# versions of the EUPL (the "Licence");
+# You may not use this work except in compliance with the
+# Licence.
+# You may obtain a copy of the Licence at:
+#
+# http://ec.europa.eu/idabc/eupl
+#
+# Unless required by applicable law or agreed to in
+# writing, software distributed under the Licence is
+# distributed on an "AS IS" basis,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+# express or implied.
+# See the Licence for the specific language governing
+# permissions and limitations under the Licence.
+#
+#############################################################
+
+include Makefile.inc

Source/Components/GAMs/Makefile.inc

@@ -0,0 +1,38 @@
+#############################################################
+#
+# Copyright 2015 F4E | European Joint Undertaking for ITER
+#  and the Development of Fusion Energy ('Fusion for Energy')
+#
+# Licensed under the EUPL, Version 1.1 or - as soon they
+# will be approved by the European Commission - subsequent
+# versions of the EUPL (the "Licence");
+# You may not use this work except in compliance with the
+# Licence.
+# You may obtain a copy of the Licence at:
+#
+# http://ec.europa.eu/idabc/eupl
+#
+# Unless required by applicable law or agreed to in
+# writing, software distributed under the Licence is
+# distributed on an "AS IS" basis,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+# express or implied.
+# See the Licence for the specific language governing
+# permissions and limitations under the Licence.
+#
+#############################################################
+
+OBJSX=
+
+SPB = SineArrayGAM.x
+
+PACKAGE=Components
+ROOT_DIR=../../..
+MAKEDEFAULTDIR=$(MARTe2_DIR)/MakeDefaults
+
+include $(MAKEDEFAULTDIR)/MakeStdLibDefs.$(TARGET)
+
+all: $(OBJS) $(SUBPROJ)
+	echo $(OBJS)
+
+include $(MAKEDEFAULTDIR)/MakeStdLibRules.$(TARGET)

Source/Components/GAMs/SineArrayGAM/Makefile.gcc

@@ -0,0 +1,25 @@
+#############################################################
+#
+# Copyright 2015 F4E | European Joint Undertaking for ITER
+#  and the Development of Fusion Energy ('Fusion for Energy')
+#
+# Licensed under the EUPL, Version 1.1 or - as soon they
+# will be approved by the European Commission - subsequent
+# versions of the EUPL (the "Licence");
+# You may not use this work except in compliance with the
+# Licence.
+# You may obtain a copy of the Licence at:
+#
+# http://ec.europa.eu/idabc/eupl
+#
+# Unless required by applicable law or agreed to in
+# writing, software distributed under the Licence is
+# distributed on an "AS IS" basis,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+# express or implied.
+# See the Licence for the specific language governing
+# permissions and limitations under the Licence.
+#
+#############################################################
+
+include Makefile.inc

Source/Components/GAMs/SineArrayGAM/Makefile.inc

@@ -0,0 +1,52 @@
+#############################################################
+#
+# Copyright 2015 F4E | European Joint Undertaking for ITER
+#  and the Development of Fusion Energy ('Fusion for Energy')
+#
+# Licensed under the EUPL, Version 1.1 or - as soon they
+# will be approved by the European Commission - subsequent
+# versions of the EUPL (the "Licence");
+# You may not use this work except in compliance with the
+# Licence.
+# You may obtain a copy of the Licence at:
+#
+# http://ec.europa.eu/idabc/eupl
+#
+# Unless required by applicable law or agreed to in
+# writing, software distributed under the Licence is
+# distributed on an "AS IS" basis,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+# express or implied.
+# See the Licence for the specific language governing
+# permissions and limitations under the Licence.
+#
+#############################################################
+
+OBJSX = SineArrayGAM.x
+
+PACKAGE=Components/GAMs
+ROOT_DIR=../../../../
+MAKEDEFAULTDIR=$(MARTe2_DIR)/MakeDefaults
+
+include $(MAKEDEFAULTDIR)/MakeStdLibDefs.$(TARGET)
+
+INCLUDES += -I.
+INCLUDES += -I$(MARTe2_DIR)/Source/Core/BareMetal/L0Types
+INCLUDES += -I$(MARTe2_DIR)/Source/Core/BareMetal/L1Portability
+INCLUDES += -I$(MARTe2_DIR)/Source/Core/BareMetal/L2Objects
+INCLUDES += -I$(MARTe2_DIR)/Source/Core/BareMetal/L3Streams
+INCLUDES += -I$(MARTe2_DIR)/Source/Core/BareMetal/L4Messages
+INCLUDES += -I$(MARTe2_DIR)/Source/Core/BareMetal/L4Configuration
+INCLUDES += -I$(MARTe2_DIR)/Source/Core/BareMetal/L5GAMs
+INCLUDES += -I$(MARTe2_DIR)/Source/Core/Scheduler/L1Portability
+INCLUDES += -I$(MARTe2_DIR)/Source/Core/Scheduler/L3Services
+INCLUDES += -I$(MARTe2_DIR)/Source/Core/Scheduler/L4Messages
+
+all: $(OBJS) \
+    $(BUILD_DIR)/SineArrayGAM$(LIBEXT) \
+    $(BUILD_DIR)/SineArrayGAM$(DLLEXT)
+	echo $(OBJS)
+
+-include depends.$(TARGET)
+
+include $(MAKEDEFAULTDIR)/MakeStdLibRules.$(TARGET)

Test/MARTeApp/TestApp.cfg

@@ -46,6 +46,23 @@ $TestApp = {
                 }
             }
         }
+        +FastTimerGAM = {
+            Class = IOGAM
+            InputSignals = {
+                Time = {
+                    Frequency = 5000
+                    DataSource = FastTimer
+                    Type = uint32
+                }
+            }
+            OutputSignals = {
+                Time = {
+                    DataSource = DDB2
+                    Type = uint32
+                }
+            }
+            
+        }
 
         // ── 1 Hz sinusoidal signal ───────────────────────────────────────────
         +SineGAM1 = {
@@ -99,7 +116,7 @@ $TestApp = {
             SamplingRate = 1000000.0
             OutputSignals = {
                 Ch1 = {
-                    DataSource = DDB
+                    DataSource = DDB2
                     Type = float32
                     NumberOfDimensions = 1
                     NumberOfElements = 1000
@@ -110,14 +127,14 @@ $TestApp = {
         // ── 1 kHz sine burst – channel 2 (phase-shifted by π/2) ──────────────
         +SineGAM4 = {
             Class = SineArrayGAM
-            Frequency    = 1000.0
+            Frequency    = 10000.0
             Amplitude    = 0.5
             Phase        = 1.5708
             Offset       = 0.0
             SamplingRate = 1000000.0
             OutputSignals = {
                 Ch2 = {
-                    DataSource = DDB
+                    DataSource = DDB2
                     Type = float32
                     NumberOfDimensions = 1
                     NumberOfElements = 1000
@@ -145,18 +162,6 @@ $TestApp = {
                     DataSource = DDB
                     Type = float32
                 }
-                Ch1 = {
-                    DataSource = DDB
-                    Type = float32
-                    NumberOfDimensions = 1
-                    NumberOfElements = 1000
-                }
-                Ch2 = {
-                    DataSource = DDB
-                    Type = float32
-                    NumberOfDimensions = 1
-                    NumberOfElements = 1000
-                }
             }
             OutputSignals = {
                 Counter = {
@@ -175,14 +180,41 @@ $TestApp = {
                     DataSource = Streamer
                     Type = float32
                 }
+            }
+        }
+        +FastStreamerGAM = {
+            Class = IOGAM
+            InputSignals = {
+                Time = {
+                    DataSource = DDB2
+                    Type = uint32
+                }
                 Ch1 = {
-                    DataSource = Streamer
+                    DataSource = DDB2
                     Type = float32
                     NumberOfDimensions = 1
                     NumberOfElements = 1000
                 }
                 Ch2 = {
-                    DataSource = Streamer
+                    DataSource = DDB2
+                    Type = float32
+                    NumberOfDimensions = 1
+                    NumberOfElements = 1000
+                }
+            }
+            OutputSignals = {
+                Time = {
+                    DataSource = FastStreamer
+                    Type = uint32
+                }
+                Ch1 = {
+                    DataSource = FastStreamer
+                    Type = float32
+                    NumberOfDimensions = 1
+                    NumberOfElements = 1000
+                }
+                Ch2 = {
+                    DataSource = FastStreamer
                     Type = float32
                     NumberOfDimensions = 1
                     NumberOfElements = 1000
@@ -199,6 +231,9 @@ $TestApp = {
         +DDB = {
             Class = GAMDataSource
         }
+        +DDB2 = {
+            Class = GAMDataSource
+        }
 
         // ── Real-time clock / trigger source ─────────────────────────────────
         +Timer = {
@@ -213,6 +248,18 @@ $TestApp = {
                 }
             }
         }
+        +FastTimer = {
+            Class = LinuxTimer
+            SleepNature = "Default"
+            Signals = {
+                Counter = {
+                    Type = uint32
+                }
+                Time = {
+                    Type = uint32
+                }
+            }
+        }
 
         // ── UDP Streamer DataSource ──────────────────────────────────────────
         +Streamer = {
@@ -240,6 +287,17 @@ $TestApp = {
                     RangeMin = -5.0
                     RangeMax = 5.0
                 }
+            }
+        }
+        +FastStreamer = {
+            Class = UDPStreamer
+            Port = 44501
+            MaxPayloadSize = 1400
+            Signals = {
+                Time = {
+                    Type = uint32
+                    Unit = "us"
+                }
                 Ch1 = {
                     Type = float32
                     Unit = "V"
@@ -247,7 +305,7 @@ $TestApp = {
                     NumberOfElements = 1000
                     TimeMode = FirstSample
                     TimeSignal = Time
-                    SamplingRate = 1000000.0
+                    SamplingRate = 5000000.0
                 }
                 Ch2 = {
                     Type = float32
@@ -256,7 +314,7 @@ $TestApp = {
                     NumberOfElements = 1000
                     TimeMode = FirstSample
                     TimeSignal = Time
-                    SamplingRate = 1000000.0
+                    SamplingRate = 5000000.0
                 }
             }
         }
@@ -276,7 +334,12 @@ $TestApp = {
                 +Thread1 = {
                     Class = RealTimeThread
                     CPUs = 0x1
-                    Functions = {TimerGAM SineGAM1 SineGAM2 SineGAM3 SineGAM4 StreamerGAM}
+                    Functions = {TimerGAM SineGAM1 SineGAM2 StreamerGAM}
+                }
+                +Thread2 = {
+                    Class = RealTimeThread
+                    CPUs = 0x2
+                    Functions = {FastTimerGAM SineGAM3 SineGAM4 FastStreamerGAM}
                 }
             }
         }

Test/MARTeApp/run.sh

@@ -6,6 +6,10 @@
 #   ./run.sh --webui      # also start the WebUI Go client in the background
 #   ./run.sh --help       # show this message
 #
+# The WebUI is started with --source "TestApp@127.0.0.1:44500".
+# Additional sources and a persistent source list file (--sources-file) can
+# be configured directly in the WebUI or by editing this script.
+#
 # Prerequisites:
 #   - marte_env.sh must be present in the repo root (sets MARTe2_DIR, etc.)
 #   - MARTe2 and MARTe2-components must already be built
@@ -44,16 +48,24 @@ if [ -z "${MARTe2_DIR}" ]; then
     exit 1
 fi
 
-# ── Build UDPStreamer ─────────────────────────────────────────────────────────
+# ── Build components ─────────────────────────────────────────────────────────
 TARGET=x86-linux
 UDPSTREAMER_SRC="${REPO_ROOT}/Source/Components/DataSources/UDPStreamer"
 UDPSTREAMER_LIB="${REPO_ROOT}/Build/${TARGET}/Components/DataSources/UDPStreamer"
+SINEARRAYGAM_SRC="${REPO_ROOT}/Source/Components/GAMs/SineArrayGAM"
+SINEARRAYGAM_LIB="${REPO_ROOT}/Build/${TARGET}/Components/GAMs/SineArrayGAM"
 
 echo "==> Building UDPStreamer (TARGET=${TARGET})..."
 make -C "${UDPSTREAMER_SRC}" \
      -f Makefile.gcc \
      TARGET="${TARGET}" \
      2>&1 | tail -5
+
+echo "==> Building SineArrayGAM (TARGET=${TARGET})..."
+make -C "${SINEARRAYGAM_SRC}" \
+     -f Makefile.gcc \
+     TARGET="${TARGET}" \
+     2>&1 | tail -5
 echo "==> Build done."
 
 # ── Build WebUI binary (if requested and not already built) ──────────────────
@@ -61,7 +73,7 @@ WEBUI_DIR="${REPO_ROOT}/Client/WebUI"
 WEBUI_BIN="${WEBUI_DIR}/udpstreamer-webui"
 if [ "${START_WEBUI}" -eq 1 ] && [ ! -x "${WEBUI_BIN}" ]; then
     echo "==> Building WebUI..."
-    (cd "${WEBUI_DIR}" && go build -o udpstreamer-webui ./...)
+    (cd "${WEBUI_DIR}" &&  go build -o udpstreamer-webui ./...)
     echo "==> WebUI build done."
 fi
 
@@ -70,6 +82,7 @@ COMP="${MARTe2_Components_DIR}/Build/${TARGET}/Components"
 
 export LD_LIBRARY_PATH="\
 ${UDPSTREAMER_LIB}:\
+${SINEARRAYGAM_LIB}:\
 ${MARTe2_DIR}/Build/${TARGET}/Core:\
 ${COMP}/DataSources/LinuxTimer:\
 ${COMP}/DataSources/LoggerDataSource:\
@@ -92,20 +105,14 @@ for cls in WaveformSin WaveformChirp WaveformPointsDef; do
     fi
 done
 
-# SineArrayGAM is bundled inside UDPStreamer.so; create a symlink so MARTe2
-# can dlopen("SineArrayGAM.so") before UDPStreamer has been registered.
-SINE_LINK="${UDPSTREAMER_LIB}/SineArrayGAM.so"
-if [ ! -e "${SINE_LINK}" ]; then
-    ln -sf "${UDPSTREAMER_LIB}/UDPStreamer.so" "${SINE_LINK}"
-fi
 
 # ── Optionally start WebUI ────────────────────────────────────────────────────
 if [ "${START_WEBUI}" -eq 1 ]; then
-    echo "==> Starting WebUI on http://localhost:8080 (streamer at 127.0.0.1:44500)..."
+    echo "==> Starting WebUI on http://localhost:8080 (source: 127.0.0.1:44500)..."
     "${WEBUI_BIN}" \
-        --streamer 127.0.0.1:44500 \
+        --source "Streamer@127.0.0.1:44500" \
-        --listen :8080 \
+        --source "FastStreamer@127.0.0.1:44501" \
-        --clientport 44900 &
+        --listen :8080 &
     WEBUI_PID=$!
     echo "==> WebUI PID ${WEBUI_PID}"
 fi