123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335 |
- /* This Source Code Form is subject to the terms of the Mozilla Public
- * License, v. 2.0. If a copy of the MPL was not distributed with this
- * file, You can obtain one at http://mozilla.org/MPL/2.0/.
- *
- * Copyright (c) 2017-2019 Fraunhofer IOSB (Author: Andreas Ebner)
- * Copyright (c) 2018 Fraunhofer IOSB (Author: Julius Pfrommer)
- */
- #include "server/ua_server_internal.h"
- #include "ua_pubsub_ns0.h"
- #ifdef UA_ENABLE_PUBSUB /* conditional compilation */
- #define UA_DATETIMESTAMP_2000 125911584000000000
- UA_StatusCode
- UA_Server_addPubSubConnection(UA_Server *server,
- const UA_PubSubConnectionConfig *connectionConfig,
- UA_NodeId *connectionIdentifier) {
- /* Find the matching UA_PubSubTransportLayers */
- UA_PubSubTransportLayer *tl = NULL;
- for(size_t i = 0; i < server->config.pubsubTransportLayersSize; i++) {
- if(connectionConfig &&
- UA_String_equal(&server->config.pubsubTransportLayers[i].transportProfileUri,
- &connectionConfig->transportProfileUri)) {
- tl = &server->config.pubsubTransportLayers[i];
- }
- }
- if(!tl) {
- UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PubSub Connection creation failed. Requested transport layer not found.");
- return UA_STATUSCODE_BADNOTFOUND;
- }
- /* Create a copy of the connection config */
- UA_PubSubConnectionConfig *tmpConnectionConfig = (UA_PubSubConnectionConfig *)
- UA_calloc(1, sizeof(UA_PubSubConnectionConfig));
- if(!tmpConnectionConfig){
- UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PubSub Connection creation failed. Out of Memory.");
- return UA_STATUSCODE_BADOUTOFMEMORY;
- }
- UA_StatusCode retval = UA_PubSubConnectionConfig_copy(connectionConfig, tmpConnectionConfig);
- if(retval != UA_STATUSCODE_GOOD){
- UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PubSub Connection creation failed. Could not copy the config.");
- return retval;
- }
- /* Create new connection and add to UA_PubSubManager */
- UA_PubSubConnection *newConnectionsField = (UA_PubSubConnection *)
- UA_calloc(1, sizeof(UA_PubSubConnection));
- if(!newConnectionsField) {
- UA_PubSubConnectionConfig_clear(tmpConnectionConfig);
- UA_free(tmpConnectionConfig);
- UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PubSub Connection creation failed. Out of Memory.");
- return UA_STATUSCODE_BADOUTOFMEMORY;
- }
- if (server->pubSubManager.connectionsSize != 0)
- TAILQ_INSERT_TAIL(&server->pubSubManager.connections, newConnectionsField, listEntry);
- else {
- TAILQ_INIT(&server->pubSubManager.connections);
- TAILQ_INSERT_HEAD(&server->pubSubManager.connections, newConnectionsField, listEntry);
- }
- server->pubSubManager.connectionsSize++;
- LIST_INIT(&newConnectionsField->writerGroups);
- newConnectionsField->config = tmpConnectionConfig;
- /* Open the channel */
- newConnectionsField->channel = tl->createPubSubChannel(newConnectionsField->config);
- if(!newConnectionsField->channel) {
- UA_PubSubConnection_clear(server, newConnectionsField);
- TAILQ_REMOVE(&server->pubSubManager.connections, newConnectionsField, listEntry);
- server->pubSubManager.connectionsSize--;
- UA_free(newConnectionsField);
- UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PubSub Connection creation failed. Transport layer creation problem.");
- return UA_STATUSCODE_BADINTERNALERROR;
- }
- UA_PubSubManager_generateUniqueNodeId(server, &newConnectionsField->identifier);
- if(connectionIdentifier)
- UA_NodeId_copy(&newConnectionsField->identifier, connectionIdentifier);
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- addPubSubConnectionRepresentation(server, newConnectionsField);
- #endif
- return UA_STATUSCODE_GOOD;
- }
- UA_StatusCode
- UA_Server_removePubSubConnection(UA_Server *server, const UA_NodeId connection) {
- //search the identified Connection and store the Connection index
- UA_PubSubConnection *currentConnection = UA_PubSubConnection_findConnectionbyId(server, connection);
- if(!currentConnection)
- return UA_STATUSCODE_BADNOTFOUND;
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- removePubSubConnectionRepresentation(server, currentConnection);
- #endif
- server->pubSubManager.connectionsSize--;
- UA_PubSubConnection_clear(server, currentConnection);
- TAILQ_REMOVE(&server->pubSubManager.connections, currentConnection, listEntry);
- UA_free(currentConnection);
- return UA_STATUSCODE_GOOD;
- }
- UA_AddPublishedDataSetResult
- UA_Server_addPublishedDataSet(UA_Server *server, const UA_PublishedDataSetConfig *publishedDataSetConfig,
- UA_NodeId *pdsIdentifier) {
- UA_AddPublishedDataSetResult result = {UA_STATUSCODE_BADINVALIDARGUMENT, 0, NULL, {0, 0}};
- if(!publishedDataSetConfig){
- UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PublishedDataSet creation failed. No config passed in.");
- return result;
- }
- if(publishedDataSetConfig->publishedDataSetType != UA_PUBSUB_DATASET_PUBLISHEDITEMS){
- UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PublishedDataSet creation failed. Unsupported PublishedDataSet type.");
- return result;
- }
- //deep copy the given connection config
- UA_PublishedDataSetConfig tmpPublishedDataSetConfig;
- memset(&tmpPublishedDataSetConfig, 0, sizeof(UA_PublishedDataSetConfig));
- if(UA_PublishedDataSetConfig_copy(publishedDataSetConfig, &tmpPublishedDataSetConfig) != UA_STATUSCODE_GOOD){
- UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PublishedDataSet creation failed. Configuration copy failed.");
- result.addResult = UA_STATUSCODE_BADINTERNALERROR;
- return result;
- }
- //create new PDS and add to UA_PubSubManager
- UA_PublishedDataSet *newPubSubDataSetField = (UA_PublishedDataSet *)
- UA_calloc(1, sizeof(UA_PublishedDataSet));
- if(!newPubSubDataSetField) {
- UA_PublishedDataSetConfig_clear(&tmpPublishedDataSetConfig);
- UA_LOG_ERROR(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "PublishedDataSet creation failed. Out of Memory.");
- result.addResult = UA_STATUSCODE_BADOUTOFMEMORY;
- return result;
- }
- memset(newPubSubDataSetField, 0, sizeof(UA_PublishedDataSet));
- TAILQ_INIT(&newPubSubDataSetField->fields);
- newPubSubDataSetField->config = tmpPublishedDataSetConfig;
- if (server->pubSubManager.publishedDataSetsSize != 0)
- TAILQ_INSERT_TAIL(&server->pubSubManager.publishedDataSets, newPubSubDataSetField, listEntry);
- else {
- TAILQ_INIT(&server->pubSubManager.publishedDataSets);
- TAILQ_INSERT_HEAD(&server->pubSubManager.publishedDataSets, newPubSubDataSetField, listEntry);
- }
- if(tmpPublishedDataSetConfig.publishedDataSetType == UA_PUBSUB_DATASET_PUBLISHEDITEMS_TEMPLATE){
- //parse template config and add fields (later PubSub batch)
- }
- //generate unique nodeId
- UA_PubSubManager_generateUniqueNodeId(server, &newPubSubDataSetField->identifier);
- if(pdsIdentifier != NULL){
- UA_NodeId_copy(&newPubSubDataSetField->identifier, pdsIdentifier);
- }
- result.addResult = UA_STATUSCODE_GOOD;
- result.fieldAddResults = NULL;
- result.fieldAddResultsSize = 0;
- //fill the DataSetMetaData
- switch(tmpPublishedDataSetConfig.publishedDataSetType){
- case UA_PUBSUB_DATASET_PUBLISHEDITEMS_TEMPLATE:
- if(UA_DataSetMetaDataType_copy(&tmpPublishedDataSetConfig.config.itemsTemplate.metaData,
- &newPubSubDataSetField->dataSetMetaData) != UA_STATUSCODE_GOOD){
- UA_Server_removeDataSetField(server, newPubSubDataSetField->identifier);
- result.addResult = UA_STATUSCODE_BADINTERNALERROR;
- }
- break;
- case UA_PUBSUB_DATASET_PUBLISHEDEVENTS_TEMPLATE:
- if(UA_DataSetMetaDataType_copy(&tmpPublishedDataSetConfig.config.eventTemplate.metaData,
- &newPubSubDataSetField->dataSetMetaData) != UA_STATUSCODE_GOOD){
- UA_Server_removeDataSetField(server, newPubSubDataSetField->identifier);
- result.addResult = UA_STATUSCODE_BADINTERNALERROR;
- }
- break;
- case UA_PUBSUB_DATASET_PUBLISHEDEVENTS:
- newPubSubDataSetField->dataSetMetaData.configurationVersion.majorVersion = UA_PubSubConfigurationVersionTimeDifference();
- newPubSubDataSetField->dataSetMetaData.configurationVersion.minorVersion = UA_PubSubConfigurationVersionTimeDifference();
- newPubSubDataSetField->dataSetMetaData.dataSetClassId = UA_GUID_NULL;
- if(UA_String_copy(&tmpPublishedDataSetConfig.name, &newPubSubDataSetField->dataSetMetaData.name) != UA_STATUSCODE_GOOD){
- UA_Server_removeDataSetField(server, newPubSubDataSetField->identifier);
- result.addResult = UA_STATUSCODE_BADINTERNALERROR;
- }
- newPubSubDataSetField->dataSetMetaData.description = UA_LOCALIZEDTEXT_ALLOC("", "");
- break;
- case UA_PUBSUB_DATASET_PUBLISHEDITEMS:
- newPubSubDataSetField->dataSetMetaData.configurationVersion.majorVersion = UA_PubSubConfigurationVersionTimeDifference();
- newPubSubDataSetField->dataSetMetaData.configurationVersion.minorVersion = UA_PubSubConfigurationVersionTimeDifference();
- if(UA_String_copy(&tmpPublishedDataSetConfig.name, &newPubSubDataSetField->dataSetMetaData.name) != UA_STATUSCODE_GOOD){
- UA_Server_removeDataSetField(server, newPubSubDataSetField->identifier);
- result.addResult = UA_STATUSCODE_BADINTERNALERROR;
- }
- newPubSubDataSetField->dataSetMetaData.description = UA_LOCALIZEDTEXT_ALLOC("", "");
- newPubSubDataSetField->dataSetMetaData.dataSetClassId = UA_GUID_NULL;
- break;
- }
- server->pubSubManager.publishedDataSetsSize++;
- result.configurationVersion.majorVersion = UA_PubSubConfigurationVersionTimeDifference();
- result.configurationVersion.minorVersion = UA_PubSubConfigurationVersionTimeDifference();
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- addPublishedDataItemsRepresentation(server, newPubSubDataSetField);
- #endif
- return result;
- }
- UA_StatusCode
- UA_Server_removePublishedDataSet(UA_Server *server, const UA_NodeId pds) {
- //search the identified PublishedDataSet and store the PDS index
- UA_PublishedDataSet *publishedDataSet = UA_PublishedDataSet_findPDSbyId(server, pds);
- if(!publishedDataSet){
- return UA_STATUSCODE_BADNOTFOUND;
- }
- if(publishedDataSet->config.configurationFrozen){
- UA_LOG_WARNING(&server->config.logger, UA_LOGCATEGORY_SERVER,
- "Remove PublishedDataSet failed. PublishedDataSet is frozen.");
- return UA_STATUSCODE_BADCONFIGURATIONERROR;
- }
- //search for referenced writers -> delete this writers. (Standard: writer must be connected with PDS)
- UA_PubSubConnection *tmpConnectoin;
- TAILQ_FOREACH(tmpConnectoin, &server->pubSubManager.connections, listEntry){
- UA_WriterGroup *writerGroup;
- LIST_FOREACH(writerGroup, &tmpConnectoin->writerGroups, listEntry){
- UA_DataSetWriter *currentWriter, *tmpWriterGroup;
- LIST_FOREACH_SAFE(currentWriter, &writerGroup->writers, listEntry, tmpWriterGroup){
- if(UA_NodeId_equal(¤tWriter->connectedDataSet, &publishedDataSet->identifier)){
- UA_Server_removeDataSetWriter(server, currentWriter->identifier);
- }
- }
- }
- }
- #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
- removePublishedDataSetRepresentation(server, publishedDataSet);
- #endif
- UA_PublishedDataSet_clear(server, publishedDataSet);
- server->pubSubManager.publishedDataSetsSize--;
- TAILQ_REMOVE(&server->pubSubManager.publishedDataSets, publishedDataSet, listEntry);
- UA_free(publishedDataSet);
- return UA_STATUSCODE_GOOD;
- }
- /* Calculate the time difference between current time and UTC (00:00) on January
- * 1, 2000. */
- UA_UInt32
- UA_PubSubConfigurationVersionTimeDifference() {
- UA_UInt32 timeDiffSince2000 = (UA_UInt32) (UA_DateTime_now() - UA_DATETIMESTAMP_2000);
- return timeDiffSince2000;
- }
- /* Generate a new unique NodeId. This NodeId will be used for the information
- * model representation of PubSub entities. */
- void
- UA_PubSubManager_generateUniqueNodeId(UA_Server *server, UA_NodeId *nodeId) {
- UA_NodeId newNodeId = UA_NODEID_NUMERIC(0, 0);
- UA_Node *newNode = UA_NODESTORE_NEW(server, UA_NODECLASS_OBJECT);
- UA_NODESTORE_INSERT(server, newNode, &newNodeId);
- UA_NodeId_copy(&newNodeId, nodeId);
- }
- /* Delete the current PubSub configuration including all nested members. This
- * action also delete the configured PubSub transport Layers. */
- void
- UA_PubSubManager_delete(UA_Server *server, UA_PubSubManager *pubSubManager) {
- UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_SERVER, "PubSub cleanup was called.");
- /* Stop and unfreeze all WriterGroups */
- UA_PubSubConnection *tmpConnection;
- TAILQ_FOREACH(tmpConnection, &server->pubSubManager.connections, listEntry){
- for(size_t i = 0; i < pubSubManager->connectionsSize; i++) {
- UA_WriterGroup *writerGroup;
- LIST_FOREACH(writerGroup, &tmpConnection->writerGroups, listEntry) {
- UA_WriterGroup_setPubSubState(server, UA_PUBSUBSTATE_DISABLED, writerGroup);
- UA_Server_unfreezeWriterGroupConfiguration(server, writerGroup->identifier);
- }
- }
- }
- //free the currently configured transport layers
- UA_free(server->config.pubsubTransportLayers);
- server->config.pubsubTransportLayersSize = 0;
- //remove Connections and WriterGroups
- UA_PubSubConnection *tmpConnection1, *tmpConnection2;
- TAILQ_FOREACH_SAFE(tmpConnection1, &server->pubSubManager.connections, listEntry, tmpConnection2){
- UA_Server_removePubSubConnection(server, tmpConnection1->identifier);
- }
- UA_PublishedDataSet *tmpPDS1, *tmpPDS2;
- TAILQ_FOREACH_SAFE(tmpPDS1, &server->pubSubManager.publishedDataSets, listEntry, tmpPDS2){
- UA_Server_removePublishedDataSet(server, tmpPDS1->identifier);
- }
- }
- /***********************************/
- /* PubSub Jobs abstraction */
- /***********************************/
- #ifndef UA_ENABLE_PUBSUB_CUSTOM_PUBLISH_HANDLING
- /* If UA_ENABLE_PUBSUB_CUSTOM_PUBLISH_INTERRUPT is enabled, a custom callback
- * management must be linked to the application */
- UA_StatusCode
- UA_PubSubManager_addRepeatedCallback(UA_Server *server, UA_ServerCallback callback,
- void *data, UA_Double interval_ms, UA_UInt64 *callbackId) {
- return UA_Timer_addRepeatedCallback(&server->timer, (UA_ApplicationCallback)callback,
- server, data, interval_ms, callbackId);
- }
- UA_StatusCode
- UA_PubSubManager_changeRepeatedCallbackInterval(UA_Server *server, UA_UInt64 callbackId,
- UA_Double interval_ms) {
- return UA_Timer_changeRepeatedCallbackInterval(&server->timer, callbackId, interval_ms);
- }
- void
- UA_PubSubManager_removeRepeatedPubSubCallback(UA_Server *server, UA_UInt64 callbackId) {
- UA_Timer_removeCallback(&server->timer, callbackId);
- }
- #endif /* UA_ENABLE_PUBSUB_CUSTOM_PUBLISH_HANDLING */
- #endif /* UA_ENABLE_PUBSUB */
|