diff --git a/Source/Applications/StreamHub/StreamHub.cpp b/Source/Applications/StreamHub/StreamHub.cpp index 5ebf96e..3f7ca4a 100644 --- a/Source/Applications/StreamHub/StreamHub.cpp +++ b/Source/Applications/StreamHub/StreamHub.cpp @@ -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(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(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(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(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(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(bytesWritten), + static_cast(rowsWritten), + static_cast(droppedRows), + static_cast(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 */ /*---------------------------------------------------------------------------*/ diff --git a/Source/Applications/StreamHub/StreamHub.h b/Source/Applications/StreamHub/StreamHub.h index 7087f9c..340a603 100644 --- a/Source/Applications/StreamHub/StreamHub.h +++ b/Source/Applications/StreamHub/StreamHub.h @@ -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_;