/** * @file UDPStreamerClientTest.cpp * @brief Source file for class UDPStreamerClientTest * @date 01/07/2026 * @author Martino Ferrari * * @copyright Copyright 2015 F4E | European Joint Undertaking for ITER and * the Development of Fusion Energy ('Fusion for Energy'). * Licensed under the EUPL, Version 1.1 or - as soon they will be approved * by the European Commission - subsequent versions of the EUPL (the "Licence") * You may not use this work except in compliance with the Licence. * You may obtain a copy of the Licence at: http://ec.europa.eu/idabc/eupl * * @warning Unless required by applicable law or agreed to in writing, * software distributed under the Licence is distributed on an "AS IS" * basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express * or implied. See the Licence permissions and limitations under the Licence. * * @details This source file contains the definition of all the methods for * the class UDPStreamerClientTest (public, protected, and private). */ #define DLL_API /*---------------------------------------------------------------------------*/ /* Standard header includes */ /*---------------------------------------------------------------------------*/ #include /*---------------------------------------------------------------------------*/ /* Project header includes */ /*---------------------------------------------------------------------------*/ #include "AdvancedErrorManagement.h" #include "BasicUDPSocket.h" #include "ConfigurationDatabase.h" #include "GAM.h" #include "MemoryOperationsHelper.h" #include "ObjectRegistryDatabase.h" #include "RealTimeApplication.h" #include "Sleep.h" #include "StandardParser.h" #include "Threads.h" #include "UDPSProtocol.h" #include "UDPStreamerClient.h" #include "UDPStreamerClientTest.h" /*---------------------------------------------------------------------------*/ /* Static definitions */ /*---------------------------------------------------------------------------*/ /** * @brief Minimal pass-through GAM used instead of IOGAM, so the test does * not depend on IOGAM.so being dlopen()-able / on the loader path. * Copies each input signal to the matching output signal. */ class UDPStreamerClientTestGAM : public MARTe::GAM { public: CLASS_REGISTER_DECLARATION() UDPStreamerClientTestGAM() : GAM() { } ~UDPStreamerClientTestGAM() { } bool Execute() { for (MARTe::uint32 i = 0u; i < GetNumberOfOutputSignals(); i++) { void *outMem = GetOutputSignalMemory(i); void *inMem = GetInputSignalMemory(i); if ((outMem != NULL_PTR(void *)) && (inMem != NULL_PTR(void *))) { MARTe::uint32 sz = 0u; GetSignalByteSize(MARTe::OutputSignals, i, sz); (void) MARTe::MemoryOperationsHelper::Copy(outMem, inMem, sz); } } return true; } bool Setup() { return true; } }; CLASS_REGISTER(UDPStreamerClientTestGAM, "1.0") /** * @brief Shared state between the test thread and a background * UDPStreamerClient::Synchronise() call (see StartSynchroniseThread). */ struct SynchroniseThreadArgs { MARTe::ReferenceT ds; bool result; bool done; }; static void SynchroniseThreadEntry(const void *const parameters) { SynchroniseThreadArgs *args = const_cast( reinterpret_cast(parameters)); args->result = args->ds->Synchronise(); args->done = true; } /** * @brief UDPStreamerClient::Synchronise() calls dataSem.ResetWait(), which * first lowers (Reset()s) the semaphore barrier and only then waits * (with an internal ~10 ms timeout) for it to be raised again. * Because of this, a Post() (triggered by OnUDPSData()) that happens * BEFORE Synchronise() is called is wiped out by that leading Reset() * and lost. In production this is never an issue because the RT * thread's Synchronise() call is already blocked in Wait() when the * receiver thread posts new data. * * To exercise that same interleaving from a single-threaded test, run * Synchronise() on a background thread and give it a very brief head * start (shorter than the internal timeout) so it has a chance to * pass through Reset() and be parked in Wait() before the caller * triggers the data delivery. Thread-scheduling latency means this is * still a race, not a guarantee, so callers should retry the * deliver+StartSynchroniseThread+JoinSynchroniseThread sequence a few * times until the expected value shows up (see e.g. * TestOnUDPSConfig_BasicAccept for the pattern). */ static SynchroniseThreadArgs *StartSynchroniseThread(MARTe::ReferenceT ds) { using namespace MARTe; SynchroniseThreadArgs *args = new SynchroniseThreadArgs(); args->ds = ds; args->result = false; args->done = false; (void) Threads::BeginThread(&SynchroniseThreadEntry, args, THREADS_DEFAULT_STACKSIZE, "SyncTest"); /* Give the thread a chance to start running and reach Reset()+Wait() * before the caller delivers data. Must stay well below Synchronise()'s * internal ~10 ms timeout, or the background call will already have * timed out and returned by the time we come back here. */ Sleep::MSec(1u); return args; } /** * @brief Waits for the background Synchronise() started by * StartSynchroniseThread() to complete, returns its result and frees * the shared state. */ static bool JoinSynchroniseThread(SynchroniseThreadArgs *args) { using namespace MARTe; uint32 waited = 0u; while ((!args->done) && (waited < 500u)) { Sleep::MSec(2u); waited += 2u; } bool result = args->done && args->result; delete args; return result; } /** * @brief Helper: build a RealTimeApplication from an MARTe2 config string. * Returns the application reference (invalid if parsing failed). */ static MARTe::ReferenceT LoadApplication( const MARTe::char8 *const config) { using namespace MARTe; ConfigurationDatabase cdb; StreamString cfgStr = config; (void) cfgStr.Seek(0LLU); StandardParser parser(cfgStr, cdb); if (!parser.Parse()) { return ReferenceT(); } ObjectRegistryDatabase *god = ObjectRegistryDatabase::Instance(); god->Purge(); if (!god->Initialise(cdb)) { return ReferenceT(); } ReferenceT app = god->Find("Test"); if (!app.IsValid()) { return ReferenceT(); } if (!app->ConfigureApplication()) { return ReferenceT(); } return app; } /** * @brief Minimal single-scalar-signal config template. * %d placeholders: Port. */ static MARTe::StreamString BuildBasicConfig(MARTe::uint16 port) { using namespace MARTe; StreamString cfg; cfg.Printf( "+Test = {\n" " Class = RealTimeApplication\n" " +Functions = {\n" " Class = ReferenceContainer\n" " +Reader = {\n" " Class = UDPStreamerClientTestGAM\n" " InputSignals = {\n" " Counter = {\n" " DataSource = ClientDS\n" " Type = uint32\n" " }\n" " }\n" " OutputSignals = {\n" " Counter = {\n" " DataSource = DDB\n" " Type = uint32\n" " }\n" " }\n" " }\n" " }\n" " +Data = {\n" " Class = ReferenceContainer\n" " DefaultDataSource = DDB\n" " +DDB = {\n" " Class = GAMDataSource\n" " }\n" " +ClientDS = {\n" " Class = UDPStreamerClient\n" " ServerAddress = \"127.0.0.1\"\n" " Port = %d\n" " MaxPayloadSize = 1400\n" " Signals = {\n" " Counter = {\n" " Type = uint32\n" " }\n" " }\n" " }\n" " +Timings = {\n" " Class = TimingDataSource\n" " }\n" " }\n" " +States = {\n" " Class = ReferenceContainer\n" " +State1 = {\n" " Class = RealTimeState\n" " +Threads = {\n" " Class = ReferenceContainer\n" " +Thread1 = {\n" " Class = RealTimeThread\n" " Functions = { Reader }\n" " }\n" " }\n" " }\n" " }\n" " +Scheduler = {\n" " Class = GAMScheduler\n" " TimingDataSource = Timings\n" " }\n" "}\n", port); return cfg; } /*---------------------------------------------------------------------------*/ /* Method definitions */ /*---------------------------------------------------------------------------*/ bool UDPStreamerClientTest::TestInitialise_Valid() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44700u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestInitialise_DefaultServerAddress() { using namespace MARTe; static const char8 *const cfg = "+Test = {\n" " Class = RealTimeApplication\n" " +Functions = {\n" " Class = ReferenceContainer\n" " +Reader = {\n" " Class = UDPStreamerClientTestGAM\n" " InputSignals = { Counter = { DataSource = ClientDS Type = uint32 } }\n" " OutputSignals = { Counter = { DataSource = DDB Type = uint32 } }\n" " }\n" " }\n" " +Data = {\n" " Class = ReferenceContainer\n" " DefaultDataSource = DDB\n" " +DDB = { Class = GAMDataSource }\n" " +ClientDS = {\n" " Class = UDPStreamerClient\n" " Port = 44701\n" " Signals = { Counter = { Type = uint32 } }\n" " }\n" " +Timings = { Class = TimingDataSource }\n" " }\n" " +States = {\n" " Class = ReferenceContainer\n" " +State1 = {\n" " Class = RealTimeState\n" " +Threads = {\n" " Class = ReferenceContainer\n" " +Thread1 = { Class = RealTimeThread Functions = { Reader } }\n" " }\n" " }\n" " }\n" " +Scheduler = { Class = GAMScheduler TimingDataSource = Timings }\n" "}\n"; ReferenceT app = LoadApplication(cfg); bool ok = app.IsValid(); ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestInitialise_DefaultPort() { using namespace MARTe; static const char8 *const cfg = "+Test = {\n" " Class = RealTimeApplication\n" " +Functions = {\n" " Class = ReferenceContainer\n" " +Reader = {\n" " Class = UDPStreamerClientTestGAM\n" " InputSignals = { Counter = { DataSource = ClientDS Type = uint32 } }\n" " OutputSignals = { Counter = { DataSource = DDB Type = uint32 } }\n" " }\n" " }\n" " +Data = {\n" " Class = ReferenceContainer\n" " DefaultDataSource = DDB\n" " +DDB = { Class = GAMDataSource }\n" " +ClientDS = {\n" " Class = UDPStreamerClient\n" " ServerAddress = \"127.0.0.1\"\n" " Signals = { Counter = { Type = uint32 } }\n" " }\n" " +Timings = { Class = TimingDataSource }\n" " }\n" " +States = {\n" " Class = ReferenceContainer\n" " +State1 = {\n" " Class = RealTimeState\n" " +Threads = {\n" " Class = ReferenceContainer\n" " +Thread1 = { Class = RealTimeThread Functions = { Reader } }\n" " }\n" " }\n" " }\n" " +Scheduler = { Class = GAMScheduler TimingDataSource = Timings }\n" "}\n"; ReferenceT app = LoadApplication(cfg); bool ok = app.IsValid(); ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestInitialise_MulticastMode_Valid() { using namespace MARTe; static const char8 *const cfg = "+Test = {\n" " Class = RealTimeApplication\n" " +Functions = {\n" " Class = ReferenceContainer\n" " +Reader = {\n" " Class = UDPStreamerClientTestGAM\n" " InputSignals = { Counter = { DataSource = ClientDS Type = uint32 } }\n" " OutputSignals = { Counter = { DataSource = DDB Type = uint32 } }\n" " }\n" " }\n" " +Data = {\n" " Class = ReferenceContainer\n" " DefaultDataSource = DDB\n" " +DDB = { Class = GAMDataSource }\n" " +ClientDS = {\n" " Class = UDPStreamerClient\n" " ServerAddress = \"127.0.0.1\"\n" " Port = 44702\n" " MulticastGroup = \"239.0.0.5\"\n" " DataPort = 44703\n" " Signals = { Counter = { Type = uint32 } }\n" " }\n" " +Timings = { Class = TimingDataSource }\n" " }\n" " +States = {\n" " Class = ReferenceContainer\n" " +State1 = {\n" " Class = RealTimeState\n" " +Threads = {\n" " Class = ReferenceContainer\n" " +Thread1 = { Class = RealTimeThread Functions = { Reader } }\n" " }\n" " }\n" " }\n" " +Scheduler = { Class = GAMScheduler TimingDataSource = Timings }\n" "}\n"; ReferenceT app = LoadApplication(cfg); bool ok = app.IsValid(); ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestInitialise_MulticastMode_DefaultDataPort() { using namespace MARTe; static const char8 *const cfg = "+Test = {\n" " Class = RealTimeApplication\n" " +Functions = {\n" " Class = ReferenceContainer\n" " +Reader = {\n" " Class = UDPStreamerClientTestGAM\n" " InputSignals = { Counter = { DataSource = ClientDS Type = uint32 } }\n" " OutputSignals = { Counter = { DataSource = DDB Type = uint32 } }\n" " }\n" " }\n" " +Data = {\n" " Class = ReferenceContainer\n" " DefaultDataSource = DDB\n" " +DDB = { Class = GAMDataSource }\n" " +ClientDS = {\n" " Class = UDPStreamerClient\n" " ServerAddress = \"127.0.0.1\"\n" " Port = 44704\n" " MulticastGroup = \"239.0.0.6\"\n" " Signals = { Counter = { Type = uint32 } }\n" " }\n" " +Timings = { Class = TimingDataSource }\n" " }\n" " +States = {\n" " Class = ReferenceContainer\n" " +State1 = {\n" " Class = RealTimeState\n" " +Threads = {\n" " Class = ReferenceContainer\n" " +Thread1 = { Class = RealTimeThread Functions = { Reader } }\n" " }\n" " }\n" " }\n" " +Scheduler = { Class = GAMScheduler TimingDataSource = Timings }\n" "}\n"; ReferenceT app = LoadApplication(cfg); bool ok = app.IsValid(); ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestSetConfiguredDatabase_MultipleSignals() { using namespace MARTe; static const char8 *const cfg = "+Test = {\n" " Class = RealTimeApplication\n" " +Functions = {\n" " Class = ReferenceContainer\n" " +Reader = {\n" " Class = UDPStreamerClientTestGAM\n" " InputSignals = {\n" " Counter = { DataSource = ClientDS Type = uint32 }\n" " Samples = { DataSource = ClientDS Type = float32 NumberOfDimensions = 1 NumberOfElements = 4 }\n" " }\n" " OutputSignals = {\n" " Counter = { DataSource = DDB Type = uint32 }\n" " Samples = { DataSource = DDB Type = float32 NumberOfDimensions = 1 NumberOfElements = 4 }\n" " }\n" " }\n" " }\n" " +Data = {\n" " Class = ReferenceContainer\n" " DefaultDataSource = DDB\n" " +DDB = { Class = GAMDataSource }\n" " +ClientDS = {\n" " Class = UDPStreamerClient\n" " ServerAddress = \"127.0.0.1\"\n" " Port = 44705\n" " Signals = {\n" " Counter = { Type = uint32 }\n" " Samples = { Type = float32 NumberOfDimensions = 1 NumberOfElements = 4 }\n" " }\n" " }\n" " +Timings = { Class = TimingDataSource }\n" " }\n" " +States = {\n" " Class = ReferenceContainer\n" " +State1 = {\n" " Class = RealTimeState\n" " +Threads = {\n" " Class = ReferenceContainer\n" " +Thread1 = { Class = RealTimeThread Functions = { Reader } }\n" " }\n" " }\n" " }\n" " +Scheduler = { Class = GAMScheduler TimingDataSource = Timings }\n" "}\n"; ReferenceT app = LoadApplication(cfg); bool ok = app.IsValid(); ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestPrepareNextState_StartsReceiver() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44706u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); if (ok) { ok = (app->PrepareNextState("State1") == MARTe::ErrorManagement::NoError); } Sleep::MSec(50u); ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestSynchronise_NoData() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44707u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { /* No CONFIG/DATA ever delivered: Synchronise must be a safe no-op * and still return true. */ ok = ds->Synchronise(); } void *sigMem = NULL_PTR(void *); if (ok) { ok = ds->GetSignalMemoryBuffer(0u, 0u, sigMem); } if (ok) { uint32 value = 0xFFFFFFFFu; (void) memcpy(&value, sigMem, sizeof(uint32)); /* Memory was zero-initialised by AllocateMemory and never touched. */ ok = (value == 0u); } ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSConfig_BasicAccept() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44708u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } /* Build a CONFIG payload declaring one scalar "Counter" uint32 signal. */ if (ok) { uint8 payload[4u + UDPS_SIGNAL_DESC_SIZE + 1u]; uint32 numSigs = 1u; (void) memcpy(payload, &numSigs, 4u); UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Counter", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_UINT32; desc.quantType = UDPS_QUANT_NONE; desc.numDimensions = 0u; desc.numRows = 1u; desc.numCols = 1u; desc.rangeMin = 0.0; desc.rangeMax = 1.0; desc.timeMode = UDPS_TIMEMODE_PACKET; desc.samplingRate = 0.0; desc.timeSignalIdx = UDPS_NO_TIME_SIGNAL; (void) memcpy(payload + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); payload[4u + UDPS_SIGNAL_DESC_SIZE] = UDPS_PUBLISH_STRICT; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); } /* Deliver one DATA payload: 8-byte timestamp + uint32 value = 12345. * The head start given to the background Synchronise() in * StartSynchroniseThread() is a race against thread-scheduling latency, * so retry the deliver+sync sequence until it lands. */ bool gotValue = false; for (uint32 attempt = 0u; ok && (!gotValue) && (attempt < 30u); attempt++) { SynchroniseThreadArgs *syncArgs = StartSynchroniseThread(ds); uint8 data[8u + 4u]; (void) memset(data, 0, 8u); uint32 value = 12345u; (void) memcpy(data + 8u, &value, 4u); ds->OnUDPSData(data, static_cast(sizeof(data))); if (JoinSynchroniseThread(syncArgs)) { void *sigMem = NULL_PTR(void *); if (ds->GetSignalMemoryBuffer(0u, 0u, sigMem)) { uint32 decoded = 0u; (void) memcpy(&decoded, sigMem, sizeof(uint32)); gotValue = (decoded == 12345u); } } } ok = ok && gotValue; ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSConfig_SignalCountMismatch() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44709u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } /* CONFIG declares 2 signals while the client only has 1. */ if (ok) { uint8 payload[4u + (2u * UDPS_SIGNAL_DESC_SIZE) + 1u]; uint32 numSigs = 2u; (void) memcpy(payload, &numSigs, 4u); UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Counter", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_UINT32; desc.numRows = 1u; desc.numCols = 1u; (void) memcpy(payload + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); (void) strncpy(desc.name, "Extra", UDPS_MAX_SIGNAL_NAME - 1u); (void) memcpy(payload + 4u + UDPS_SIGNAL_DESC_SIZE, &desc, UDPS_SIGNAL_DESC_SIZE); payload[4u + (2u * UDPS_SIGNAL_DESC_SIZE)] = UDPS_PUBLISH_STRICT; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); } /* A DATA payload must be silently ignored since CONFIG was rejected. */ if (ok) { uint8 data[8u + 4u]; (void) memset(data, 0, 8u); uint32 value = 999u; (void) memcpy(data + 8u, &value, 4u); ds->OnUDPSData(data, static_cast(sizeof(data))); ok = ds->Synchronise(); } void *sigMem = NULL_PTR(void *); if (ok) { ok = ds->GetSignalMemoryBuffer(0u, 0u, sigMem); } if (ok) { uint32 value = 0xFFFFFFFFu; (void) memcpy(&value, sigMem, sizeof(uint32)); ok = (value == 0u); } ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSConfig_NameMismatch() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44710u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { uint8 payload[4u + UDPS_SIGNAL_DESC_SIZE + 1u]; uint32 numSigs = 1u; (void) memcpy(payload, &numSigs, 4u); UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "WrongName", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_UINT32; desc.numRows = 1u; desc.numCols = 1u; (void) memcpy(payload + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); payload[4u + UDPS_SIGNAL_DESC_SIZE] = UDPS_PUBLISH_STRICT; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); } if (ok) { uint8 data[8u + 4u]; (void) memset(data, 0, 8u); uint32 value = 999u; (void) memcpy(data + 8u, &value, 4u); ds->OnUDPSData(data, static_cast(sizeof(data))); ok = ds->Synchronise(); } void *sigMem = NULL_PTR(void *); if (ok) { ok = ds->GetSignalMemoryBuffer(0u, 0u, sigMem); } if (ok) { uint32 value = 0xFFFFFFFFu; (void) memcpy(&value, sigMem, sizeof(uint32)); ok = (value == 0u); } ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSConfig_ElementCountMismatch() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44711u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { uint8 payload[4u + UDPS_SIGNAL_DESC_SIZE + 1u]; uint32 numSigs = 1u; (void) memcpy(payload, &numSigs, 4u); UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Counter", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_UINT32; desc.numRows = 3u; /* client declares a scalar (1 element) */ desc.numCols = 1u; (void) memcpy(payload + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); payload[4u + UDPS_SIGNAL_DESC_SIZE] = UDPS_PUBLISH_STRICT; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); } if (ok) { uint8 data[8u + 4u]; (void) memset(data, 0, 8u); uint32 value = 999u; (void) memcpy(data + 8u, &value, 4u); ds->OnUDPSData(data, static_cast(sizeof(data))); ok = ds->Synchronise(); } void *sigMem = NULL_PTR(void *); if (ok) { ok = ds->GetSignalMemoryBuffer(0u, 0u, sigMem); } if (ok) { uint32 value = 0xFFFFFFFFu; (void) memcpy(&value, sigMem, sizeof(uint32)); ok = (value == 0u); } ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSConfig_TooSmallPayload() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44712u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { /* Fewer than 4 bytes: cannot even hold numSigs. Must not crash. */ uint8 payload[2u] = { 0u, 0u }; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); ok = ds->Synchronise(); } ObjectRegistryDatabase::Instance()->Purge(); return ok; } /** * @brief Config with two signals: scalar "Counter" (uint32) and a * two-element "Pair" (float64[2]) — used for ordering/accumulate tests. */ static MARTe::StreamString BuildTwoSignalConfig(MARTe::uint16 port) { using namespace MARTe; StreamString cfg; cfg.Printf( "+Test = {\n" " Class = RealTimeApplication\n" " +Functions = {\n" " Class = ReferenceContainer\n" " +Reader = {\n" " Class = UDPStreamerClientTestGAM\n" " InputSignals = {\n" " Counter = { DataSource = ClientDS Type = uint32 }\n" " Pair = { DataSource = ClientDS Type = float64 NumberOfDimensions = 1 NumberOfElements = 2 }\n" " }\n" " OutputSignals = {\n" " Counter = { DataSource = DDB Type = uint32 }\n" " Pair = { DataSource = DDB Type = float64 NumberOfDimensions = 1 NumberOfElements = 2 }\n" " }\n" " }\n" " }\n" " +Data = {\n" " Class = ReferenceContainer\n" " DefaultDataSource = DDB\n" " +DDB = { Class = GAMDataSource }\n" " +ClientDS = {\n" " Class = UDPStreamerClient\n" " ServerAddress = \"127.0.0.1\"\n" " Port = %d\n" " Signals = {\n" " Counter = { Type = uint32 }\n" " Pair = { Type = float64 NumberOfDimensions = 1 NumberOfElements = 2 }\n" " }\n" " }\n" " +Timings = { Class = TimingDataSource }\n" " }\n" " +States = {\n" " Class = ReferenceContainer\n" " +State1 = {\n" " Class = RealTimeState\n" " +Threads = {\n" " Class = ReferenceContainer\n" " +Thread1 = { Class = RealTimeThread Functions = { Reader } }\n" " }\n" " }\n" " }\n" " +Scheduler = { Class = GAMScheduler TimingDataSource = Timings }\n" "}\n", port); return cfg; } /** * @brief Config with a single Float32Bit scalar signal "Value" — used for * the UINT16 quantisation test. */ static MARTe::StreamString BuildFloat32Config(MARTe::uint16 port) { using namespace MARTe; StreamString cfg; cfg.Printf( "+Test = {\n" " Class = RealTimeApplication\n" " +Functions = {\n" " Class = ReferenceContainer\n" " +Reader = {\n" " Class = UDPStreamerClientTestGAM\n" " InputSignals = { Value = { DataSource = ClientDS Type = float32 } }\n" " OutputSignals = { Value = { DataSource = DDB Type = float32 } }\n" " }\n" " }\n" " +Data = {\n" " Class = ReferenceContainer\n" " DefaultDataSource = DDB\n" " +DDB = { Class = GAMDataSource }\n" " +ClientDS = {\n" " Class = UDPStreamerClient\n" " ServerAddress = \"127.0.0.1\"\n" " Port = %d\n" " Signals = { Value = { Type = float32 } }\n" " }\n" " +Timings = { Class = TimingDataSource }\n" " }\n" " +States = {\n" " Class = ReferenceContainer\n" " +State1 = {\n" " Class = RealTimeState\n" " +Threads = {\n" " Class = ReferenceContainer\n" " +Thread1 = { Class = RealTimeThread Functions = { Reader } }\n" " }\n" " }\n" " }\n" " +Scheduler = { Class = GAMScheduler TimingDataSource = Timings }\n" "}\n", port); return cfg; } /** * @brief Config with a single Float64Bit scalar signal "Value2" — used for * the INT8 quantisation test. */ static MARTe::StreamString BuildFloat64Config(MARTe::uint16 port) { using namespace MARTe; StreamString cfg; cfg.Printf( "+Test = {\n" " Class = RealTimeApplication\n" " +Functions = {\n" " Class = ReferenceContainer\n" " +Reader = {\n" " Class = UDPStreamerClientTestGAM\n" " InputSignals = { Value2 = { DataSource = ClientDS Type = float64 } }\n" " OutputSignals = { Value2 = { DataSource = DDB Type = float64 } }\n" " }\n" " }\n" " +Data = {\n" " Class = ReferenceContainer\n" " DefaultDataSource = DDB\n" " +DDB = { Class = GAMDataSource }\n" " +ClientDS = {\n" " Class = UDPStreamerClient\n" " ServerAddress = \"127.0.0.1\"\n" " Port = %d\n" " Signals = { Value2 = { Type = float64 } }\n" " }\n" " +Timings = { Class = TimingDataSource }\n" " }\n" " +States = {\n" " Class = ReferenceContainer\n" " +State1 = {\n" " Class = RealTimeState\n" " +Threads = {\n" " Class = ReferenceContainer\n" " +Thread1 = { Class = RealTimeThread Functions = { Reader } }\n" " }\n" " }\n" " }\n" " +Scheduler = { Class = GAMScheduler TimingDataSource = Timings }\n" "}\n", port); return cfg; } bool UDPStreamerClientTest::TestOnUDPSData_QuantizedUint16() { using namespace MARTe; StreamString cfg = BuildFloat32Config(44713u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { uint8 payload[4u + UDPS_SIGNAL_DESC_SIZE + 1u]; uint32 numSigs = 1u; (void) memcpy(payload, &numSigs, 4u); UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Value", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_FLOAT32; desc.quantType = UDPS_QUANT_UINT16; desc.numRows = 1u; desc.numCols = 1u; desc.rangeMin = -10.0; desc.rangeMax = 10.0; (void) memcpy(payload + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); payload[4u + UDPS_SIGNAL_DESC_SIZE] = UDPS_PUBLISH_STRICT; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); } /* raw=65535 (max uint16) -> rangeMin + (65535/65535)*range = rangeMax = 10.0 exactly. */ bool gotValue = false; for (uint32 attempt = 0u; ok && (!gotValue) && (attempt < 30u); attempt++) { SynchroniseThreadArgs *syncArgs = StartSynchroniseThread(ds); uint8 data[8u + 2u]; (void) memset(data, 0, 8u); uint16 raw = 65535u; (void) memcpy(data + 8u, &raw, 2u); ds->OnUDPSData(data, static_cast(sizeof(data))); if (JoinSynchroniseThread(syncArgs)) { void *sigMem = NULL_PTR(void *); if (ds->GetSignalMemoryBuffer(0u, 0u, sigMem)) { float32 decoded = 0.0f; (void) memcpy(&decoded, sigMem, sizeof(float32)); gotValue = (decoded == 10.0f); } } } ok = ok && gotValue; ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSData_QuantizedInt8() { using namespace MARTe; StreamString cfg = BuildFloat64Config(44714u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { uint8 payload[4u + UDPS_SIGNAL_DESC_SIZE + 1u]; uint32 numSigs = 1u; (void) memcpy(payload, &numSigs, 4u); UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Value2", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_FLOAT64; desc.quantType = UDPS_QUANT_INT8; desc.numRows = 1u; desc.numCols = 1u; desc.rangeMin = 0.0; desc.rangeMax = 100.0; (void) memcpy(payload + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); payload[4u + UDPS_SIGNAL_DESC_SIZE] = UDPS_PUBLISH_STRICT; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); } /* raw=127 (max int8) -> rangeMin + ((127+127)/254)*range = rangeMax = 100.0 exactly. */ bool gotValue = false; for (uint32 attempt = 0u; ok && (!gotValue) && (attempt < 30u); attempt++) { SynchroniseThreadArgs *syncArgs = StartSynchroniseThread(ds); uint8 data[8u + 1u]; (void) memset(data, 0, 8u); int8 raw = 127; (void) memcpy(data + 8u, &raw, 1u); ds->OnUDPSData(data, static_cast(sizeof(data))); if (JoinSynchroniseThread(syncArgs)) { void *sigMem = NULL_PTR(void *); if (ds->GetSignalMemoryBuffer(0u, 0u, sigMem)) { float64 decoded = 0.0; (void) memcpy(&decoded, sigMem, sizeof(float64)); gotValue = (decoded == 100.0); } } } ok = ok && gotValue; ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSData_MultipleSignalsOrder() { using namespace MARTe; StreamString cfg = BuildTwoSignalConfig(44715u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { uint8 payload[4u + (2u * UDPS_SIGNAL_DESC_SIZE) + 1u]; uint32 numSigs = 2u; (void) memcpy(payload, &numSigs, 4u); UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Counter", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_UINT32; desc.numRows = 1u; desc.numCols = 1u; (void) memcpy(payload + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Pair", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_FLOAT64; desc.numRows = 2u; desc.numCols = 1u; (void) memcpy(payload + 4u + UDPS_SIGNAL_DESC_SIZE, &desc, UDPS_SIGNAL_DESC_SIZE); payload[4u + (2u * UDPS_SIGNAL_DESC_SIZE)] = UDPS_PUBLISH_STRICT; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); } bool gotValue = false; for (uint32 attempt = 0u; ok && (!gotValue) && (attempt < 30u); attempt++) { SynchroniseThreadArgs *syncArgs = StartSynchroniseThread(ds); uint8 data[8u + 4u + 16u]; (void) memset(data, 0, sizeof(data)); uint32 counterVal = 777u; (void) memcpy(data + 8u, &counterVal, 4u); float64 pairVal[2] = { 1.5, 2.5 }; (void) memcpy(data + 12u, pairVal, 16u); ds->OnUDPSData(data, static_cast(sizeof(data))); if (JoinSynchroniseThread(syncArgs)) { void *counterMem = NULL_PTR(void *); void *pairMem = NULL_PTR(void *); if (ds->GetSignalMemoryBuffer(0u, 0u, counterMem) && ds->GetSignalMemoryBuffer(1u, 0u, pairMem)) { uint32 decodedCounter = 0u; (void) memcpy(&decodedCounter, counterMem, 4u); float64 decodedPair[2] = { 0.0, 0.0 }; (void) memcpy(decodedPair, pairMem, 16u); gotValue = (decodedCounter == 777u) && (decodedPair[0] == 1.5) && (decodedPair[1] == 2.5); } } } ok = ok && gotValue; ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSData_AccumulateMode() { using namespace MARTe; StreamString cfg = BuildTwoSignalConfig(44716u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { uint8 payload[4u + (2u * UDPS_SIGNAL_DESC_SIZE) + 1u]; uint32 numSigs = 2u; (void) memcpy(payload, &numSigs, 4u); UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Counter", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_UINT32; desc.numRows = 1u; desc.numCols = 1u; (void) memcpy(payload + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Pair", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_FLOAT64; desc.numRows = 2u; desc.numCols = 1u; (void) memcpy(payload + 4u + UDPS_SIGNAL_DESC_SIZE, &desc, UDPS_SIGNAL_DESC_SIZE); payload[4u + (2u * UDPS_SIGNAL_DESC_SIZE)] = UDPS_PUBLISH_ACCUMULATE; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); } /* [8-byte ts][4-byte numSamples=3][3x uint32 Counter samples][2x float64 Pair, once] */ bool gotValue = false; for (uint32 attempt = 0u; ok && (!gotValue) && (attempt < 30u); attempt++) { SynchroniseThreadArgs *syncArgs = StartSynchroniseThread(ds); uint8 data[8u + 4u + 12u + 16u]; (void) memset(data, 0, sizeof(data)); uint32 numSamples = 3u; (void) memcpy(data + 8u, &numSamples, 4u); uint32 samples[3] = { 10u, 20u, 30u }; (void) memcpy(data + 12u, samples, 12u); float64 pairVal[2] = { 1.5, 2.5 }; (void) memcpy(data + 24u, pairVal, 16u); ds->OnUDPSData(data, static_cast(sizeof(data))); if (JoinSynchroniseThread(syncArgs)) { void *counterMem = NULL_PTR(void *); void *pairMem = NULL_PTR(void *); if (ds->GetSignalMemoryBuffer(0u, 0u, counterMem) && ds->GetSignalMemoryBuffer(1u, 0u, pairMem)) { uint32 decodedCounter = 0u; (void) memcpy(&decodedCounter, counterMem, 4u); float64 decodedPair[2] = { 0.0, 0.0 }; (void) memcpy(decodedPair, pairMem, 16u); /* Scalar publishes only the most recent (last) accumulated sample. */ gotValue = (decodedCounter == 30u) && (decodedPair[0] == 1.5) && (decodedPair[1] == 2.5); } } } ok = ok && gotValue; ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSData_TooSmallPayload() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44717u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { uint8 payload[4u + UDPS_SIGNAL_DESC_SIZE + 1u]; uint32 numSigs = 1u; (void) memcpy(payload, &numSigs, 4u); UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Counter", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_UINT32; desc.numRows = 1u; desc.numCols = 1u; (void) memcpy(payload + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); payload[4u + UDPS_SIGNAL_DESC_SIZE] = UDPS_PUBLISH_STRICT; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); } if (ok) { /* Fewer than 8 bytes: cannot even hold the packet timestamp. */ uint8 data[4u] = { 1u, 2u, 3u, 4u }; ds->OnUDPSData(data, static_cast(sizeof(data))); ok = ds->Synchronise(); } void *sigMem = NULL_PTR(void *); if (ok) { ok = ds->GetSignalMemoryBuffer(0u, 0u, sigMem); } if (ok) { uint32 value = 0xFFFFFFFFu; (void) memcpy(&value, sigMem, sizeof(uint32)); ok = (value == 0u); } ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSData_BeforeConfig() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44718u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { /* No OnUDPSConfig was ever delivered: configValidated is false. */ uint8 data[8u + 4u]; (void) memset(data, 0, 8u); uint32 value = 555u; (void) memcpy(data + 8u, &value, 4u); ds->OnUDPSData(data, static_cast(sizeof(data))); ok = ds->Synchronise(); } void *sigMem = NULL_PTR(void *); if (ok) { ok = ds->GetSignalMemoryBuffer(0u, 0u, sigMem); } if (ok) { uint32 value = 0xFFFFFFFFu; (void) memcpy(&value, sigMem, sizeof(uint32)); ok = (value == 0u); } ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestOnUDPSDisconnected_InvalidatesConfig() { using namespace MARTe; StreamString cfg = BuildBasicConfig(44719u); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } if (ok) { uint8 payload[4u + UDPS_SIGNAL_DESC_SIZE + 1u]; uint32 numSigs = 1u; (void) memcpy(payload, &numSigs, 4u); UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Counter", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_UINT32; desc.numRows = 1u; desc.numCols = 1u; (void) memcpy(payload + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); payload[4u + UDPS_SIGNAL_DESC_SIZE] = UDPS_PUBLISH_STRICT; ds->OnUDPSConfig(payload, static_cast(sizeof(payload))); } void *sigMem = NULL_PTR(void *); bool gotValue = false; for (uint32 attempt = 0u; ok && (!gotValue) && (attempt < 30u); attempt++) { SynchroniseThreadArgs *syncArgs = StartSynchroniseThread(ds); uint8 data[8u + 4u]; (void) memset(data, 0, 8u); uint32 value = 111u; (void) memcpy(data + 8u, &value, 4u); ds->OnUDPSData(data, static_cast(sizeof(data))); if (JoinSynchroniseThread(syncArgs)) { if (ds->GetSignalMemoryBuffer(0u, 0u, sigMem)) { uint32 decoded = 0u; (void) memcpy(&decoded, sigMem, sizeof(uint32)); gotValue = (decoded == 111u); } } } ok = ok && gotValue; if (ok) { ds->OnUDPSDisconnected(); uint8 data[8u + 4u]; (void) memset(data, 0, 8u); uint32 value = 222u; (void) memcpy(data + 8u, &value, 4u); ds->OnUDPSData(data, static_cast(sizeof(data))); /* No semaphore post occurred (OnUDPSData returned early), so * Synchronise() must time out and leave the previous value in place. */ ok = ds->Synchronise(); } if (ok) { uint32 value = 0u; (void) memcpy(&value, sigMem, sizeof(uint32)); ok = (value == 111u); } ObjectRegistryDatabase::Instance()->Purge(); return ok; } bool UDPStreamerClientTest::TestExecute_ConnectConfigDataEndToEnd() { using namespace MARTe; static const uint16 serverPort = 44720u; StreamString cfg = BuildBasicConfig(serverPort); ReferenceT app = LoadApplication(cfg.Buffer()); bool ok = app.IsValid(); /* Bind the mock server socket BEFORE starting the receiver, so the * CONNECT datagram is never missed. */ BasicUDPSocket serverSock; if (ok) { ok = serverSock.Open() && serverSock.Listen(serverPort); } if (ok) { ok = (app->PrepareNextState("State1") == MARTe::ErrorManagement::NoError); } /* Receive the CONNECT and learn the client's ephemeral source address. */ if (ok) { uint8 recvBuf[64u]; uint32 recvSize = static_cast(sizeof(recvBuf)); TimeoutType timeout(1000u); bool received = serverSock.Read(reinterpret_cast(recvBuf), recvSize, timeout); ok = received && (recvSize >= UDPS_HEADER_SIZE); if (ok) { const UDPSPacketHeader *hdr = reinterpret_cast(recvBuf); ok = (hdr->magic == UDPS_MAGIC) && (hdr->type == UDPS_TYPE_CONNECT); } if (ok) { InternetHost clientHost = serverSock.GetSource(); serverSock.SetDestination(clientHost); } } /* Send CONFIG: one scalar "Counter" uint32 signal, unquantised. */ if (ok) { UDPSSignalDescriptor desc; (void) memset(&desc, 0, sizeof(desc)); (void) strncpy(desc.name, "Counter", UDPS_MAX_SIGNAL_NAME - 1u); desc.typeCode = UDPS_TYPECODE_UINT32; desc.numRows = 1u; desc.numCols = 1u; uint8 buf[UDPS_HEADER_SIZE + 4u + UDPS_SIGNAL_DESC_SIZE + 1u]; const uint32 configPayloadBytes = 4u + UDPS_SIGNAL_DESC_SIZE + 1u; UDPSBuildHeader(buf, UDPS_TYPE_CONFIG, 1u, 0u, 1u, configPayloadBytes); uint32 numSigs = 1u; (void) memcpy(buf + UDPS_HEADER_SIZE, &numSigs, 4u); (void) memcpy(buf + UDPS_HEADER_SIZE + 4u, &desc, UDPS_SIGNAL_DESC_SIZE); buf[UDPS_HEADER_SIZE + 4u + UDPS_SIGNAL_DESC_SIZE] = UDPS_PUBLISH_STRICT; uint32 sendSize = static_cast(sizeof(buf)); ok = serverSock.Write(reinterpret_cast(buf), sendSize); } Sleep::MSec(100u); /* Config processing does not Post() the data semaphore, so it is safe to * look the DataSource up and start Synchronise() on a background thread * now: by the time the DATA datagram is received (on the DataSource's * own receiver thread) and OnUDPSData() Post()s, this thread's * ResetWait() will already be blocked in Wait(), so the two race * correctly instead of the Post() being silently discarded. */ ReferenceT ds; if (ok) { ds = ObjectRegistryDatabase::Instance()->Find("Test.Data.ClientDS"); ok = ds.IsValid(); } /* Send DATA: 8-byte HRT timestamp + uint32 value = 424242. Each attempt * uses a fresh packet counter, since UDPSClient's fragment-reassembly * layer would otherwise silently drop an exact retransmission of a * counter it has already fully reassembled. */ bool gotValue = false; for (uint32 attempt = 0u; ok && (!gotValue) && (attempt < 30u); attempt++) { SynchroniseThreadArgs *syncArgs = StartSynchroniseThread(ds); uint8 buf[UDPS_HEADER_SIZE + 8u + 4u]; const uint32 dataPayloadBytes = 8u + 4u; UDPSBuildHeader(buf, UDPS_TYPE_DATA, 2u + attempt, 0u, 1u, dataPayloadBytes); (void) memset(buf + UDPS_HEADER_SIZE, 0, 8u); uint32 value = 424242u; (void) memcpy(buf + UDPS_HEADER_SIZE + 8u, &value, 4u); uint32 sendSize = static_cast(sizeof(buf)); ok = serverSock.Write(reinterpret_cast(buf), sendSize); if (ok && JoinSynchroniseThread(syncArgs)) { void *sigMem = NULL_PTR(void *); if (ds->GetSignalMemoryBuffer(0u, 0u, sigMem)) { uint32 decoded = 0u; (void) memcpy(&decoded, sigMem, sizeof(uint32)); gotValue = (decoded == 424242u); } } else if (!ok) { (void) JoinSynchroniseThread(syncArgs); } } ok = ok && gotValue; (void) serverSock.Close(); Sleep::MSec(50u); ObjectRegistryDatabase::Instance()->Purge(); return ok; }