359 lines
14 KiB
C++
359 lines
14 KiB
C++
/**
|
||
* @file StreamHub.h
|
||
* @brief Top-level orchestrator: N UDP sources, M WebSocket clients, push loop, trigger.
|
||
*
|
||
* StreamHub owns:
|
||
* - Up to kMaxSessions UDPSourceSessions (one per MARTe2 UDPStreamer source)
|
||
* - WSServer (multi-client WebSocket server)
|
||
* - TriggerEngine (continuous / armed / triggered FSM)
|
||
*
|
||
* The push loop runs at pushRateHz Hz and:
|
||
* 1. For each configured session, reads ring buffers, applies LTTB decimation,
|
||
* serializes a binary data frame, broadcasts to all WS clients.
|
||
* 2. At statsRateHz intervals, serializes a JSON stats frame and broadcasts.
|
||
*
|
||
* Commands are received as JSON text frames from any connected WS client via the
|
||
* WSCommandCallback interface.
|
||
*/
|
||
|
||
#ifndef STREAMHUB_H_
|
||
#define STREAMHUB_H_
|
||
|
||
#include "CompilerTypes.h"
|
||
#include "FastPollingMutexSem.h"
|
||
#include "StreamString.h"
|
||
#include "StructuredDataI.h"
|
||
#include "Threads.h"
|
||
|
||
#include "UDPSourceSession.h"
|
||
#include "WSServer.h"
|
||
#include "TriggerEngine.h"
|
||
#include "HistoryWriter.h"
|
||
|
||
namespace StreamHub {
|
||
|
||
using MARTe::uint8;
|
||
using MARTe::uint16;
|
||
using MARTe::uint32;
|
||
using MARTe::uint64;
|
||
using MARTe::float64;
|
||
using MARTe::StreamString;
|
||
using MARTe::FastPollingMutexSem;
|
||
using MARTe::StructuredDataI;
|
||
|
||
/** Maximum number of simultaneously connected UDPStreamer sources. */
|
||
static const uint32 kMaxSessions = 32u;
|
||
|
||
/** Maximum number of stored per-signal calibration entries. */
|
||
static const uint32 kMaxCalibration = 256u;
|
||
|
||
/** Maximum length of a calibration unit override (mirrors the Go maxUnitLen). */
|
||
static const uint32 kMaxUnitLen = 16u;
|
||
|
||
/**
|
||
* @brief One per-signal affine calibration: y = raw*scale + offset.
|
||
*
|
||
* Keyed by the source LABEL (not the runtime "sN" id, which is assigned in
|
||
* add-order and would rebind if the source list were reordered) and by the
|
||
* BASE signal name (no "[i]" suffix: one entry covers a whole array signal).
|
||
*
|
||
* Fixed-size char arrays are used deliberately: they avoid per-entry heap
|
||
* churn (no StreamString allocation per calibration slot), keep the type free
|
||
* of STL, and make a CalibrationEntry snapshot trivially copyable under the
|
||
* calibration mutex lock.
|
||
*/
|
||
struct CalibrationEntry {
|
||
char source[128]; ///< Source label
|
||
char signal[128]; ///< Base signal name (no "[i]" suffix)
|
||
char unit[17]; ///< Unit override (max kMaxUnitLen bytes + NUL)
|
||
MARTe::float64 scale;
|
||
MARTe::float64 offset;
|
||
};
|
||
|
||
/**
|
||
* @brief Top-level StreamHub orchestrator.
|
||
*
|
||
* Usage:
|
||
* StreamHub hub;
|
||
* hub.Initialise(cfg); // parse WSPort, MaxPoints, PushRate, Sources
|
||
* hub.Run(); // blocking push loop; returns on Stop()
|
||
*/
|
||
class StreamHub : public WSCommandCallback {
|
||
public:
|
||
|
||
StreamHub();
|
||
virtual ~StreamHub();
|
||
|
||
/**
|
||
* @brief Configure the hub from a MARTe2 StructuredDataI.
|
||
*
|
||
* Expected keys:
|
||
* WSPort (uint32, default 8090)
|
||
* MaxPoints (uint32, default 20000) — ring buffer capacity per signal
|
||
* PushRate (uint32, default 30) — push loop rate in Hz
|
||
* MaxPushPoints (uint32, default 50) — LTTB threshold for live push
|
||
* StatsRate (uint32, default 1) — stats broadcast rate in Hz
|
||
* +Sources { +<id> { Label=...; Addr=...; Port=... } }
|
||
*
|
||
* @return true on success.
|
||
*/
|
||
bool Initialise(StructuredDataI &cfg);
|
||
|
||
/**
|
||
* @brief Start the WebSocket server and run the push loop (blocking).
|
||
* Returns when Stop() is called (e.g. from SIGINT handler).
|
||
*/
|
||
bool Run();
|
||
|
||
/**
|
||
* @brief Signal the push loop to exit. Thread-safe.
|
||
*/
|
||
void Stop();
|
||
|
||
/* ---- WSCommandCallback (called from WS client read threads) ---------- */
|
||
virtual void OnWSCommand(const char *json, uint32 len, uint32 slotIdx);
|
||
virtual void OnWSClientConnected();
|
||
virtual void OnWSClientDisconnected();
|
||
|
||
private:
|
||
|
||
/* ---- Push loop helpers ----------------------------------------------- */
|
||
|
||
/** One push tick: for each session emit a binary frame; handle trigger. */
|
||
void PushData();
|
||
|
||
/** Serialize one binary data frame for session idx into pushBuf_. */
|
||
uint32 SerializeBinaryFrame(uint32 sessionIdx, uint8 *buf, uint32 bufSize);
|
||
|
||
/** Broadcast JSON stats for all sessions. */
|
||
void PushStats();
|
||
|
||
/** Broadcast {"type":"sources"} listing. */
|
||
void BroadcastSources();
|
||
|
||
/** Broadcast {"type":"config","sourceId":...} for one session. */
|
||
void BroadcastConfig(uint32 sessionIdx);
|
||
|
||
/** Broadcast {"type":"calibration","cal":[...]} to all clients. */
|
||
void BroadcastCalibration();
|
||
|
||
/** Broadcast {"type":"configSaved"|"configReloaded","ok":...} to all clients. */
|
||
void BroadcastConfigAck(const char *msgType, bool ok, const char *errText);
|
||
|
||
/* ---- Trigger (push loop side) ----------------------------------------- */
|
||
|
||
/**
|
||
* @brief Per-tick trigger servicing: finalize captures, auto-rearm,
|
||
* broadcast state transitions.
|
||
*/
|
||
void TriggerTick(float64 wallNowS);
|
||
|
||
/** Broadcast {"type":"triggerState",...} reflecting the current FSM. */
|
||
void BroadcastTriggerState();
|
||
|
||
/**
|
||
* @brief Size every ring so it retains the current trigger window.
|
||
* Called from the push loop once per stats tick; a no-op once the rings
|
||
* are large enough. Rates are measured from the rings themselves because
|
||
* most sources advertise samplingRate = 0.
|
||
*/
|
||
void GrowRingsForTrigger();
|
||
|
||
/** @return Largest ring capacity currently allocated across all sessions. */
|
||
uint32 CurrentMaxRingCapacity() const;
|
||
|
||
/**
|
||
* @brief How far source @p i has produced, in the trigger's time base;
|
||
* @p wallNowS when it publishes no producer clock (its samples are then
|
||
* stamped on arrival, so they share the hub's wall clock).
|
||
*/
|
||
float64 SourceFrontierTime(uint32 i, float64 wallNowS) const;
|
||
|
||
/* ---- Trigger capture assembly ---------------------------------------
|
||
* Sources are harvested one at a time, each as soon as *it* has produced
|
||
* past the end of the window, rather than all together once the slowest
|
||
* has. Sources free-run on their own clocks and can lag each other by
|
||
* seconds; making every source wait for the slowest lets the leaders' ring
|
||
* buffers roll past the pre-trigger region before it is ever read. */
|
||
|
||
/** @brief Start a version=2 capture frame:
|
||
* [u8 2][f64 trigTime][f64 preSec][f64 postSec][u32 nSig]. */
|
||
void BeginTriggerCapture(float64 trigTime, float64 preSec, float64 postSec);
|
||
|
||
/** @brief Append session @p i's signals to the pending frame, each as
|
||
* {[u16 keyLen][fullKey][u32 N][t f64×N][v f64×N]}. */
|
||
void HarvestTriggerCapture(uint32 i, float64 t0, float64 t1);
|
||
|
||
/** @brief Patch nSig, broadcast the pending frame and release it. */
|
||
void FinishTriggerCapture();
|
||
|
||
/* ---- Command handlers (called from OnWSCommand) ---------------------- */
|
||
|
||
void HandleAddSource(const char *json);
|
||
void HandleRemoveSource(const char *json);
|
||
void HandleSaveSources();
|
||
void HandleGetSources();
|
||
void HandleGetConfig(const char *json);
|
||
void HandleGetStats();
|
||
void HandleArm();
|
||
void HandleDisarm();
|
||
void HandleRearm();
|
||
void HandleTrigStop(const char *json);
|
||
void HandleSetTrigger(const char *json);
|
||
void HandleForceTrigger();
|
||
void HandleZoom(const char *json, uint32 slotIdx);
|
||
void HandleHistoryZoom(const char *json, uint32 slotIdx);
|
||
void HandleHistoryInfo(uint32 slotIdx);
|
||
void HandleSetMaxPoints(const char *json);
|
||
void HandlePing(uint32 slotIdx);
|
||
void HandleSetCalibration(const char *json);
|
||
void HandleReloadConfig();
|
||
|
||
/* ---- Binary recorder commands --------------------------------------- */
|
||
|
||
/** Arm recording on all enabled sessions (optional "sourceId" filter). */
|
||
void HandleRecStart(const char *json);
|
||
/** Disarm recording on all enabled sessions (optional "sourceId" filter). */
|
||
void HandleRecStop(const char *json);
|
||
/** Unicast a {"type":"recStatus",...} snapshot to one client. */
|
||
void HandleRecInfo(uint32 slotIdx);
|
||
/** Broadcast a {"type":"recStatus",...} snapshot to all clients. */
|
||
void BroadcastRecStatus();
|
||
|
||
/** Append the full recStatus message body into a growable buffer. */
|
||
void AppendRecStatusJSON(char *&buf, uint32 &off, uint32 &cap);
|
||
|
||
/* ---- Sources persistence (Go SourceConfig schema) --------------------- */
|
||
|
||
/**
|
||
* @brief Start a new dynamic session from "host:port" + options.
|
||
* Generates the session id ("s1", "s2", ...).
|
||
* @return true if the session was started.
|
||
*/
|
||
bool AddSourceInternal(const char *label, const char *addrPort,
|
||
const char *mcGroup, uint16 dataPort);
|
||
|
||
/**
|
||
* @brief Load sources and calibration from sourcesFile_ (a flat JSON array
|
||
* of {"label","addr","multicastGroup","dataPort"} source blocks and
|
||
* {"source","signal","scale","offset","unit"} calibration blocks).
|
||
* @param skipActive when true, a source whose "host:port" is already
|
||
* streaming is left alone instead of being started a second time.
|
||
* @param clearCalibration when true, the calibration table is cleared
|
||
* after a successful fread (never before), so a transient I/O failure
|
||
* does not silently wipe user calibration data.
|
||
* @return true if the file was read.
|
||
*/
|
||
bool LoadSourcesFile(bool skipActive, bool clearCalibration = false);
|
||
|
||
/** @return true if a session for this "host:port" is already active. */
|
||
bool SourceIsActive(const char *addrPort);
|
||
|
||
/**
|
||
* @brief Store or replace one calibration entry. An identity entry
|
||
* (scale 1, offset 0, empty unit) removes any stored one instead.
|
||
* @return true if the entry was valid (and therefore stored or removed).
|
||
*/
|
||
bool SetCalibrationEntry(const char *source, const char *signal,
|
||
MARTe::float64 scale, MARTe::float64 offset,
|
||
const char *unit);
|
||
|
||
/** Drop every calibration entry (used by reload, which replaces wholesale). */
|
||
void ClearCalibration();
|
||
|
||
/* ---- Tiny JSON helpers ----------------------------------------------- */
|
||
|
||
/**
|
||
* @brief Extract a string field from flat JSON: "key":"value".
|
||
* @return true if found; value written into out (up to outSize-1 chars).
|
||
*/
|
||
static bool JsonGetString(const char *json, const char *key,
|
||
char *out, uint32 outSize);
|
||
|
||
/**
|
||
* @brief Extract a numeric field from flat JSON: "key":number.
|
||
* @return true if found.
|
||
*/
|
||
static bool JsonGetFloat(const char *json, const char *key, float64 &out);
|
||
static bool JsonGetUint32(const char *json, const char *key, uint32 &out);
|
||
static bool JsonGetBool(const char *json, const char *key, bool &out);
|
||
|
||
/* ---- Members --------------------------------------------------------- */
|
||
|
||
UDPSourceSession sessions_[kMaxSessions];
|
||
uint32 numSessions_; ///< Count of active sessions (not an upper slot bound)
|
||
bool sessionActive_[kMaxSessions]; ///< Per-slot in-use flag (slots are not compacted on removal)
|
||
bool configBroadcast_[kMaxSessions]; ///< True once BroadcastConfig was sent for this session
|
||
FastPollingMutexSem sessionsMutex_; ///< Serializes add/remove (WS threads); push thread reads flags lock-free
|
||
|
||
WSServer wsServer_;
|
||
TriggerEngine trigger_;
|
||
HistoryWriter history_;
|
||
|
||
/* Recorder configuration (parsed once, applied to each session). */
|
||
RecorderConfig recorderCfg_;
|
||
|
||
/* Configuration */
|
||
uint16 wsPort_;
|
||
uint32 maxPoints_;
|
||
uint32 pushRateHz_;
|
||
uint32 maxPushPoints_;
|
||
uint32 statsRateHz_;
|
||
uint32 ringTemporal_; ///< Initial ring capacity for multi-element (waveform) signals
|
||
uint32 ringScalar_; ///< Ring capacity for scalar signals
|
||
uint32 ringMaxPts_; ///< Ceiling a ring may be grown to for a trigger window
|
||
volatile float64 trigRetentionSec_; ///< Retention the current trigger window needs
|
||
StreamString sourcesFile_; ///< Persistent dynamic source list (JSON)
|
||
uint32 nextSourceId_; ///< Counter for generated session ids ("sN")
|
||
|
||
CalibrationEntry *calibration_; ///< Heap-allocated array[kMaxCalibration]
|
||
uint32 numCalibration_;
|
||
FastPollingMutexSem calibrationMutex_; ///< Serializes calibration reads/writes
|
||
|
||
/* Push loop state */
|
||
volatile bool running_;
|
||
uint32 tickCount_; ///< incremented each push tick
|
||
|
||
/* Deferred setMaxPoints (applied by the push loop, not the WS thread) */
|
||
volatile bool pendingMaxPointsSet_;
|
||
uint32 pendingMaxPoints_;
|
||
|
||
/* Per-push scratch buffer (~8 MiB): one binary frame at a time */
|
||
static const uint32 kPushBufSize = 8u * 1024u * 1024u;
|
||
uint8 *pushBuf_;
|
||
|
||
/* Decimated output scratch (LTTB). Sized like the read scratch rather
|
||
* than maxPushPoints_: a PACKET-timed array raises its own threshold to
|
||
* one packet's worth of elements, which can exceed maxPushPoints_. */
|
||
float64 *lttbT_;
|
||
float64 *lttbV_;
|
||
|
||
/* Ring-buffer read scratch for the push loop (preallocated in Initialise) */
|
||
static const uint32 kPushScratchPts = 262144u; ///< Max new pts/signal/tick
|
||
float64 *pushT_;
|
||
float64 *pushV_;
|
||
|
||
/* Per-(session, signal) monotonic read cursors for ReadSignalSince.
|
||
* Only new samples since the last tick are pushed (fixes re-LTTB
|
||
* corruption of overlapping windows). */
|
||
uint64 pushCursor_[kMaxSessions][UDPSS_MAX_SIGNALS];
|
||
|
||
/* Trigger servicing state (push thread only) */
|
||
static const uint32 kTrigCapturePts = 20000u; ///< LTTB cap per captured signal
|
||
TrigState lastTrigState_; ///< Last broadcast FSM state
|
||
bool rearmPending_; ///< Normal-mode auto-rearm scheduled
|
||
float64 rearmAtWallS_; ///< Wall time of the scheduled auto-rearm
|
||
float64 collectStartWallS_; ///< Wall time COLLECTING began (watchdog only)
|
||
|
||
/* Capture frame under assembly across ticks (push thread only) */
|
||
MARTe::uint8 *capBuf_; ///< Pending frame, NULL when idle
|
||
uint32 capCap_; ///< Allocated size of capBuf_
|
||
uint32 capOff_; ///< Bytes written so far
|
||
uint32 capNSig_; ///< Signals appended so far
|
||
bool capHarvested_[kMaxSessions]; ///< Session already appended
|
||
};
|
||
|
||
} /* namespace StreamHub */
|
||
|
||
#endif /* STREAMHUB_H_ */
|