MARTe-Integrated-Components/Docs/StreamHub-Developer.md

# StreamHub — Developer Guide

`Source/Applications/StreamHub/` is a headless C++ application (MARTe2-linked,
MARTe2 coding style, no STL) that aggregates one or more UDPS sources
(`UDPStreamer` DataSources) and serves them to oscilloscope clients over
WebSocket. The Go hub (`Client/udpstreamer`) is the feature-complete reference
implementation of the same protocol and is kept untouched.

Wire protocols: [Protocol.md](Protocol.md) (UDPS, source → hub) and
[StreamHub-API.md](StreamHub-API.md) (WebSocket, hub → clients).

---

## 1. Source layout

| File | Role |
|------|------|
| `main.cpp` | CLI entry (`-cfg file.cfg -port N -maxPoints N`), signal handling |
| `StreamHub.{h,cpp}` | Top-level object: config, push loop, WS command dispatch, broadcasts |
| `UDPSourceSession.{h,cpp}` | One UDPS source: `UDPSClient` listener, payload decode, wall-clock calibration, ring buffers, stats |
| `SignalRingBuffer.h` | Per-signal (t,v) ring: monotonic write counter, `ReadSince` cursor reads, binary-search `ReadRange` |
| `TriggerEngine.{h,cpp}` | Trigger FSM (IDLE/ARMED/COLLECTING/TRIGGERED), edge detection per decoded sample |
| `UDPSourceStats.h` | 512-entry cycle/frag/byte rings → avg/std/min/max, rate, 20-bin histogram |
| `HistoryWriter.{h,cpp}` | Disk-backed circular history: per-signal `.shist` files, `WriteTick`, `ReadRange` (binary search via `pread`), disk space monitoring |
| `WSServer.{h,cpp}` | RFC 6455 server: handshake, framing, per-client write mutex, broadcast/unicast |
| `WSFrame.h`, `SHA1.h`, `Base64.h` | Header-only WS plumbing, shared with the ImGui client |
| `LTTB.h` | Largest-Triangle-Three-Buckets decimation |

The UDPS client itself lives in the shared library
`Source/Components/Interfaces/UDPStream/` (`UDPSClient`, also used by
`DebugService`). Note: in multicast mode the server delivers CONFIG over the
**TCP control connection**, so `UDPSClient` selects on both the multicast UDP
socket and the TCP socket and frames TCP reads.

## 2. Thread model

| Thread | Created by | Work |
|--------|-----------|------|
| main / push loop | `StreamHub::Run()` | At `PushRate` Hz: `PushData()` (serialise v1 frames), trigger servicing (capture finalisation, auto-rearm), `PushStats()` at `StatsRate` Hz |
| WS accept | `WSServer::Start()` | `accept()` + handshake, spawns client readers |
| WS client reader ×16 | `WSServer` | Reads frames, may contain several coalesced frames per TCP read; dispatches JSON to `StreamHub::OnWSCommand(json, len, slotIdx)` |
| UDPS receive ×32 | `UDPSClient::Start()` (one per session) | select() on UDP (+TCP in multicast), reassembles fragments, calls `UDPSourceSession` listener callbacks |

Synchronisation:
- Each `SignalRingBuffer` has its own `FastPollingMutexSem`; writers are the
  UDPS receive threads, readers are the push loop and zoom handlers.
- `WSServer` has a per-client write mutex (push loop and command replies can
  write concurrently).
- WS commands run on reader threads, but mutating operations (ring resize via
  `setMaxPoints`) are deferred to the push loop through pending atomics.
- Session slots use an `active` flag; removal never compacts the array, so
  indices stay stable.

## 3. Time base (wall-clock calibration)

All timestamps exposed to clients are **Unix wall-clock seconds** (float64).
Each session calibrates per time-source:

- First DATA packet anchors `pktCalibOffset = wallNow − hrt/hrtFreq`; thereafter
  `packetT = pktCalibOffset + hrt/hrtFreq`.
- Each referenced time signal gets its own offset on first value;
  `timerToSec = 1e-9` for `uint64` time signals, `1e-6` otherwise.
- Re-anchoring on reconnect, CONFIG change, or if computed time drifts > 2 s
  from wall clock (source restart / remote-vs-local HRT frequency drift).

Per `timeMode`:

| timeMode | Timestamping |
|----------|-------------|
| FIRST/LAST_SAMPLE | anchor = calibrated time-signal value (fallback `packetT`); `t = anchor ± k·dt` from `samplingRate` |
| FULL_ARRAY | `t[k]` = calibrated time array element |
| PACKET, n=1 | `packetT` |
| PACKET, n>1 | elements span `(lastPktWall, wallNow]` — backward anchoring, deliberately different from the Go hub (which extrapolates forward and overlaps the next packet under jitter); keeps ring time strictly monotonic |

Multi-element PACKET signals are exposed per element as `name[i]`.

## 4. Push path (only-new samples)

`SignalRingBuffer` keeps a monotonic `totalWritten`; the hub keeps a
per-(session, signal) cursor and uses `ReadSince(cursor, …)` (clamped to the
oldest sample on overrun). Each tick serialises only new samples, LTTB-capped
to `MaxPushPoints` (default 50) per signal — LTTB is applied only to temporal
signals (multi-element, timeMode ≠ PACKET). Cursors advance even with zero WS
clients so a connecting client never receives a backlog burst. Cursors are
reset on (re)CONFIG and on ring resize.

This replaces the original "re-send the last N points each tick" design, which
caused visible trace corruption (LTTB picked different points per overlapping
window).

## 5. Trigger engine

Hub-side, web-client semantics (`setTrigger` fields in
[StreamHub-API.md](StreamHub-API.md)):

```
IDLE --arm--> ARMED --edge crossing--> COLLECTING --wallNow ≥ trigTime+postSec+0.15s--> TRIGGERED
TRIGGERED --rearm (single) / auto ~200ms (normal, unless stopped)--> ARMED
any --disarm--> IDLE
```

`UDPSourceSession` calls `TriggerEngine::CheckSample` for every decoded sample
of the configured signal (signal index cached per config epoch). On
finalisation the push loop reads `[trigTime−preSec, trigTime+postSec]` from all
rings, LTTB-caps to 20 000 pts/signal and broadcasts a binary **version 2**
capture frame; every FSM transition broadcasts a `triggerState` event.

## 6. Configuration

```
WSPort        = 8090
MaxPoints     = 20000      // legacy global cap (overridable with -maxPoints)
PushRate      = 30         // Hz
MaxPushPoints = 50         // per signal per push
StatsRate     = 1          // Hz
RingTemporal  = 1000000    // ring capacity, temporal signals (pts)
RingScalar    = 100000     // ring capacity, scalar/PACKET signals (pts)
SourcesFile   = "streamhub_sources.json"   // saveSources persistence
Sources = {
    Src1 = { Label = "PSU"  Addr = "127.0.0.1" Port = 44500
             MulticastGroup = "239.0.0.1" DataPort = 44503 }  // multicast optional
}
```

Sources from `SourcesFile` are loaded after the static `Sources` block;
sources added at runtime via WS `addSource` get ids `s1, s2, …`.

### History configuration

An optional `+History` block enables disk-backed circular storage
(`HistoryWriter`). The `+` prefix is MARTe2 `StandardParser` syntax for a
child node; the hub looks for both `+History` and `History` as the node name.

```
+History = {
    Directory      = "/data/streamhub_history"   // required
    DurationHours  = 1          // hours of data to retain per signal (default 1)
    Decimation     = 10         // keep every Nth sample (default 1)
    FlushIntervalSec = 5        // header flush period in seconds (default 5)
    MinDiskFreeMB  = 500        // pause writes below this threshold (default 500)
}
```

Per-signal file capacity is computed at source CONFIG time:
`capacity = ceil(DurationHours × 3600 × samplingRate / Decimation)`, minimum
1000 pairs.

### `.shist` binary file format

Each signal gets one file: `<Directory>/<sourceId>/<signalName>.shist`.
The file size is fixed at creation (`64 + capacity × 16` bytes) and never grows.

```
Offset  Size  Field
0       4     Magic: "SHR1"
4       4     uint32 version (1)
8       4     uint32 capacity (max pairs)
12      4     uint32 head (next write position, 0-based, wraps at capacity)
16      4     uint32 count (valid entries, ≤ capacity)
20      4     uint32 decimation
24      8     float64 tOldest (Unix seconds)
32      8     float64 tNewest (Unix seconds)
40      24    reserved (zero-padded to 64 bytes)
64      …     data: capacity × 16 bytes (float64 time + float64 value per pair)
```

Data is written at the `head` position and wraps circularly. The oldest valid
entry is at logical index `(head + capacity − count) % capacity`. All I/O
uses `pwrite`/`pread` (no `mmap`), so concurrent reads from WS threads are
safe without locking.

Headers are flushed to disk every `FlushIntervalSec` seconds and on shutdown.
On restart, if an existing file has matching magic, version and capacity, it is
reopened — head/count/time bounds are restored from the on-disk header.

### History query path

`historyZoom` requests (see [StreamHub-API.md](StreamHub-API.md)) call
`HistoryWriter::ReadRange` which performs binary search over the circular file
using `pread` to locate the `[t0, t1]` window, then copies matching pairs.
If the result exceeds the requested `n`, LTTB decimation is applied (same
`LTTBDecimate` as in-memory zoom).

Both the web SPA and ImGui client issue `historyZoom` in parallel with regular
`zoom` and merge the results: history covers the older part of the visible
window, the in-memory ring covers the recent part.

## 7. Build & test

```bash
source env.sh                                    # always, for build and run

make -f Makefile.gcc apps                        # builds StreamHub.ex
# or: make -C Source/Applications/StreamHub -f Makefile.gcc
./Build/x86-linux/StreamHub/StreamHub.ex -cfg hub.cfg

make -f Makefile.gcc test
./Build/x86-linux/GTest/MainGTest.ex             # 71 unit tests, incl.
   # SignalRingBuffer (ReadSince / binary-search ReadRange / wrap),
   # TriggerEngine FSM, LTTB — sources in Test/Applications/StreamHub/

cd Test/E2E/suite && ./run_e2e.sh                # full-stack E2E (see below)
./run_streamhub.sh -w -g                         # interactive demo stack
```

### End-to-end test

`Test/E2E/suite/run_e2e.sh` is the unified E2E suite covering the whole
streaming + debug chain (`chain`/`direct`/`recorder`/`debug`/`tcplogger`
scenario kinds, see `Test/E2E/suite/scenarios.py`), including StreamHub live
push, zoom, window and trigger checks via the Go `chain-client`. It builds
everything, runs the scenario matrix plus the stress matrix, and produces a
consolidated `report_data.json` + Typst PDF report
(`Test/E2E/suite/E2E_Report.typ`). See the script's `--help` for options.

When changing the WS protocol, update **in lockstep**: this hub, the Go hub
(`Common/Client/go/wshub`), the browser SPA (`Client/udpstreamer/static`), the
ImGui client (`Client/streamhub/Protocol.cpp`), the E2E `chain-client`
(`Test/E2E/suite/client`), and [StreamHub-API.md](StreamHub-API.md).

## 8. Gotchas

- **No STL** in this directory (MARTe2 style); use `StreamString`,
  `FastPollingMutexSem`, fixed arrays.
- Link against UDPStream **dynamically** (`LIBRARIES += -lUDPStream`), never
  `LIBRARIES_STATIC` — the GTest binary links every `.a` it finds and would get
  duplicate symbols.
- WS text frames may arrive coalesced; never NUL-terminate a payload in place
  without restoring the byte (it is the first header byte of the next frame).
- Zoom replies must print `t` with `%.17g`: at Unix-epoch magnitudes `%.6f`
  destroys µs resolution.