martino/MARTe-Integrated-Components
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
MARTe-Integrated-Components/Test/Components/DataSources/UDPStreamerClient/UDPStreamerClientTest.cpp
T
Martino Ferrari 0bea41f866 Implemented better testing and fixed skipepd frames
2026-07-01 16:39:34 +02:00

1449 lines
53 KiB
C++
Raw Permalink Blame History

/**
* @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 <string.h>
/*---------------------------------------------------------------------------*/
/* 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<MARTe::UDPStreamerClient> ds;
bool result;
bool done;
};
static void SynchroniseThreadEntry(const void *const parameters) {
SynchroniseThreadArgs *args = const_cast<SynchroniseThreadArgs *>(
reinterpret_cast<const SynchroniseThreadArgs *>(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<MARTe::UDPStreamerClient> 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<MARTe::RealTimeApplication> 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<RealTimeApplication>();
}
ObjectRegistryDatabase *god = ObjectRegistryDatabase::Instance();
god->Purge();
if (!god->Initialise(cdb)) {
return ReferenceT<RealTimeApplication>();
}
ReferenceT<RealTimeApplication> app = god->Find("Test");
if (!app.IsValid()) {
return ReferenceT<RealTimeApplication>();
}
if (!app->ConfigureApplication()) {
return ReferenceT<RealTimeApplication>();
}
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<RealTimeApplication> 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<RealTimeApplication> 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<RealTimeApplication> 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<RealTimeApplication> 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<RealTimeApplication> 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<RealTimeApplication> app = LoadApplication(cfg);
bool ok = app.IsValid();
ObjectRegistryDatabase::Instance()->Purge();
return ok;
}
bool UDPStreamerClientTest::TestPrepareNextState_StartsReceiver() {
using namespace MARTe;
StreamString cfg = BuildBasicConfig(44706u);
ReferenceT<RealTimeApplication> 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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<RealTimeApplication> app = LoadApplication(cfg.Buffer());
bool ok = app.IsValid();
ReferenceT<UDPStreamerClient> 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<uint32>(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<uint32>(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<uint32>(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<RealTimeApplication> 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<uint32>(sizeof(recvBuf));
TimeoutType timeout(1000u);
bool received = serverSock.Read(reinterpret_cast<char8 *>(recvBuf), recvSize, timeout);
ok = received && (recvSize >= UDPS_HEADER_SIZE);
if (ok) {
const UDPSPacketHeader *hdr = reinterpret_cast<const UDPSPacketHeader *>(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<uint32>(sizeof(buf));
ok = serverSock.Write(reinterpret_cast<const char8 *>(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<UDPStreamerClient> 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<uint32>(sizeof(buf));
ok = serverSock.Write(reinterpret_cast<const char8 *>(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;
}
Reference in New Issue View Git Blame Copy Permalink