martino/MARTe-Integrated-Components
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(streamhub): parse +Recorder config and add recorder WS commands

Browse Source 
Parse the optional +Recorder block in Initialise (Enabled/AutoStart/
Directory/MaxFileMB/KeepFiles/StagingMB/FlushIntervalSec/MinDiskFreeMB/
Signals, MB->bytes), apply it to each session before Start (static and
dynamic sources). Drive recorder disk I/O from the push loop via
RecorderFlushTick, and add recStart/recStop/recInfo WS commands plus a
recStatus broadcast/snapshot.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Martino Ferrari
2026-06-25 01:13:23 +02:00
parent 93fc11c88f
commit 19e649aa38
2 changed files with 165 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/Applications/StreamHub/StreamHub.cpp
+148 -1
View File
@@ -74,6 +74,7 @@ StreamHub::StreamHub()
lastTrigState_(kTrigIdle),
rearmPending_(false),
rearmAtWallS_(0.0) {
memset(&recorderCfg_, 0, sizeof(recorderCfg_));
for (uint32 i = 0u; i < kMaxSessions; i++) {
sessionActive_[i] = false;
configBroadcast_[i] = false;
@@ -137,6 +138,48 @@ bool StreamHub::Initialise(StructuredDataI &cfg) {
cfg.MoveToAncestor(1u);
}
/* Parse +Recorder block (optional). Defaults: disabled, auto-start when
* enabled, 256 MiB files, keep 8, 8 MiB staging, 5 s fsync cadence,
* 500 MiB free-disk guard, all signals. MB sizes are converted to bytes. */
recorderCfg_.enabled = false;
recorderCfg_.autoStart = true;
strncpy(recorderCfg_.directory, "streamhub_rec",
sizeof(recorderCfg_.directory) - 1u);
recorderCfg_.maxFileBytes = 256u * 1024u * 1024u;
recorderCfg_.keepFiles = 8u;
recorderCfg_.stagingBytes = 8u * 1024u * 1024u;
recorderCfg_.flushIntervalSec = 5u;
recorderCfg_.minDiskFreeMB = 500u;
strncpy(recorderCfg_.signals, "all", sizeof(recorderCfg_.signals) - 1u);
if (cfg.MoveRelative("+Recorder") || cfg.MoveRelative("Recorder")) {
uint32 rtmp = 0u;
if (cfg.Read("Enabled", rtmp)) { recorderCfg_.enabled = (rtmp != 0u); }
if (cfg.Read("AutoStart", rtmp)) { recorderCfg_.autoStart = (rtmp != 0u); }
StreamString rdir;
if (cfg.Read("Directory", rdir)) {
strncpy(recorderCfg_.directory, rdir.Buffer(),
sizeof(recorderCfg_.directory) - 1u);
}
if (cfg.Read("MaxFileMB", rtmp) && rtmp > 0u) {
recorderCfg_.maxFileBytes = rtmp * 1024u * 1024u;
}
if (cfg.Read("KeepFiles", rtmp)) { recorderCfg_.keepFiles = rtmp; }
if (cfg.Read("StagingMB", rtmp) && rtmp > 0u) {
recorderCfg_.stagingBytes = rtmp * 1024u * 1024u;
}
if (cfg.Read("FlushIntervalSec", rtmp) && rtmp > 0u) {
recorderCfg_.flushIntervalSec = rtmp;
}
if (cfg.Read("MinDiskFreeMB", rtmp)) { recorderCfg_.minDiskFreeMB = rtmp; }
StreamString rsigs;
if (cfg.Read("Signals", rsigs)) {
strncpy(recorderCfg_.signals, rsigs.Buffer(),
sizeof(recorderCfg_.signals) - 1u);
}
cfg.MoveToAncestor(1u);
}
/* Allocate scratch buffers */
pushBuf_ = new uint8[kPushBufSize];
lttbT_ = new float64[maxPushPoints_];
@@ -186,6 +229,7 @@ bool StreamHub::Initialise(StructuredDataI &cfg) {
mcGroup, static_cast<uint16>(dataPort32));
if (ok) {
sessions_[numSessions_].SetRecorderConfig(recorderCfg_);
ok = sessions_[numSessions_].Start();
}
if (ok) {
@@ -319,6 +363,14 @@ void StreamHub::PushData() {
wsServer_.BroadcastText(resp, static_cast<uint32>(strlen(resp)));
}
/* Wall-clock seconds for recorder fsync cadence and file timestamps. */
uint32 nowSec = 0u;
{
struct timespec tsRec;
(void) clock_gettime(CLOCK_REALTIME, &tsRec);
nowSec = static_cast<uint32>(tsRec.tv_sec);
}
for (uint32 i = 0u; i < kMaxSessions; i++) {
if (!sessionActive_[i]) { continue; }
if (!sessions_[i].IsConfigured()) { continue; }
@@ -355,6 +407,11 @@ void StreamHub::PushData() {
if (history_.IsEnabled()) {
history_.WriteTick(i, sessions_[i]);
}
/* Binary recorder: all disk I/O happens here on the push thread. */
if (sessions_[i].IsRecorderEnabled()) {
sessions_[i].RecorderFlushTick(nowSec);
}
}
}
@@ -697,6 +754,9 @@ void StreamHub::OnWSCommand(const char *json, uint32 /*len*/, uint32 slotIdx) {
else if (strcmp(type, "historyZoom") == 0) { HandleHistoryZoom(json, slotIdx); }
else if (strcmp(type, "historyInfo") == 0) { HandleHistoryInfo(slotIdx); }
else if (strcmp(type, "setMaxPoints") == 0) { HandleSetMaxPoints(json); }
else if (strcmp(type, "recStart") == 0) { HandleRecStart(json); }
else if (strcmp(type, "recStop") == 0) { HandleRecStop(json); }
else if (strcmp(type, "recInfo") == 0) { HandleRecInfo(slotIdx); }
else if (strcmp(type, "ping") == 0) { HandlePing(slotIdx); }
}
@@ -772,7 +832,10 @@ bool StreamHub::AddSourceInternal(const char *label, const char *addrPort,
bool ok = sessions_[idx].Initialise(id, lbl, host,
static_cast<uint16>(port), maxPoints_,
mcGroup, dataPort);
if (ok) { ok = sessions_[idx].Start(); }
if (ok) {
sessions_[idx].SetRecorderConfig(recorderCfg_);
ok = sessions_[idx].Start();
}
if (ok) {
sessionActive_[idx] = true;
@@ -1415,6 +1478,90 @@ void StreamHub::HandlePing(uint32 slotIdx) {
wsServer_.SendText(slotIdx, msg, static_cast<uint32>(strlen(msg)));
}
/*---------------------------------------------------------------------------*/
/* Binary recorder commands */
/*---------------------------------------------------------------------------*/
void StreamHub::HandleRecStart(const char *json) {
char wantId[64] = "";
const bool filtered = JsonGetString(json, "sourceId", wantId, sizeof(wantId));
for (uint32 i = 0u; i < kMaxSessions; i++) {
if (!sessionActive_[i]) { continue; }
if (!sessions_[i].IsRecorderEnabled()) { continue; }
if (filtered) {
StreamString sid = sessions_[i].GetId();
if (strcmp(sid.Buffer(), wantId) != 0) { continue; }
}
sessions_[i].RequestRecArm();
}
BroadcastRecStatus();
}
void StreamHub::HandleRecStop(const char *json) {
char wantId[64] = "";
const bool filtered = JsonGetString(json, "sourceId", wantId, sizeof(wantId));
for (uint32 i = 0u; i < kMaxSessions; i++) {
if (!sessionActive_[i]) { continue; }
if (!sessions_[i].IsRecorderEnabled()) { continue; }
if (filtered) {
StreamString sid = sessions_[i].GetId();
if (strcmp(sid.Buffer(), wantId) != 0) { continue; }
}
sessions_[i].RequestRecDisarm();
}
BroadcastRecStatus();
}
void StreamHub::AppendRecStatusJSON(char *&buf, uint32 &off, uint32 &cap) {
JsonAppendf(buf, off, cap, "{\"type\":\"recStatus\",\"sources\":{");
bool first = true;
for (uint32 i = 0u; i < kMaxSessions; i++) {
if (!sessionActive_[i]) { continue; }
if (!sessions_[i].IsRecorderEnabled()) { continue; }
bool recording = false;
char file[768] = "";
uint64 bytesWritten = 0u;
uint64 rowsWritten = 0u;
uint64 droppedRows = 0u;
uint64 freeMB = 0u;
sessions_[i].GetRecorderInfo(recording, file, sizeof(file),
bytesWritten, rowsWritten,
droppedRows, freeMB);
StreamString sid = sessions_[i].GetId();
JsonAppendf(buf, off, cap, "%s\"%s\":{\"recording\":%s,\"file\":\"%s\","
"\"bytesWritten\":%llu,\"rowsWritten\":%llu,"
"\"droppedRows\":%llu,\"freeMB\":%llu}",
first ? "" : ",", sid.Buffer(),
recording ? "true" : "false", file,
static_cast<unsigned long long>(bytesWritten),
static_cast<unsigned long long>(rowsWritten),
static_cast<unsigned long long>(droppedRows),
static_cast<unsigned long long>(freeMB));
first = false;
}
JsonAppendf(buf, off, cap, "}}");
}
void StreamHub::HandleRecInfo(uint32 slotIdx) {
uint32 cap = 4096u;
char *buf = new char[cap];
uint32 off = 0u;
AppendRecStatusJSON(buf, off, cap);
wsServer_.SendText(slotIdx, buf, off);
delete[] buf;
}
void StreamHub::BroadcastRecStatus() {
uint32 cap = 4096u;
char *buf = new char[cap];
uint32 off = 0u;
AppendRecStatusJSON(buf, off, cap);
wsServer_.BroadcastText(buf, off);
delete[] buf;
}
/*---------------------------------------------------------------------------*/
/* Tiny JSON helpers */
/*---------------------------------------------------------------------------*/
Source/Applications/StreamHub/StreamHub.h
+17
View File
@@ -146,6 +146,20 @@ private:
void HandleSetMaxPoints(const char *json);
void HandlePing(uint32 slotIdx);
/* ---- 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) --------------------- */
/**
@@ -191,6 +205,9 @@ private:
TriggerEngine trigger_;
HistoryWriter history_;
/* Recorder configuration (parsed once, applied to each session). */
RecorderConfig recorderCfg_;
/* Configuration */
uint16 wsPort_;
uint32 maxPoints_;