diff --git a/CHANGES.md b/CHANGES.md index 588610393..9d3f3a88d 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -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 diff --git a/src/pubsub/ua_pubsub_writergroup.c b/src/pubsub/ua_pubsub_writergroup.c index bed30015c..313c31a28 100644 --- a/src/pubsub/ua_pubsub_writergroup.c +++ b/src/pubsub/ua_pubsub_writergroup.c @@ -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) { diff --git a/tests/pubsub/check_pubsub_publish.c b/tests/pubsub/check_pubsub_publish.c index 51f90bdb4..2418efbab 100644 --- a/tests/pubsub/check_pubsub_publish.c +++ b/tests/pubsub/check_pubsub_publish.c @@ -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);