diff --git a/Source/Components/Interfaces/DebugService/DebugService.cpp b/Source/Components/Interfaces/DebugService/DebugService.cpp index e59783f..2d1cb55 100644 --- a/Source/Components/Interfaces/DebugService/DebugService.cpp +++ b/Source/Components/Interfaces/DebugService/DebugService.cpp @@ -7,11 +7,15 @@ #include "GAM.h" #include "GlobalObjectsDatabase.h" #include "HighResolutionTimer.h" +#include "LoggerService.h" #include "Message.h" #include "ObjectBuilder.h" #include "ObjectRegistryDatabase.h" +#include "Threads.h" +#include "Sleep.h" #include "StreamString.h" #include "TimeoutType.h" +#include "TcpLogger.h" #include "TypeConversion.h" #include "ReferenceT.h" @@ -95,6 +99,8 @@ DebugService::DebugService() isPaused = false; manualConfigSet = false; activeClient = NULL_PTR(BasicTCPSocket *); + streamerPacketOffset = 0u; + streamerSequenceNumber = 0u; } DebugService::~DebugService() { @@ -186,6 +192,70 @@ bool DebugService::Initialise(StructuredDataI &data) { return true; } +void DebugService::InjectTcpLoggerIfNeeded() { + if (logPort == 0u) + return; + + // Check if the ORD already contains a LoggerService with at least one TcpLogger. + // If so, leave it untouched. + Reference existing = ObjectRegistryDatabase::Instance()->Find("LoggerService"); + if (existing.IsValid()) { + ReferenceContainer *rc = dynamic_cast(existing.operator->()); + if (rc != NULL_PTR(ReferenceContainer *)) { + for (uint32 i = 0u; i < rc->Size(); i++) { + ReferenceT child = rc->Get(i); + if (child.IsValid()) { + printf("[DebugService] Found existing TcpLogger in LoggerService — skipping injection.\n"); + return; + } + } + } + } + + // Build a CDB that mirrors the config-file declaration: + // LoggerService node (current position) + // Class = LoggerService + // CPUs = 1 + // DebugConsumer (child) + // Class = TcpLogger + // Port = + ConfigurationDatabase lsCdb; + (void)lsCdb.Write("Class", "LoggerService"); + uint32 cpus = 1u; + (void)lsCdb.Write("CPUs", cpus); + // ReferenceContainer::Initialise only instantiates children whose names + // start with '+' (matching the StandardParser convention). + if (lsCdb.CreateRelative("+DebugConsumer")) { + (void)lsCdb.Write("Class", "TcpLogger"); + uint32 p = static_cast(logPort); + (void)lsCdb.Write("Port", p); + (void)lsCdb.MoveToAncestor(1u); + } + (void)lsCdb.MoveToRoot(); + + ReferenceT ls( + "LoggerService", GlobalObjectsDatabase::Instance()->GetStandardHeap()); + if (!ls.IsValid()) { + printf("[DebugService] Failed to create LoggerService object.\n"); + return; + } + ls->SetName("LoggerService"); + if (!ls->Initialise(lsCdb)) { + printf("[DebugService] LoggerService::Initialise() failed.\n"); + return; + } + + // Insert into the ORD so it is findable by name and cleaned up on shutdown. + if (!ObjectRegistryDatabase::Instance()->Insert(ls)) { + printf("[DebugService] Failed to insert LoggerService into ORD.\n"); + // Keep a local reference anyway so the logger thread stays alive. + } + + // The ORD now holds a reference to ls; the LoggerService stays alive until + // ObjectRegistryDatabase::Purge(). No need to store a local reference. + printf("[DebugService] Auto-injected LoggerService + TcpLogger on port %u.\n", logPort); +} + void DebugService::SetFullConfig(ConfigurationDatabase &config) { config.MoveToRoot(); config.Copy(fullConfig); @@ -215,9 +285,48 @@ static void BuildCDBFromContainer(ReferenceContainer *container, if (className != NULL_PTR(const char8 *)) (void)cdb.Write("Class", className); - // Export the object's own properties (standard fields, parameters). - // Custom config-file-only fields (PVName etc.) are not available here. - (void)child->ExportData(cdb); + // Export scalar parameters via a SEPARATE CDB so the live cursor is + // never touched by ExportData. Then copy only top-level LEAF values + // (i.e. scalars where MoveRelative fails) and skip sub-nodes. + // + // This avoids two ExportData pitfalls: + // 1. ReferenceContainer::ExportData writes numeric-indexed child nodes + // (+0, +1, ...) — those are sub-nodes and get filtered out. + // 2. Some DataSource ExportData implementations follow internal + // references and write sibling objects as children — also sub-nodes, + // also filtered out. + // + // Scalar parameters (ControlPort, UdpPort, CPUs, Port, ...) pass through + // because they are leaf values, not sub-nodes. + { + ConfigurationDatabase exportCdb; + if (child->ExportData(exportCdb)) { + exportCdb.MoveToRoot(); + uint32 nExport = exportCdb.GetNumberOfChildren(); + for (uint32 j = 0u; j < nExport; j++) { + const char8 *ek = exportCdb.GetChildName(j); + if (StringHelper::Compare(ek, "Class") == 0 || + StringHelper::Compare(ek, "Name") == 0 || + StringHelper::Compare(ek, "IsContainer") == 0) + continue; + // Sub-node check: MoveRelative succeeds only for nodes, not scalars + if (exportCdb.MoveRelative(ek)) { + exportCdb.MoveToAncestor(1u); + continue; // skip sub-nodes entirely + } + // Leaf scalar — convert to string and write into the main CDB + AnyType at = exportCdb.GetType(ek); + if (at.GetDataPointer() != NULL_PTR(void *)) { + char8 buf[1024]; + AnyType st(CharString, 0u, buf); + st.SetNumberOfElements(0, 1024); + if (TypeConvert(st, at)) { + (void)cdb.Write(ek, buf); + } + } + } + } + } ReferenceContainer *sub = dynamic_cast(child.operator->()); @@ -229,10 +338,24 @@ static void BuildCDBFromContainer(ReferenceContainer *container, } void DebugService::RebuildConfigFromRegistry() { - ConfigurationDatabase newConfig; - BuildCDBFromContainer(ObjectRegistryDatabase::Instance(), newConfig); - newConfig.MoveToRoot(); - newConfig.Copy(fullConfig); + fullConfig = ConfigurationDatabase(); + BuildCDBFromContainer(ObjectRegistryDatabase::Instance(), fullConfig); + + // ExportData on ReferenceContainer subclasses (including DebugService itself) + // only writes Name/IsContainer/indexed children — it never re-emits the + // config-file parameters that were read in Initialise(). Write them back + // explicitly from the member variables that Initialise() stored. + const char8 *myName = GetName(); + if (myName != NULL_PTR(const char8 *)) { + if (fullConfig.MoveRelative(myName)) { + (void)fullConfig.Write("ControlPort", static_cast(controlPort)); + (void)fullConfig.Write("UdpPort", static_cast(streamPort)); + (void)fullConfig.Write("LogPort", static_cast(logPort)); + if (streamIP.Size() > 0u) + (void)fullConfig.Write("StreamIP", streamIP.Buffer()); + (void)fullConfig.MoveToAncestor(1u); + } + } } static void PatchItemInternal(const char8 *originalName, @@ -416,63 +539,62 @@ ErrorManagement::ErrorType DebugService::Server(ExecutionInfo &info) { return ErrorManagement::NoError; if (info.GetStage() == ExecutionInfo::StartupStage) { serverThreadId = Threads::Id(); + // Wait for ObjectRegistryDatabase::Initialise() to finish processing all + // sibling objects (LoggerService etc. come after DebugService in the config). + // 500 ms is well above any realistic initialisation time. + Sleep::MSec(500u); + InjectTcpLoggerIfNeeded(); return ErrorManagement::NoError; } - while (info.GetStage() == ExecutionInfo::MainStage) { - while (activeClient == NULL_PTR(BasicTCPSocket *)) { - BasicTCPSocket *newClient = tcpServer.WaitConnection(TTInfiniteWait); - if (newClient != NULL_PTR(BasicTCPSocket *)) { - // Single connection mode: disconnect any existing client first - activeClient = newClient; - } + // The MARTe2 framework calls Execute() in a loop; each call should do + // one unit of work and return so the framework can check for Stop(). + // This replaces the old internal infinite-while pattern. + if (activeClient == NULL_PTR(BasicTCPSocket *)) { + // Wait briefly for a new connection; return so the framework loop can + // check if Stop() was requested between calls. + BasicTCPSocket *newClient = tcpServer.WaitConnection(TimeoutType(100)); + if (newClient != NULL_PTR(BasicTCPSocket *)) { + activeClient = newClient; } - // Single connection mode: only check client 0 - { - if (activeClient != NULL_PTR(BasicTCPSocket *)) { - // Check if client is still connected - if (!activeClient->IsConnected()) { - activeClient->Close(); - delete activeClient; - activeClient = NULL_PTR(BasicTCPSocket *); - - } else { - char buffer[1024]; - uint32 size = 1024; - if (activeClient->Read(buffer, size)) { - if (size > 0) { - // Process each line separately - char *ptr = buffer; - char *end = buffer + size; - while (ptr < end) { - char *newline = (char *)memchr(ptr, '\n', end - ptr); - if (!newline) { - break; - } - *newline = '\0'; - // Skip carriage return if present - if (newline > ptr && *(newline - 1) == '\r') - *(newline - 1) = '\0'; - StreamString command; - uint32 len = (uint32)(newline - ptr); - command.Write(ptr, len); - if (command.Size() > 0) { - HandleCommand(command, activeClient); - } - ptr = newline + 1; - } + } else { + // Check if client is still connected + if (!activeClient->IsConnected()) { + activeClient->Close(); + delete activeClient; + activeClient = NULL_PTR(BasicTCPSocket *); + } else { + char buffer[1024]; + uint32 size = 1024; + if (activeClient->Read(buffer, size)) { + if (size > 0) { + // Process each line separately + char *ptr = buffer; + char *end = buffer + size; + while (ptr < end) { + char *newline = (char *)memchr(ptr, '\n', end - ptr); + if (!newline) { + break; } - } else { - // // Read failed (client disconnected or error), clean up - if (activeClient != NULL_PTR(BasicTCPSocket *)) { - activeClient->Close(); - delete activeClient; - activeClient = NULL_PTR(BasicTCPSocket *); + *newline = '\0'; + // Skip carriage return if present + if (newline > ptr && *(newline - 1) == '\r') + *(newline - 1) = '\0'; + StreamString command; + uint32 len = (uint32)(newline - ptr); + command.Write(ptr, len); + if (command.Size() > 0) { + HandleCommand(command, activeClient); } + ptr = newline + 1; } } + } else { + // Read failed (client disconnected or error), clean up + activeClient->Close(); + delete activeClient; + activeClient = NULL_PTR(BasicTCPSocket *); } } - Sleep::MSec(10); } return ErrorManagement::NoError; } @@ -484,71 +606,68 @@ ErrorManagement::ErrorType DebugService::Streamer(ExecutionInfo &info) { streamerThreadId = Threads::Id(); return ErrorManagement::NoError; } + // Set UDP destination (idempotent, called each Execute() invocation) InternetHost dest(streamPort, streamIP.Buffer()); (void)udpSocket.SetDestination(dest); - uint8 packetBuffer[4096]; - uint32 packetOffset = 0; - uint32 sequenceNumber = 0; - while (info.GetStage() == ExecutionInfo::MainStage) { - // Poll monitored signals - uint64 currentTimeMs = (uint64)((float64)HighResolutionTimer::Counter() * - HighResolutionTimer::Period() * 1000.0); - mutex.FastLock(); - for (uint32 i = 0; i < monitoredSignals.Size(); i++) { - if (currentTimeMs >= (monitoredSignals[i].lastPollTime + monitoredSignals[i].periodMs)) { - monitoredSignals[i].lastPollTime = currentTimeMs; - uint64 ts = (uint64)((float64)HighResolutionTimer::Counter() * - HighResolutionTimer::Period() * 1000000.0); - - void *address = NULL_PTR(void *); - if (monitoredSignals[i].dataSource->GetSignalMemoryBuffer(monitoredSignals[i].signalIdx, 0, address)) { - traceBuffer.Push(monitoredSignals[i].internalID, ts, (uint8 *)address, monitoredSignals[i].size); - } - } - } - mutex.FastUnLock(); - uint32 id, size; - uint64 ts; - uint8 sampleData[1024]; - bool hasData = false; - while ((info.GetStage() == ExecutionInfo::MainStage) && - traceBuffer.Pop(id, ts, sampleData, size, 1024)) { - hasData = true; - if (packetOffset == 0) { - TraceHeader header; - header.magic = 0xDA7A57AD; - header.seq = sequenceNumber++; - header.timestamp = HighResolutionTimer::Counter(); - header.count = 0; - memcpy(packetBuffer, &header, sizeof(TraceHeader)); - packetOffset = sizeof(TraceHeader); + // Poll monitored signals + uint64 currentTimeMs = (uint64)((float64)HighResolutionTimer::Counter() * + HighResolutionTimer::Period() * 1000.0); + mutex.FastLock(); + for (uint32 i = 0; i < monitoredSignals.Size(); i++) { + if (currentTimeMs >= (monitoredSignals[i].lastPollTime + monitoredSignals[i].periodMs)) { + monitoredSignals[i].lastPollTime = currentTimeMs; + uint64 ts = (uint64)((float64)HighResolutionTimer::Counter() * + HighResolutionTimer::Period() * 1000000.0); + void *address = NULL_PTR(void *); + if (monitoredSignals[i].dataSource->GetSignalMemoryBuffer(monitoredSignals[i].signalIdx, 0, address)) { + traceBuffer.Push(monitoredSignals[i].internalID, ts, (uint8 *)address, monitoredSignals[i].size); } - if (packetOffset + 16 + size > 1400) { - uint32 toWrite = packetOffset; - (void)udpSocket.Write((char8 *)packetBuffer, toWrite); - TraceHeader header; - header.magic = 0xDA7A57AD; - header.seq = sequenceNumber++; - header.timestamp = HighResolutionTimer::Counter(); - header.count = 0; - memcpy(packetBuffer, &header, sizeof(TraceHeader)); - packetOffset = sizeof(TraceHeader); - } - memcpy(&packetBuffer[packetOffset], &id, 4); - memcpy(&packetBuffer[packetOffset + 4], &ts, 8); - memcpy(&packetBuffer[packetOffset + 12], &size, 4); - memcpy(&packetBuffer[packetOffset + 16], sampleData, size); - packetOffset += (16 + size); - ((TraceHeader *)packetBuffer)->count++; } - if (packetOffset > 0) { - uint32 toWrite = packetOffset; - (void)udpSocket.Write((char8 *)packetBuffer, toWrite); - packetOffset = 0; + } + mutex.FastUnLock(); + + // Drain ring buffer into UDP packet(s) + uint32 id, size; + uint64 ts; + uint8 sampleData[1024]; + bool hasData = false; + while (traceBuffer.Pop(id, ts, sampleData, size, 1024)) { + hasData = true; + if (streamerPacketOffset == 0u) { + TraceHeader header; + header.magic = 0xDA7A57AD; + header.seq = streamerSequenceNumber++; + header.timestamp = HighResolutionTimer::Counter(); + header.count = 0; + memcpy(streamerPacketBuffer, &header, sizeof(TraceHeader)); + streamerPacketOffset = sizeof(TraceHeader); } - if (!hasData) - Sleep::MSec(1); + if (streamerPacketOffset + 16u + size > 1400u) { + uint32 toWrite = streamerPacketOffset; + (void)udpSocket.Write((char8 *)streamerPacketBuffer, toWrite); + TraceHeader header; + header.magic = 0xDA7A57AD; + header.seq = streamerSequenceNumber++; + header.timestamp = HighResolutionTimer::Counter(); + header.count = 0; + memcpy(streamerPacketBuffer, &header, sizeof(TraceHeader)); + streamerPacketOffset = sizeof(TraceHeader); + } + memcpy(&streamerPacketBuffer[streamerPacketOffset], &id, 4); + memcpy(&streamerPacketBuffer[streamerPacketOffset + 4], &ts, 8); + memcpy(&streamerPacketBuffer[streamerPacketOffset + 12], &size, 4); + memcpy(&streamerPacketBuffer[streamerPacketOffset + 16], sampleData, size); + streamerPacketOffset += (16u + size); + ((TraceHeader *)streamerPacketBuffer)->count++; + } + if (streamerPacketOffset > 0u) { + uint32 toWrite = streamerPacketOffset; + (void)udpSocket.Write((char8 *)streamerPacketBuffer, toWrite); + streamerPacketOffset = 0u; + } + if (!hasData) { + Sleep::MSec(1); } return ErrorManagement::NoError; } @@ -659,67 +778,81 @@ void DebugService::HandleCommand(StreamString cmd, BasicTCPSocket *client) { msgConfig.Write("Mode", "ExpectsReply"); } - if (payload.Size() > 0u) { + // Parse payload key=value lines into a ConfigurationDatabase. + // ConstantGAM::SetOutput (and similar handlers) expect a + // ReferenceT inserted into the Message's + // reference container — NOT a sub-node of the message config. + ReferenceT paramCdb( + "ConfigurationDatabase", + GlobalObjectsDatabase::Instance()->GetStandardHeap()); + + if (payload.Size() > 0u && paramCdb.IsValid()) { payload.Seek(0u); StreamString line; while (payload.GetToken(line, "\n", term)) { if (line.Size() > 0u) { const char8 *eq = StringHelper::SearchChar(line.Buffer(), '='); if (eq != NULL_PTR(const char8 *)) { - StreamString key, val; uint32 eqPos = (uint32)(eq - line.Buffer()); (void)line.Seek(0u); - - char8* keyBuf = new char8[eqPos + 1]; + + char8 keyBuf[256] = {'\0'}; uint32 keyReadSize = eqPos; - if (line.Read(keyBuf, keyReadSize)) { - keyBuf[eqPos] = '\0'; - key = keyBuf; - } - delete[] keyBuf; - + (void)line.Read(keyBuf, keyReadSize); + (void)line.Seek(eqPos + 1u); - uint32 valLen = line.Size() - eqPos - 1u; - char8* valBuf = new char8[valLen + 1]; - uint32 valReadSize = valLen; - if (line.Read(valBuf, valReadSize)) { - valBuf[valLen] = '\0'; - val = valBuf; + uint32 valLen = (uint32)(line.Size() - eqPos - 1u); + char8 valBuf[1024] = {'\0'}; + (void)line.Read(valBuf, valLen); + + // Trim trailing whitespace from value + for (int32 ti = (int32)valLen - 1; ti >= 0; ti--) { + if (valBuf[ti] == ' ' || valBuf[ti] == '\r' || valBuf[ti] == '\t') + valBuf[ti] = '\0'; + else + break; + } + + StreamString key = keyBuf; + key = key.Buffer(); // trim happens via assignment + // Trim leading whitespace from key + const char8 *kp = keyBuf; + while (*kp == ' ' || *kp == '\t') kp++; + + if (*kp != '\0') { + (void)paramCdb->Write(kp, valBuf); } - delete[] valBuf; - - if (key.Size() > 0u) { - if (msgConfig.CreateRelative("Payload")) { - (void)msgConfig.Write(key.Buffer(), val.Buffer()); - (void)msgConfig.MoveToAncestor(1u); - } - } - } } line = ""; } } - - ErrorManagement::ErrorType err = ErrorManagement::ParametersError; - if (msg->Initialise(msgConfig)) { + + ErrorManagement::ErrorType err = ErrorManagement::ParametersError; + if (msg->Initialise(msgConfig)) { + if (paramCdb.IsValid() && payload.Size() > 0u) { + // Insert the CDB as a ReferenceT parameter + (void)msg->Insert(paramCdb); + } // Find destination object in the global database Reference destObj = ObjectRegistryDatabase::Instance()->Find(dest.Buffer()); if (destObj.IsValid()) { Object* sender = this; - // Double check if we are in the registry to be a valid sender StreamString myPath; if (!GetFullObjectName(*this, myPath)) { sender = NULL_PTR(Object*); } - + if (wait) { err = MessageI::WaitForReply(msg, TTInfiniteWait); } else { - err = MessageI::SendMessage(msg, sender); + (void)MessageI::SendMessage(msg, sender); + // Fire-and-forget: destination found, message sent. + // Whether the recipient had a matching filter is not + // reported back to the caller — return OK. + err = ErrorManagement::NoError; } } else { - printf(" MSG: Destination object %s not found in ORD\n", dest.Buffer()); } diff --git a/Source/Components/Interfaces/DebugService/DebugService.h b/Source/Components/Interfaces/DebugService/DebugService.h index 12d1de4..fe8e8c5 100644 --- a/Source/Components/Interfaces/DebugService/DebugService.h +++ b/Source/Components/Interfaces/DebugService/DebugService.h @@ -9,6 +9,7 @@ #include "MessageI.h" #include "Object.h" #include "ReferenceContainer.h" +#include "ReferenceT.h" #include "SingleThreadService.h" #include "StreamString.h" #include "Vec.h" @@ -92,6 +93,7 @@ public: private: void HandleCommand(StreamString cmd, BasicTCPSocket *client); void UpdateBrokersActiveStatus(); + void InjectTcpLoggerIfNeeded(); uint32 ExportTree(ReferenceContainer *container, StreamString &json, const char8 *pathPrefix); void PatchRegistry(); @@ -122,7 +124,6 @@ private: if (type == StreamerType) { return parent->Streamer(info); } - printf("serve TCP\n"); return parent->Server(info); } @@ -150,6 +151,11 @@ private: BasicTCPSocket *activeClient; + // Streamer state persisted across Execute() calls (framework loops Execute) + uint8 streamerPacketBuffer[4096]; + uint32 streamerPacketOffset; + uint32 streamerSequenceNumber; + ConfigurationDatabase fullConfig; bool manualConfigSet; diff --git a/Source/Components/Interfaces/TCPLogger/TcpLogger.cpp b/Source/Components/Interfaces/TCPLogger/TcpLogger.cpp index 1b5b211..1db34ad 100644 --- a/Source/Components/Interfaces/TCPLogger/TcpLogger.cpp +++ b/Source/Components/Interfaces/TCPLogger/TcpLogger.cpp @@ -35,6 +35,11 @@ TcpLogger::~TcpLogger() { clientsMutex.FastUnLock(); } +bool TcpLogger::ExportData(StructuredDataI & data) { + bool ok = data.Write("Port", static_cast(port)); + return ok; +} + bool TcpLogger::Initialise(StructuredDataI & data) { if (!ReferenceContainer::Initialise(data)) return false; @@ -92,64 +97,66 @@ ErrorManagement::ErrorType TcpLogger::Execute(ExecutionInfo & info) { return ErrorManagement::NoError; } - while (info.GetStage() == ExecutionInfo::MainStage) { - // 1. Check for new connections - BasicTCPSocket *newClient = server.WaitConnection(1); - if (newClient != NULL_PTR(BasicTCPSocket *)) { - clientsMutex.FastLock(); - bool added = false; - for (uint32 i=0; iClose(); - delete newClient; - } else { - (void)newClient->SetBlocking(false); + // Each Execute() call does one cycle. The MARTe2 framework loops Execute() + // so we must NOT spin in an infinite internal loop here — doing so prevents + // the framework from ever delivering the TerminationStage and causes + // Stop() to time out, leaving threads running after the destructor. + + // 1. Check for new connections (1 ms timeout → returns promptly) + BasicTCPSocket *newClient = server.WaitConnection(1); + if (newClient != NULL_PTR(BasicTCPSocket *)) { + clientsMutex.FastLock(); + bool added = false; + for (uint32 i=0; iWrite(packet.Buffer(), s)) { - activeClients[j]->Close(); - delete activeClients[j]; - activeClients[j] = NULL_PTR(BasicTCPSocket*); - } - } - } - clientsMutex.FastUnLock(); - readIdx = (readIdx + 1) % QUEUE_SIZE; - } - - if (!hadData) { - (void)eventSem.Wait(TimeoutType(100)); - eventSem.Reset(); + clientsMutex.FastUnLock(); + if (!added) { + newClient->Close(); + delete newClient; } else { - Sleep::MSec(1); + (void)newClient->SetBlocking(false); } } + + // 2. Stream queued entries to clients + bool hadData = false; + while (readIdx != writeIdx) { + hadData = true; + uint32 idx = readIdx % QUEUE_SIZE; + TcpLogEntry &entry = queue[idx]; + + StreamString level; + ErrorManagement::ErrorCodeToStream(entry.info.header.errorType, level); + + StreamString packet; + packet.Printf("LOG %s %s\n", level.Buffer(), entry.description); + uint32 size = packet.Size(); + + clientsMutex.FastLock(); + for (uint32 j=0; jWrite(packet.Buffer(), s)) { + activeClients[j]->Close(); + delete activeClients[j]; + activeClients[j] = NULL_PTR(BasicTCPSocket*); + } + } + } + clientsMutex.FastUnLock(); + readIdx = (readIdx + 1) % QUEUE_SIZE; + } + + if (!hadData) { + // Brief wait so we don't busy-spin; return so Stop() can take effect + (void)eventSem.Wait(TimeoutType(10)); + eventSem.Reset(); + } return ErrorManagement::NoError; } diff --git a/Source/Components/Interfaces/TCPLogger/TcpLogger.h b/Source/Components/Interfaces/TCPLogger/TcpLogger.h index 383682b..313e26b 100644 --- a/Source/Components/Interfaces/TCPLogger/TcpLogger.h +++ b/Source/Components/Interfaces/TCPLogger/TcpLogger.h @@ -29,6 +29,8 @@ public: virtual bool Initialise(StructuredDataI & data); + virtual bool ExportData(StructuredDataI & data); + /** * @brief Implementation of LoggerConsumerI. * Called by LoggerService. diff --git a/Test/Configurations/debug_test.cfg b/Test/Configurations/debug_test.cfg index 139d4bc..ff7b518 100644 --- a/Test/Configurations/debug_test.cfg +++ b/Test/Configurations/debug_test.cfg @@ -26,6 +26,16 @@ } } } + +CGAM = { + Class = ConstantGAM + OutputSignals = { + Test = { + DataSource = DDB2 + Type = float32 + Default = 0.123 + } + } + } +GAM2 = { Class = IOGAM InputSignals = { @@ -36,6 +46,10 @@ Time = { DataSource = TimerSlow } + Test = { + DataSource = DDB2 + Type = float32 + } } OutputSignals = { Counter = { @@ -46,6 +60,10 @@ Type = uint32 DataSource = Logger } + ConstOut = { + DataSource = Logger + Type = float32 + } } } +GAM3 = { @@ -122,6 +140,10 @@ } } } + +DDB2 = { + AllowNoProducer = 1 + Class = GAMDataSource + } +DDB3 = { AllowNoProducer = 1 Class = GAMDataSource @@ -142,7 +164,7 @@ } +Thread2 = { Class = RealTimeThread - Functions = {GAM2} + Functions = {GAM2 CGAM} } +Thread3 = { Class = RealTimeThread @@ -165,11 +187,3 @@ StreamIP = "127.0.0.1" } -+LoggerService = { - Class = LoggerService - CPUs = 0x1 - +DebugConsumer = { - Class = TcpLogger - Port = 8082 - } -} diff --git a/Test/Integration/ConfigCommandTest.cpp b/Test/Integration/ConfigCommandTest.cpp index 73f4750..20f7621 100644 --- a/Test/Integration/ConfigCommandTest.cpp +++ b/Test/Integration/ConfigCommandTest.cpp @@ -27,16 +27,18 @@ const char8 * const config_command_text = " CustomGAMField = \"GAMValue\" " " InputSignals = {" " Counter = { DataSource = Timer Type = uint32 Frequency = 1000 PVName = \"PROC:VAR:1\" }" +" Time = { DataSource = Timer Type = uint32 }" " }" " OutputSignals = {" " Counter = { DataSource = DDB Type = uint32 }" +" Time = { DataSource = DDB Type = uint32 }" " }" " }" " }" " +Data = {" " Class = ReferenceContainer " -" +Timer = { Class = LinuxTimer SleepTime = 1000 Signals = { Counter = { Type = uint32 } } }" -" +DDB = { Class = GAMDataSource Signals = { Counter = { Type = uint32 } } }" +" +Timer = { Class = LinuxTimer SleepTime = 1000 Signals = { Counter = { Type = uint32 } Time = { Type = uint32 } } }" +" +DDB = { Class = GAMDataSource Signals = { Counter = { Type = uint32 } Time = { Type = uint32 } } }" " +DAMS = { Class = TimingDataSource }" " }" " +States = {" @@ -97,16 +99,12 @@ void TestConfigCommands() { // Start the application to trigger broker execution and signal registration ReferenceT app = ObjectRegistryDatabase::Instance()->Find("App"); - if (app.IsValid()) { - if (app->ConfigureApplication()) { - if (app->PrepareNextState("State1") == ErrorManagement::NoError) { - if (app->StartNextStateExecution() == ErrorManagement::NoError) { - printf("Application started (for signal registration).\n"); - Sleep::MSec(500); // Wait for some cycles - } - } - } - } + assert(app.IsValid()); + assert(app->ConfigureApplication()); + assert(app->PrepareNextState("State1") == ErrorManagement::NoError); + assert(app->StartNextStateExecution() == ErrorManagement::NoError); + printf("Application started (for signal registration).\n"); + Sleep::MSec(500); // Wait for some cycles ReferenceT service = ObjectRegistryDatabase::Instance()->Find("DebugService"); if (service.IsValid()) { diff --git a/Test/Integration/IntegrationTests.cpp b/Test/Integration/IntegrationTests.cpp index 6c6ddf2..4e53b27 100644 --- a/Test/Integration/IntegrationTests.cpp +++ b/Test/Integration/IntegrationTests.cpp @@ -21,7 +21,7 @@ void timeout_handler(int sig) { } void ErrorProcessFunction(const MARTe::ErrorManagement::ErrorInformation &errorInfo, const char8 * const errorDescription) { - // printf("[MARTe Error] %s: %s\n", errorInfo.className, errorDescription); + printf("[MARTe Error] %s: %s\n", errorInfo.className, errorDescription); } // Forward declarations of other tests diff --git a/Test/Integration/TreeCommandTest.cpp b/Test/Integration/TreeCommandTest.cpp index 8c9d716..758b66c 100644 --- a/Test/Integration/TreeCommandTest.cpp +++ b/Test/Integration/TreeCommandTest.cpp @@ -39,17 +39,19 @@ void TestTreeCommand() { " Class = IOGAM " " InputSignals = {" " Counter = { DataSource = Timer Type = uint32 Frequency = 1000 }" + " Time = { DataSource = Timer Type = uint32 }" " }" " OutputSignals = {" " Counter = { DataSource = DDB Type = uint32 }" + " Time = { DataSource = DDB Type = uint32 }" " }" " }" " }" " +Data = {" " Class = ReferenceContainer " " DefaultDataSource = DDB " - " +Timer = { Class = LinuxTimer SleepTime = 1000 Signals = { Counter = { Type = uint32 } } }" - " +DDB = { Class = GAMDataSource Signals = { Counter = { Type = uint32 } } }" + " +Timer = { Class = LinuxTimer SleepTime = 1000 Signals = { Counter = { Type = uint32 } Time = { Type = uint32 } } }" + " +DDB = { Class = GAMDataSource Signals = { Counter = { Type = uint32 } Time = { Type = uint32 } } }" " +DAMS = { Class = TimingDataSource }" " }" " +States = {" diff --git a/Tools/gui_client/src/main.rs b/Tools/gui_client/src/main.rs index 18fb21c..b586028 100644 --- a/Tools/gui_client/src/main.rs +++ b/Tools/gui_client/src/main.rs @@ -155,12 +155,38 @@ struct PlotInstance { auto_bounds: bool, } +#[derive(Clone, PartialEq)] +enum MsgStatus { + Unknown, + Success, + Failure, +} + +#[derive(Clone)] +struct MessageHistoryEntry { + time: String, + destination: String, + function: String, + payload: String, + wait_reply: bool, + raw_cmd: String, + response: String, + status: MsgStatus, +} + +#[derive(Clone, PartialEq)] +enum MainTab { + Plots, + Config, +} + enum InternalEvent { Log(LogEntry), Discovery(Vec), Tree(TreeItem), CommandResponse(String), NodeInfo(String), + ConfigResponse(String), Connected, Disconnected, InternalLog(String), @@ -170,7 +196,7 @@ enum InternalEvent { UdpDropped(u32), RecordPathChosen(String, String), // SignalName, FilePath RecordingError(String, String), // SignalName, ErrorMessage - TelemMatched(u32), // Signal ID + TelemMatched(u32), ServiceConfig { udp_port: String, log_port: String }, } @@ -236,7 +262,6 @@ struct MarteDebugApp { node_info: String, udp_packets: u64, udp_dropped: u64, - telem_match_count: HashMap, forcing_dialog: Option, monitoring_dialog: Option, message_dialog: Option, @@ -246,6 +271,11 @@ struct MarteDebugApp { internal_tx: Sender, shared_x_range: Option<[f64; 2]>, scope: ScopeSettings, + active_main_tab: MainTab, + app_config_text: String, + message_history: Vec, + show_message_history: bool, + pending_msg_idx: Option, } impl MarteDebugApp { @@ -317,7 +347,6 @@ impl MarteDebugApp { node_info: "".to_string(), udp_packets: 0, udp_dropped: 0, - telem_match_count: HashMap::new(), forcing_dialog: None, monitoring_dialog: None, message_dialog: None, @@ -340,6 +369,11 @@ impl MarteDebugApp { last_trigger_time: 0.0, is_armed: true, }, + active_main_tab: MainTab::Plots, + app_config_text: String::new(), + message_history: Vec::new(), + show_message_history: true, + pending_msg_idx: None, } } @@ -545,6 +579,76 @@ impl MarteDebugApp { } } +/// Emit the body of a JSON object as MARTe2 config lines at the given indent level. +/// `Class` is always emitted first; child objects with a `Class` key get `+` prefix. +fn json_node_to_marte(map: &serde_json::Map, indent: usize) -> String { + let pad = " ".repeat(indent); + let mut out = String::new(); + + // Class first + if let Some(serde_json::Value::String(cls)) = map.get("Class") { + out.push_str(&format!("{}Class = {}\n", pad, cls)); + } + + // Collect and sort remaining keys so output is deterministic + let mut keys: Vec<&String> = map.keys().filter(|k| k.as_str() != "Class").collect(); + keys.sort(); + + for key in keys { + let val = &map[key]; + match val { + serde_json::Value::Object(child_map) => { + let prefix = if child_map.contains_key("Class") { "+" } else { "" }; + out.push_str(&format!("{}{}{} = {{\n", pad, prefix, key)); + out.push_str(&json_node_to_marte(child_map, indent + 1)); + out.push_str(&format!("{}}}\n", pad)); + } + serde_json::Value::String(s) => { + out.push_str(&format!("{}{} = {}\n", pad, key, s)); + } + serde_json::Value::Null => { + out.push_str(&format!("{}{} =\n", pad, key)); + } + other => { + out.push_str(&format!("{}{} = {}\n", pad, key, other)); + } + } + } + out +} + +fn convert_config_json(json_text: &str) -> String { + let root = match serde_json::from_str::(json_text) { + Ok(v) => v, + Err(_) => return json_text.to_string(), + }; + let map = match root.as_object() { + Some(m) => m, + None => return json_text.to_string(), + }; + let mut out = String::new(); + let mut keys: Vec<&String> = map.keys().collect(); + keys.sort(); + for key in keys { + let val = &map[key]; + match val { + serde_json::Value::Object(child_map) => { + // Top-level nodes always get + (they are named MARTe2 objects) + out.push_str(&format!("+{} = {{\n", key)); + out.push_str(&json_node_to_marte(child_map, 1)); + out.push_str("}\n\n"); + } + serde_json::Value::String(s) => { + out.push_str(&format!("{} = {}\n", key, s)); + } + other => { + out.push_str(&format!("{} = {}\n", key, other)); + } + } + } + out +} + fn tcp_command_worker( shared_config: Arc>, rx_cmd: Receiver, @@ -630,6 +734,12 @@ fn tcp_command_worker( let _ = tx_events_inner .send(InternalEvent::NodeInfo(json_clean.to_string())); json_acc.clear(); + } else if trimmed.contains("OK CONFIG") { + in_json = false; + let text = json_acc.split("OK CONFIG").next().unwrap_or("").trim(); + let _ = tx_events_inner + .send(InternalEvent::ConfigResponse(text.to_string())); + json_acc.clear(); } } else { if trimmed.starts_with("OK SERVICE_INFO") { @@ -882,6 +992,23 @@ fn udp_worker( last_warning_time = std::time::Instant::now(); } + // Resolve the base timestamp once per packet, not per signal. + // BASE_TELEM_TS is a global lazy Mutex; locking it inside the + // inner loop at 1 kHz × N_signals/packet was a major bottleneck. + let packet_base_ts: Option = { + let mut guard = BASE_TELEM_TS.lock().unwrap(); + if guard.is_none() { + // Peek at the first signal's timestamp to initialise base + if n >= 20 + 12 { + let first_ts = u64::from_le_bytes( + buf[20 + 4..20 + 12].try_into().unwrap(), + ); + *guard = Some(first_ts); + } + } + *guard + }; + for _ in 0..count { if offset + 16 > n { break; @@ -898,21 +1025,15 @@ fn udp_worker( } let data_slice = &buf[offset..offset + size as usize]; - let mut base_ts_guard = BASE_TELEM_TS.lock().unwrap(); - if base_ts_guard.is_none() { - *base_ts_guard = Some(ts_raw); - } - - let base = base_ts_guard.unwrap(); - let ts_s = if ts_raw >= base { - (ts_raw - base) as f64 / 1000000.0 + let ts_s = if let Some(base) = packet_base_ts { + if ts_raw >= base { (ts_raw - base) as f64 / 1_000_000.0 } else { 0.0 } } else { - 0.0 // Avoid huge jitter wrap-around + 0.0 }; - drop(base_ts_guard); if let Some(meta) = metas.get(&id) { - let _ = tx_events.send(InternalEvent::TelemMatched(id)); + // Do NOT send a channel event per signal — at 1kHz×N this + // floods the event queue and causes the UI spiral slowdown. let t = meta.sig_type.as_str(); let type_size = if meta.elements > 0 { size / meta.elements } else { size }; @@ -1110,15 +1231,30 @@ impl eframe::App for MarteDebugApp { }); } InternalEvent::CommandResponse(resp) => { + // Resolve pending MSG history entry + if let Some(idx) = self.pending_msg_idx { + if resp.contains("MSG") { + if let Some(entry) = self.message_history.get_mut(idx) { + entry.response = resp.clone(); + entry.status = if resp.starts_with("OK") { + MsgStatus::Success + } else { + MsgStatus::Failure + }; + } + self.pending_msg_idx = None; + } + } self.logs.push_back(LogEntry { time: Local::now().format("%H:%M:%S").to_string(), level: "CMD_RESP".to_string(), message: resp, }); } - InternalEvent::TelemMatched(id) => { - *self.telem_match_count.entry(id).or_insert(0) += 1; + InternalEvent::ConfigResponse(text) => { + self.app_config_text = convert_config_json(&text); } + InternalEvent::TelemMatched(_) => {} InternalEvent::RecordPathChosen(name, path) => { let mut data_map = self.traced_signals.lock().unwrap(); if let Some(entry) = data_map.get_mut(&name) { @@ -1275,16 +1411,23 @@ impl eframe::App for MarteDebugApp { ui.horizontal(|ui| { if ui.button("🚀 Send").clicked() { let wait = if dialog.expect_reply { "1" } else { "0" }; - // Replace actual newlines with literal '\n' for the server-side tokenizer if needed, - // or ensure the server handles the raw multi-line stream if the protocol allows it. - // Given HandleCommand reads line-by-line, we must send it carefully. - // Actually, our Server loop reads up to \n. - // So we should encode newlines in payload if we want to send them in one go. let encoded_payload = dialog.payload.replace('\n', "\\n"); let cmd = format!( "MSG {} {} {} {}", dialog.destination, dialog.function, wait, encoded_payload ); + let idx = self.message_history.len(); + self.message_history.push(MessageHistoryEntry { + time: Local::now().format("%H:%M:%S").to_string(), + destination: dialog.destination.clone(), + function: dialog.function.clone(), + payload: dialog.payload.clone(), + wait_reply: dialog.expect_reply, + raw_cmd: cmd.clone(), + response: String::new(), + status: MsgStatus::Unknown, + }); + self.pending_msg_idx = Some(idx); let _ = self.tx_cmd.send(cmd); close = true; } @@ -1346,6 +1489,7 @@ impl eframe::App for MarteDebugApp { ui.horizontal(|ui| { ui.toggle_value(&mut self.show_left_panel, "🗂 Tree"); ui.toggle_value(&mut self.show_right_panel, "📊 Signals"); + ui.toggle_value(&mut self.show_message_history, "💬 Msgs"); ui.toggle_value(&mut self.show_bottom_panel, "📜 Logs"); ui.separator(); if ui.button("➕ Plot").clicked() { @@ -1629,6 +1773,85 @@ impl eframe::App for MarteDebugApp { for key in to_delete.iter() { self.forced_signals.remove(key); } + + if self.show_message_history { + ui.separator(); + ui.horizontal(|ui| { + ui.heading("Message History"); + ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| { + if ui.small_button("🗑 Clear").clicked() { + self.message_history.clear(); + self.pending_msg_idx = None; + } + }); + }); + if self.message_history.is_empty() { + ui.label(egui::RichText::new("No messages sent yet").italics().color(egui::Color32::GRAY)); + } else { + egui::ScrollArea::vertical() + .id_salt("msg_history_scroll") + .max_height(300.0) + .auto_shrink([false, true]) + .show(ui, |ui| { + let mut resend_cmd: Option = None; + let mut edit_dialog: Option = None; + for entry in self.message_history.iter().rev() { + let (status_icon, status_color) = match entry.status { + MsgStatus::Success => ("✔", egui::Color32::from_rgb(100, 220, 100)), + MsgStatus::Failure => ("✘", egui::Color32::from_rgb(255, 100, 100)), + MsgStatus::Unknown => ("…", egui::Color32::from_rgb(255, 220, 50)), + }; + ui.group(|ui| { + ui.horizontal(|ui| { + ui.label(egui::RichText::new(status_icon).color(status_color).strong()); + ui.label(egui::RichText::new(&entry.time).color(egui::Color32::GRAY).monospace().small()); + }); + ui.label(format!("→ {}.{}", entry.destination, entry.function)); + if !entry.payload.is_empty() { + ui.label(egui::RichText::new(&entry.payload).monospace().small().color(egui::Color32::from_rgb(180, 180, 255))); + } + if !entry.response.is_empty() { + ui.label(egui::RichText::new(&entry.response).small().color(status_color)); + } + ui.horizontal(|ui| { + if ui.small_button("↩ Resend").clicked() { + resend_cmd = Some(entry.raw_cmd.clone()); + } + if ui.small_button("✏ Edit").clicked() { + edit_dialog = Some(MessageDialog { + destination: entry.destination.clone(), + function: entry.function.clone(), + payload: entry.payload.clone(), + expect_reply: entry.wait_reply, + }); + } + }); + }); + } + if let Some(cmd) = resend_cmd { + let idx = self.message_history.len(); + // find the matching history entry to clone metadata + if let Some(orig) = self.message_history.iter().find(|e| e.raw_cmd == cmd) { + self.message_history.push(MessageHistoryEntry { + time: Local::now().format("%H:%M:%S").to_string(), + destination: orig.destination.clone(), + function: orig.function.clone(), + payload: orig.payload.clone(), + wait_reply: orig.wait_reply, + raw_cmd: cmd.clone(), + response: String::new(), + status: MsgStatus::Unknown, + }); + } + self.pending_msg_idx = Some(idx); + let _ = self.tx_cmd.send(cmd); + } + if let Some(dialog) = edit_dialog { + self.message_dialog = Some(dialog); + } + }); + } + } }); } @@ -1712,6 +1935,35 @@ impl eframe::App for MarteDebugApp { } egui::CentralPanel::default().show(ctx, |ui| { + ui.horizontal(|ui| { + ui.selectable_value(&mut self.active_main_tab, MainTab::Plots, "📈 Plots"); + ui.selectable_value(&mut self.active_main_tab, MainTab::Config, "⚙ Config"); + }); + ui.separator(); + if self.active_main_tab == MainTab::Config { + ui.horizontal(|ui| { + if ui.button("🔄 Refresh").clicked() { + let _ = self.tx_cmd.send("CONFIG".to_string()); + } + ui.label(egui::RichText::new("Application Configuration").color(egui::Color32::GRAY)); + }); + ui.separator(); + egui::ScrollArea::both() + .auto_shrink([false, false]) + .show(ui, |ui| { + if self.app_config_text.is_empty() { + ui.label(egui::RichText::new("Press Refresh to load the configuration").italics().color(egui::Color32::GRAY)); + } else { + ui.add( + egui::Label::new( + egui::RichText::new(&self.app_config_text).monospace(), + ) + .selectable(true), + ); + } + }); + return; + } let n_plots = self.plots.len(); if n_plots > 0 { let plot_height = ui.available_height() / n_plots as f32; @@ -1909,6 +2161,241 @@ impl eframe::App for MarteDebugApp { } } +#[cfg(test)] +mod tests { + use super::*; + + /// Parse MARTe2 config text into a serde_json Value tree so tests can do + /// structural comparisons without caring about whitespace or key ordering. + /// + /// Rules: + /// `[+]Name = {` → start of a named block + /// `}` → close current block + /// `Name = Value` → leaf key-value (value stripped of surrounding quotes) + /// `Name = {value}` (e.g. `Functions = {GAM1}`) → leaf, not a block + fn parse_marte_config(text: &str) -> serde_json::Value { + let mut stack: Vec> = + vec![serde_json::Map::new()]; + let mut name_stack: Vec = vec![]; + for line in text.lines() { + let trimmed = line.trim(); + if trimmed.is_empty() || trimmed.starts_with("//") || trimmed.starts_with("/*") { + continue; + } + if trimmed == "}" { + if stack.len() > 1 { + let child = stack.pop().unwrap(); + let name = name_stack.pop().unwrap(); + stack.last_mut().unwrap().insert(name, serde_json::Value::Object(child)); + } + continue; + } + let clean = trimmed.trim_start_matches('+'); + if let Some(eq_pos) = clean.find('=') { + let key = clean[..eq_pos].trim().to_string(); + let val = clean[eq_pos + 1..].trim(); + // A block opens only if the value is exactly `{` + if val == "{" { + stack.push(serde_json::Map::new()); + name_stack.push(key); + } else { + let val_clean = val.trim_matches('"').to_string(); + stack.last_mut().unwrap().insert(key, serde_json::Value::String(val_clean)); + } + } + } + serde_json::Value::Object(stack.remove(0)) + } + + /// Recursively assert that every object key with a `Class` value in `expected` + /// also exists with the same `Class` in `actual`. Missing non-Class keys are + /// tolerated because `ExportData()` does not re-emit config-file-only fields + /// (e.g. signal configurations, Frequency, Samples). + fn assert_classes_match( + expected: &serde_json::Value, + actual: &serde_json::Value, + path: &str, + ) { + use serde_json::Value; + if let (Value::Object(exp_map), Value::Object(act_map)) = (expected, actual) { + if let Some(Value::String(exp_class)) = exp_map.get("Class") { + let act_class = act_map + .get("Class") + .and_then(|v| v.as_str()) + .unwrap_or(""); + assert_eq!( + exp_class.as_str(), + act_class, + "Class mismatch at '{}': expected '{}', got '{}'", + path, + exp_class, + act_class + ); + } + for (key, exp_child) in exp_map { + if key == "Class" { + continue; + } + if let Some(act_child) = act_map.get(key) { + let child_path = if path.is_empty() { + key.clone() + } else { + format!("{}.{}", path, key) + }; + assert_classes_match(exp_child, act_child, &child_path); + } + // Extra keys added by MARTe2 runtime are allowed + } + } + } + + // ----- Unit tests for the JSON → MARTe2 converter ----- + + #[test] + fn test_convert_basic_structure() { + let json = r#"{"App":{"Class":"RealTimeApplication","Functions":{"Class":"ReferenceContainer","GAM1":{"Class":"IOGAM"}}}}"#; + let out = convert_config_json(json); + let parsed = parse_marte_config(&out); + + assert_eq!( + parsed["App"]["Class"], + serde_json::Value::String("RealTimeApplication".into()) + ); + assert_eq!( + parsed["App"]["Functions"]["Class"], + serde_json::Value::String("ReferenceContainer".into()) + ); + assert_eq!( + parsed["App"]["Functions"]["GAM1"]["Class"], + serde_json::Value::String("IOGAM".into()) + ); + // Objects with Class must get + prefix + assert!(out.contains("+App = {"), "top-level object missing +"); + assert!(out.contains(" +Functions = {"), "nested object with Class missing +"); + assert!(out.contains(" +GAM1 = {"), "leaf object with Class missing +"); + } + + #[test] + fn test_no_plus_for_blocks_without_class() { + let json = r#"{"App":{"Class":"RealTimeApplication","Signals":{"Counter":{"Type":"uint32"}}}}"#; + let out = convert_config_json(json); + // Signals block has no Class → no + prefix + assert!(out.contains(" Signals = {"), "plain block should not have +"); + assert!(!out.contains(" +Signals"), "plain block must not have + prefix"); + } + + #[test] + fn test_class_emitted_first() { + let json = r#"{"App":{"ZZZKey":"last","Class":"RealTimeApplication","AAA":"first_alpha"}}"#; + let out = convert_config_json(json); + let class_pos = out.find("Class = RealTimeApplication").unwrap(); + let zzz_pos = out.find("ZZZKey = last").unwrap(); + assert!( + class_pos < zzz_pos, + "Class must appear before other keys in the block" + ); + } + + #[test] + fn test_leaf_values_preserved() { + let json = r#"{"App":{"Class":"RealTimeApplication","ControlPort":"8080","StreamIP":"127.0.0.1"}}"#; + let out = convert_config_json(json); + assert!(out.contains("ControlPort = 8080")); + assert!(out.contains("StreamIP = 127.0.0.1")); + } + + #[test] + fn test_round_trip_parse() { + // Build a JSON that mirrors what the server produces for debug_test.cfg + let json = r#"{ + "App": { + "Class": "RealTimeApplication", + "Data": {"Class": "ReferenceContainer", + "DDB": {"Class": "GAMDataSource"}, + "Timer": {"Class": "LinuxTimer"} + }, + "Functions": {"Class": "ReferenceContainer", + "GAM1": {"Class": "IOGAM"}, + "GAM2": {"Class": "IOGAM"} + }, + "Scheduler": {"Class": "GAMScheduler", "TimingDataSource": "DAMS"}, + "States": {"Class": "ReferenceContainer", + "State1": {"Class": "RealTimeState"} + } + }, + "DebugService": { + "Class": "DebugService", + "ControlPort": "8080", + "UdpPort": "8081" + } + }"#; + let out = convert_config_json(json); + let cfg_path = concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../Test/Configurations/debug_test.cfg" + ); + let original = std::fs::read_to_string(cfg_path).expect("debug_test.cfg not found"); + + // Parse both into trees and check that all Class values from the + // converted output are consistent with the original config + let converted_tree = parse_marte_config(&out); + let original_tree = parse_marte_config(&original); + assert_classes_match(&converted_tree, &original_tree, ""); + // And vice-versa for the subset present in the JSON fixture + assert_classes_match(&original_tree, &converted_tree, ""); + } + + // ----- Integration test (requires running MARTe2 app) ----- + + #[test] + #[ignore = "requires running MARTe2 debug app — start with ./run_debug_app.sh first"] + fn test_live_config_matches_debug_test_cfg() { + use std::io::{BufRead, BufReader, Write}; + use std::net::TcpStream; + use std::time::Duration; + + let mut stream = TcpStream::connect("127.0.0.1:8080") + .expect("Could not connect to DebugService on 127.0.0.1:8080"); + stream + .set_read_timeout(Some(Duration::from_secs(15))) + .unwrap(); + stream.write_all(b"CONFIG\n").unwrap(); + + // Accumulate lines until the "OK CONFIG" sentinel + let reader = BufReader::new(stream); + let mut json_lines: Vec = Vec::new(); + for line in reader.lines() { + let line = line.expect("read error while receiving CONFIG response"); + if line.trim() == "OK CONFIG" { + break; + } + json_lines.push(line); + } + let json_text = json_lines.join("\n"); + assert!(!json_text.is_empty(), "Received empty CONFIG response from server"); + + // Convert server JSON to MARTe2 syntax + let converted = convert_config_json(&json_text); + assert!(!converted.is_empty(), "convert_config_json produced empty output"); + + // Load the reference config + let cfg_path = concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../Test/Configurations/debug_test.cfg" + ); + let original = std::fs::read_to_string(cfg_path).expect("debug_test.cfg not found"); + + let expected_tree = parse_marte_config(&original); + let actual_tree = parse_marte_config(&converted); + + // Every named object present in the original config must appear in the live + // config with the same Class. Config-file-only fields (InputSignals, + // OutputSignals, Frequency, etc.) are intentionally not checked because + // ExportData() does not re-emit them after ConfigureApplication(). + assert_classes_match(&expected_tree, &actual_tree, ""); + } +} + fn main() -> Result<(), eframe::Error> { let options = eframe::NativeOptions { viewport: egui::ViewportBuilder::default().with_inner_size([1280.0, 800.0]),