diff --git a/Source/Applications/StreamHub/Makefile.inc b/Source/Applications/StreamHub/Makefile.inc index 3ff8c56..848075d 100644 --- a/Source/Applications/StreamHub/Makefile.inc +++ b/Source/Applications/StreamHub/Makefile.inc @@ -1,4 +1,4 @@ -OBJSX = StreamHub.x UDPSourceSession.x WSServer.x TriggerEngine.x HistoryWriter.x +OBJSX = StreamHub.x UDPSourceSession.x WSServer.x TriggerEngine.x HistoryWriter.x BinaryRecorder.x PACKAGE = ROOT_DIR = ../../.. diff --git a/Source/Applications/StreamHub/UDPSourceSession.cpp b/Source/Applications/StreamHub/UDPSourceSession.cpp index 754597d..5d09e8a 100644 --- a/Source/Applications/StreamHub/UDPSourceSession.cpp +++ b/Source/Applications/StreamHub/UDPSourceSession.cpp @@ -71,8 +71,15 @@ UDPSourceSession::UDPSourceSession() trigEngine_(static_cast(0)), trigEpochSeen_(0xFFFFFFFFu), trigSigIdx_(-1), - trigElemIdx_(-1) { + trigElemIdx_(-1), + recPendingEpoch_(0u), + recSeenEpoch_(0u) { stateStr_ = "disconnected"; + memset(&recCfg_, 0, sizeof(recCfg_)); + recPendingSpec_[0] = '\0'; + strncpy(recActiveSpec_, "all", sizeof(recActiveSpec_) - 1u); + recActiveSpec_[sizeof(recActiveSpec_) - 1u] = '\0'; + recSpecMutex_.Create(); ResetCalibration(); } @@ -244,6 +251,19 @@ void UDPSourceSession::ParseConfigPayload(const uint8 *payload, uint32 size) { /* Allocate ring buffers (outside metaMutex to avoid long hold) */ AllocateRingBuffers(); + /* (Re)configure the recorder: a CONFIG change re-headers a fresh file and + * adopts any pending subset spec. */ + if (recCfg_.enabled) { + bool mask[UDPSS_MAX_SIGNALS]; + (void) recSpecMutex_.FastLock(); + strncpy(recActiveSpec_, recPendingSpec_, sizeof(recActiveSpec_) - 1u); + recActiveSpec_[sizeof(recActiveSpec_) - 1u] = '\0'; + recSeenEpoch_ = recPendingEpoch_; + recSpecMutex_.FastUnLock(); + ComputeIncludeMask(sigDescs_, numSigs, recActiveSpec_, mask); + recorder_.Configure(sigDescs_, numSigs, mask); + } + REPORT_ERROR_STATIC(MARTe::ErrorManagement::Information, "UDPSourceSession[%s]: CONFIG received — %u signals.", id_.Buffer(), numSigs); @@ -506,6 +526,21 @@ void UDPSourceSession::ParseDataPayload(const uint8 *payload, uint32 size) { } } + /* Binary recorder: adopt any pending subset spec, then capture the packet + * verbatim (native types, lossless) for disk recording. */ + if (recCfg_.enabled) { + if (recPendingEpoch_ != recSeenEpoch_) { + bool mask[UDPSS_MAX_SIGNALS]; + (void) recSpecMutex_.FastLock(); + strncpy(recActiveSpec_, recPendingSpec_, sizeof(recActiveSpec_) - 1u); + recActiveSpec_[sizeof(recActiveSpec_) - 1u] = '\0'; + recSeenEpoch_ = recPendingEpoch_; + recSpecMutex_.FastUnLock(); + ComputeIncludeMask(descs, nSigs, recActiveSpec_, mask); + recorder_.Configure(descs, nSigs, mask); + } + recorder_.CapturePacket(payload, sigOff, sigElems, pm, numSamples); + } } float64 UDPSourceSession::DecodeOneElem(const uint8 *payload, uint32 off, @@ -808,6 +843,79 @@ void UDPSourceSession::SetTriggerEngine(TriggerEngine *engine) { trigElemIdx_ = -1; } +/*---------------------------------------------------------------------------*/ +/* Binary recorder control */ +/*---------------------------------------------------------------------------*/ + +void UDPSourceSession::SetRecorderConfig(const RecorderConfig &cfg) { + recCfg_ = cfg; + if (!cfg.enabled) { return; } + recorder_.Init(cfg, id_.Buffer()); + (void) recSpecMutex_.FastLock(); + strncpy(recPendingSpec_, cfg.signals, sizeof(recPendingSpec_) - 1u); + recPendingSpec_[sizeof(recPendingSpec_) - 1u] = '\0'; + strncpy(recActiveSpec_, cfg.signals, sizeof(recActiveSpec_) - 1u); + recActiveSpec_[sizeof(recActiveSpec_) - 1u] = '\0'; + recPendingEpoch_++; + recSpecMutex_.FastUnLock(); +} + +void UDPSourceSession::SetRecorderSignals(const char *spec) { + if (spec == static_cast(0)) { return; } + (void) recSpecMutex_.FastLock(); + strncpy(recPendingSpec_, spec, sizeof(recPendingSpec_) - 1u); + recPendingSpec_[sizeof(recPendingSpec_) - 1u] = '\0'; + recPendingEpoch_++; + recSpecMutex_.FastUnLock(); +} + +void UDPSourceSession::RequestRecArm() { + if (recCfg_.enabled) { recorder_.RequestArm(); } +} + +void UDPSourceSession::RequestRecDisarm() { + if (recCfg_.enabled) { recorder_.RequestDisarm(); } +} + +void UDPSourceSession::RecorderFlushTick(uint32 nowSec) { + if (recCfg_.enabled) { recorder_.FlushTick(nowSec); } +} + +void UDPSourceSession::GetRecorderInfo(bool &recording, char *file, uint32 fileSz, + uint64 &bytesWritten, uint64 &rowsWritten, + uint64 &droppedRows, uint64 &freeMB) const { + recorder_.GetInfo(recording, file, fileSz, bytesWritten, rowsWritten, + droppedRows, freeMB); +} + +void UDPSourceSession::ComputeIncludeMask(const UDPSSignalDescriptor *descs, + uint32 n, const char *spec, + bool *maskOut) { + if ((spec == static_cast(0)) || (spec[0] == '\0') || + (strcmp(spec, "all") == 0)) { + for (uint32 i = 0u; i < n; i++) { maskOut[i] = true; } + return; + } + for (uint32 i = 0u; i < n; i++) { + char key[160]; + (void) snprintf(key, sizeof(key), "%s:%s", id_.Buffer(), descs[i].name); + const size_t klen = strlen(key); + maskOut[i] = false; + const char *p = spec; + while (*p != '\0') { + while ((*p == ',') || (*p == ' ')) { p++; } + const char *start = p; + while ((*p != '\0') && (*p != ',')) { p++; } + size_t tlen = static_cast(p - start); + while ((tlen > 0u) && (start[tlen - 1u] == ' ')) { tlen--; } + if ((tlen == klen) && (strncmp(start, key, klen) == 0)) { + maskOut[i] = true; + break; + } + } + } +} + void UDPSourceSession::ResolveTriggerSignal(const UDPSSignalDescriptor *descs, uint32 nSigs) { trigSigIdx_ = -1; diff --git a/Source/Applications/StreamHub/UDPSourceSession.h b/Source/Applications/StreamHub/UDPSourceSession.h index c1e10fe..ae7f2b0 100644 --- a/Source/Applications/StreamHub/UDPSourceSession.h +++ b/Source/Applications/StreamHub/UDPSourceSession.h @@ -19,6 +19,7 @@ #include "SignalRingBuffer.h" #include "UDPSourceStats.h" #include "TriggerEngine.h" +#include "BinaryRecorder.h" #include namespace StreamHub { @@ -161,8 +162,40 @@ public: */ void SetTriggerEngine(TriggerEngine *engine); + /* ---- Binary recorder control -------------------------------------- */ + + /** @brief Set the recorder configuration; initialises the recorder when + * enabled. Must be called before Start(). */ + void SetRecorderConfig(const RecorderConfig &cfg); + + /** @brief Override the recorded signal subset for this run (any thread). + * Adopted on the receive thread via an epoch check. */ + void SetRecorderSignals(const char *spec); + + /** @brief Arm the recorder (any thread). */ + void RequestRecArm(); + + /** @brief Disarm the recorder (any thread). */ + void RequestRecDisarm(); + + /** @brief Drive recorder disk I/O (push thread). */ + void RecorderFlushTick(uint32 nowSec); + + /** @brief Recorder status snapshot (push thread). */ + void GetRecorderInfo(bool &recording, char *file, uint32 fileSz, + uint64 &bytesWritten, uint64 &rowsWritten, + uint64 &droppedRows, uint64 &freeMB) const; + + /** @return true if the recorder is enabled for this session. */ + bool IsRecorderEnabled() const { return recCfg_.enabled; } + private: + /** @brief Build the per-signal include mask from a subset spec + * ("all" or comma-separated "src:sig" keys). Receive thread only. */ + void ComputeIncludeMask(const UDPSSignalDescriptor *descs, uint32 n, + const char *spec, bool *maskOut); + /* DATA payload parsing */ void ParseConfigPayload(const uint8 *payload, uint32 size); void ParseDataPayload(const uint8 *payload, uint32 size); @@ -293,6 +326,15 @@ private: uint32 trigEpochSeen_; ///< Last resolved trigger config epoch MARTe::int32 trigSigIdx_; ///< Watched signal index (-1 = none) MARTe::int32 trigElemIdx_; ///< Watched element index (-1 = all) + + /* Binary recorder (capture on receive thread, I/O on push thread). */ + BinaryRecorder recorder_; + RecorderConfig recCfg_; ///< Recorder config (enabled flag gates use) + mutable FastPollingMutexSem recSpecMutex_; + char recPendingSpec_[1024]; ///< Subset spec awaiting adoption + char recActiveSpec_[1024]; ///< Currently applied subset spec + volatile uint32 recPendingEpoch_; ///< Bumped when the spec changes + uint32 recSeenEpoch_; ///< Last spec epoch adopted (receive thread) }; } /* namespace StreamHub */