GithubStar/open62541
1
0
Fork 0
mirror of https://github.com/open62541/open62541.git synced 2026-10-28 20:59:28 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(pubsub): implement DataSetOrdering support and add related tests

Browse Source
This commit is contained in:
Andreas Ebner
2026-02-04 18:32:04 +01:00
committed by Julius Pfrommer
parent 938fcc5386
commit 23f5a80764
3 changed files with 183 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGES.md
+7
View File
@@ -3,6 +3,13 @@ refactorings and bug fixes are not reported here.
# Development
### PubSub DataSetOrdering Support (OPC UA Part 14)
Support for DataSetOrdering mechanism as defined in OPC UA Part 14, section
6.3.1.1.3 has been implemented for UADP encoding. The ordering of DataSetMessages
within NetworkMessages can be controlled via the `dataSetOrdering` field in the
`UA_UadpWriterGroupMessageDataType` configuration.
### Event API uses string-encoded of BrowsePaths
The select-clause of EventFilters defines the fields to be returned in
src/pubsub/ua_pubsub_writergroup.c
+54 -18
View File
@@ -930,6 +930,18 @@ UA_WriterGroup_publishCallback(UA_PubSubManager *psm, UA_WriterGroup *wg) {
return;
}
/* Extract DataSetOrdering from messageSettings */
UA_DataSetOrderingType dataSetOrdering = UA_DATASETORDERINGTYPE_UNDEFINED;
if(wg->config.messageSettings.encoding == UA_EXTENSIONOBJECT_DECODED ||
wg->config.messageSettings.encoding == UA_EXTENSIONOBJECT_DECODED_NODELETE) {
if(wg->config.messageSettings.content.decoded.type ==
&UA_TYPES[UA_TYPES_UADPWRITERGROUPMESSAGEDATATYPE]) {
UA_UadpWriterGroupMessageDataType *wgm =
(UA_UadpWriterGroupMessageDataType *)wg->config.messageSettings.content.decoded.data;
dataSetOrdering = wgm->dataSetOrdering;
}
}
/* How many DSM can be sent in one NM? */
UA_Byte maxDSM = (UA_Byte)wg->config.maxEncapsulatedDataSetMessageCount;
if(wg->config.maxEncapsulatedDataSetMessageCount > UA_BYTE_MAX)
@@ -937,23 +949,55 @@ UA_WriterGroup_publishCallback(UA_PubSubManager *psm, UA_WriterGroup *wg) {
if(maxDSM == 0)
maxDSM = 1; /* Send at least one dsm */
/* For AscendingWriterIdSingle, only one DataSetMessage per NetworkMessage */
if(dataSetOrdering == UA_DATASETORDERINGTYPE_ASCENDINGWRITERIDSINGLE)
maxDSM = 1;
/* Build array of enabled writers for ordering (DataSetOrdering, OPC UA Part 14 6.3.1.1.3) */
UA_STACKARRAY(UA_DataSetWriter*, writers, wg->writersCount);
size_t enabledWriters = 0;
UA_DataSetWriter *dsw;
LIST_FOREACH(dsw, &wg->writers, listEntry) {
if(dsw->head.state != UA_PUBSUBSTATE_OPERATIONAL)
continue;
writers[enabledWriters++] = dsw;
}
/* No enabled Writers */
if(enabledWriters == 0) {
UA_LOG_WARNING_PUBSUB(psm->logging, wg,
"Cannot publish -- No Writers are enabled");
unlockServer(psm->sc.server);
return;
}
/* Sort (insertion sort) writers by WriterId if ordering is AscendingWriterId
* or AscendingWriterIdSingle */
if(dataSetOrdering == UA_DATASETORDERINGTYPE_ASCENDINGWRITERID ||
dataSetOrdering == UA_DATASETORDERINGTYPE_ASCENDINGWRITERIDSINGLE) {
for(size_t i = 1; i < enabledWriters; i++) {
UA_DataSetWriter *key = writers[i];
UA_UInt16 keyWriterId = key->config.dataSetWriterId;
size_t j = i;
while(j > 0 && writers[j - 1]->config.dataSetWriterId > keyWriterId) {
writers[j] = writers[j - 1];
j--;
}
writers[j] = key;
}
}
/* It is possible to put several DataSetMessages into one NetworkMessage.
* But only if they do not contain promoted fields. NM with promoted fields
* are sent out right away. The others are kept in a buffer for
* "batching". */
size_t dsmCount = 0;
UA_STACKARRAY(UA_UInt16, dsWriterIds, wg->writersCount);
UA_STACKARRAY(UA_DataSetMessage, dsmStore, wg->writersCount);
UA_STACKARRAY(UA_UInt16, dsWriterIds, enabledWriters);
UA_STACKARRAY(UA_DataSetMessage, dsmStore, enabledWriters);
size_t enabledWriters = 0;
UA_DataSetWriter *dsw;
UA_EventLoop *el = psm->sc.server->config.eventLoop;
LIST_FOREACH(dsw, &wg->writers, listEntry) {
if(dsw->head.state != UA_PUBSUBSTATE_OPERATIONAL)
continue;
enabledWriters++;
for(size_t i = 0; i < enabledWriters; i++) {
dsw = writers[i];
/* PDS can be NULL -> Heartbeat */
UA_PublishedDataSet *pds = dsw->connectedDataSet;
@@ -982,14 +1026,6 @@ UA_WriterGroup_publishCallback(UA_PubSubManager *psm, UA_WriterGroup *wg) {
dsmCount++;
}
/* No enabled Writers */
if(enabledWriters == 0) {
UA_LOG_WARNING_PUBSUB(psm->logging, wg,
"Cannot publish -- No Writers are enabled");
unlockServer(psm->sc.server);
return;
}
/* Send the NetworkMessages with batched DataSetMessages */
UA_Byte nmDsmCount = 0;
for(size_t i = 0; i < dsmCount; i += nmDsmCount) {
tests/pubsub/check_pubsub_publish.c
+122
View File
@@ -557,6 +557,123 @@ START_TEST(PublishDataSetFieldAsDeltaFrame){
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
} END_TEST
/* Test DataSetOrdering reconfiguration (OPC UA Part 14, section 6.3.1.1.3)
*
* NOTE: This test validates that the DataSetOrdering mechanism is invoked correctly
* with different configurations. It does not decode NetworkMessage buffers to verify
* actual WriterId order in the wire format.
*/
START_TEST(DataSetOrderingReconfiguration) {
UA_NodeId writerGroup, pds1, pds2, pds3;
UA_StatusCode retVal = UA_STATUSCODE_GOOD;
/* Create PublishedDataSets */
UA_PublishedDataSetConfig pdsConfig;
memset(&pdsConfig, 0, sizeof(UA_PublishedDataSetConfig));
pdsConfig.publishedDataSetType = UA_PUBSUB_DATASET_PUBLISHEDITEMS;
pdsConfig.name = UA_STRING("OrderingDataSet1");
retVal |= UA_Server_addPublishedDataSet(server, &pdsConfig, &pds1).addResult;
pdsConfig.name = UA_STRING("OrderingDataSet2");
retVal |= UA_Server_addPublishedDataSet(server, &pdsConfig, &pds2).addResult;
pdsConfig.name = UA_STRING("OrderingDataSet3");
retVal |= UA_Server_addPublishedDataSet(server, &pdsConfig, &pds3).addResult;
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
/* Create WriterGroup with UNDEFINED ordering initially */
UA_WriterGroupConfig writerGroupConfig;
memset(&writerGroupConfig, 0, sizeof(writerGroupConfig));
writerGroupConfig.name = UA_STRING("OrderingWriterGroup");
writerGroupConfig.publishingInterval = 100;
writerGroupConfig.encodingMimeType = UA_PUBSUB_ENCODING_UADP;
UA_UadpWriterGroupMessageDataType *writerGroupMessage =
UA_UadpWriterGroupMessageDataType_new();
writerGroupMessage->networkMessageContentMask =
(UA_UadpNetworkMessageContentMask)(UA_UADPNETWORKMESSAGECONTENTMASK_PUBLISHERID |
UA_UADPNETWORKMESSAGECONTENTMASK_PAYLOADHEADER);
writerGroupMessage->dataSetOrdering = UA_DATASETORDERINGTYPE_UNDEFINED;
writerGroupConfig.messageSettings.content.decoded.data = writerGroupMessage;
writerGroupConfig.messageSettings.content.decoded.type =
&UA_TYPES[UA_TYPES_UADPWRITERGROUPMESSAGEDATATYPE];
writerGroupConfig.messageSettings.encoding = UA_EXTENSIONOBJECT_DECODED;
retVal = UA_Server_addWriterGroup(server, connection1, &writerGroupConfig, &writerGroup);
UA_UadpWriterGroupMessageDataType_delete(writerGroupMessage);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
/* Create DataSetWriters with mixed WriterIds: 300, 100, 200 */
UA_DataSetWriterConfig dataSetWriterConfig;
memset(&dataSetWriterConfig, 0, sizeof(dataSetWriterConfig));
dataSetWriterConfig.name = UA_STRING("Writer300");
dataSetWriterConfig.dataSetWriterId = 300;
retVal = UA_Server_addDataSetWriter(server, writerGroup, pds1, &dataSetWriterConfig, NULL);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
memset(&dataSetWriterConfig, 0, sizeof(dataSetWriterConfig));
dataSetWriterConfig.name = UA_STRING("Writer100");
dataSetWriterConfig.dataSetWriterId = 100;
retVal = UA_Server_addDataSetWriter(server, writerGroup, pds2, &dataSetWriterConfig, NULL);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
memset(&dataSetWriterConfig, 0, sizeof(dataSetWriterConfig));
dataSetWriterConfig.name = UA_STRING("Writer200");
dataSetWriterConfig.dataSetWriterId = 200;
retVal = UA_Server_addDataSetWriter(server, writerGroup, pds3, &dataSetWriterConfig, NULL);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
retVal = UA_Server_enableWriterGroup(server, writerGroup);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
/* Test 1: Publish with UNDEFINED ordering */
retVal = UA_Server_triggerWriterGroupPublish(server, writerGroup);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
/* Disable before reconfiguration */
retVal = UA_Server_setWriterGroupDisabled(server, writerGroup);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
/* Reconfigure to ASCENDINGWRITERID */
UA_WriterGroupConfig configCopy;
retVal = UA_Server_getWriterGroupConfig(server, writerGroup, &configCopy);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
UA_UadpWriterGroupMessageDataType *messageSettings =
(UA_UadpWriterGroupMessageDataType*)configCopy.messageSettings.content.decoded.data;
messageSettings->dataSetOrdering = UA_DATASETORDERINGTYPE_ASCENDINGWRITERID;
retVal = UA_Server_updateWriterGroupConfig(server, writerGroup, &configCopy);
UA_WriterGroupConfig_clear(&configCopy);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
/* Re-enable and test 2: Publish with ASCENDINGWRITERID ordering */
retVal = UA_Server_enableWriterGroup(server, writerGroup);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
retVal = UA_Server_triggerWriterGroupPublish(server, writerGroup);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
/* Disable before reconfiguration */
retVal = UA_Server_setWriterGroupDisabled(server, writerGroup);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
/* Reconfigure to ASCENDINGWRITERIDSINGLE */
retVal = UA_Server_getWriterGroupConfig(server, writerGroup, &configCopy);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
messageSettings = (UA_UadpWriterGroupMessageDataType*)configCopy.messageSettings.content.decoded.data;
messageSettings->dataSetOrdering = UA_DATASETORDERINGTYPE_ASCENDINGWRITERIDSINGLE;
retVal = UA_Server_updateWriterGroupConfig(server, writerGroup, &configCopy);
UA_WriterGroupConfig_clear(&configCopy);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
/* Re-enable and test 3: Publish with ASCENDINGWRITERIDSINGLE ordering */
retVal = UA_Server_enableWriterGroup(server, writerGroup);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
retVal = UA_Server_triggerWriterGroupPublish(server, writerGroup);
ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
} END_TEST
int main(void) {
TCase *tc_add_pubsub_writergroup = tcase_create("PubSub WriterGroup items handling");
tcase_add_checked_fixture(tc_add_pubsub_writergroup, setup, teardown);
@@ -590,11 +707,16 @@ int main(void) {
tcase_add_test(tc_pubsub_publish, SinglePublishDataSetFieldAndPublishTimestampTest);
tcase_add_test(tc_pubsub_publish, PublishDataSetFieldAsDeltaFrame);
TCase *tc_pubsub_datasetordering = tcase_create("PubSub DataSetOrdering (OPC UA Part 14)");
tcase_add_checked_fixture(tc_pubsub_datasetordering, setup, teardown);
tcase_add_test(tc_pubsub_datasetordering, DataSetOrderingReconfiguration);
Suite *s = suite_create("PubSub WriterGroups/Writer/Fields handling and publishing");
suite_add_tcase(s, tc_add_pubsub_writergroup);
suite_add_tcase(s, tc_add_pubsub_datasetwriter);
suite_add_tcase(s, tc_add_pubsub_datasetfields);
suite_add_tcase(s, tc_pubsub_publish);
suite_add_tcase(s, tc_pubsub_datasetordering);
SRunner *sr = srunner_create(s);
srunner_set_fork_status(sr, CK_NOFORK);