Add pubsub PublishedDataSet handling

documentation fixes, code project adjustments

removed hard dependency between main lib and log plugin

pubsub API delte PubSub prefix

examples/CMakeLists.txt

@@ -126,4 +126,5 @@ add_subdirectory(nodeset)
 
 if(UA_ENABLE_PUBSUB)
     add_example(tutorial_pubsub_connection pubsub/tutorial_pubsub_connection.c)
+    add_example(tutorial_pubsub_publish pubsub/tutorial_pubsub_publish.c)
 endif()

examples/pubsub/tutorial_pubsub_publish.c

@@ -0,0 +1,68 @@
+/* This work is licensed under a Creative Commons CCZero 1.0 Universal License.
+ * See http://creativecommons.org/publicdomain/zero/1.0/ for more information. */
+
+#include <signal.h>
+#include "open62541.h"
+
+/* Work in progress: This Tutorial/Example will be continuously extended during the next PubSub batches */
+
+UA_Boolean running = true;
+static void stopHandler(int sign) {
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "received ctrl-c");
+    running = false;
+}
+
+static UA_StatusCode
+addPubSubConnection(UA_Server *server){
+    /* Details about the connection configuration and handling are located in the pubsub connection tutorial */
+    UA_PubSubConnectionConfig connectionConfig;
+    memset(&connectionConfig, 0, sizeof(connectionConfig));
+    connectionConfig.name = UA_STRING("UDP-UADP Connection 1");
+    connectionConfig.transportProfileUri = UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-udp-uadp");
+    connectionConfig.enabled = UA_TRUE;
+    UA_NetworkAddressUrlDataType networkAddressUrl = {UA_STRING_NULL , UA_STRING("opc.udp://224.0.0.22:4840/")};
+    UA_Variant_setScalar(&connectionConfig.address, &networkAddressUrl, &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]);
+    connectionConfig.publisherId.numeric = UA_UInt32_random();
+    UA_NodeId connectionIdentifier;
+    UA_StatusCode retval = UA_Server_addPubSubConnection(server, &connectionConfig, &connectionIdentifier);
+    return retval;
+}
+
+/**
+ * The PubSub publish example demonstrate the simplest way to publish
+ * informations from the information model over UDP Multicast.
+ */
+int main(void) {
+    signal(SIGINT, stopHandler);
+    signal(SIGTERM, stopHandler);
+
+    UA_StatusCode retval = UA_STATUSCODE_GOOD;
+    UA_ServerConfig *config = UA_ServerConfig_new_default();
+    /* Details about the connection configuration and handling are located in the pubsub connection tutorial */
+    config->pubsubTransportLayers = (UA_PubSubTransportLayer *) UA_malloc(sizeof(UA_PubSubTransportLayer));
+    if(!config->pubsubTransportLayers) {
+        UA_ServerConfig_delete(config);
+        return -1;
+    }
+    config->pubsubTransportLayers[0] = UA_PubSubTransportLayerUDPMP();
+    config->pubsubTransportLayersSize++;
+    UA_Server *server = UA_Server_new(config);
+
+    if(addPubSubConnection(server) != UA_STATUSCODE_GOOD)
+        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "PubSub Connection creation failed!");
+
+    /* The PublishedDataSetConfig contains all necessary public informations for the creation of a new PublishedDataSet */
+    UA_PublishedDataSetConfig publishedDataSetConfig;
+    publishedDataSetConfig.publishedDataSetType = UA_PUBSUB_DATASET_PUBLISHEDITEMS;
+    publishedDataSetConfig.name = UA_STRING("Robot Axis");
+
+    /* Create new PublishedDataSet based on the PublishedDataSetConfig. */
+    UA_NodeId publishedDataSetIdent;
+    if(UA_Server_addPublishedDataSet(server, &publishedDataSetConfig, &publishedDataSetIdent).addResult == UA_STATUSCODE_GOOD)
+        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "PubSub PublishedDataSet creation successful!");
+
+    retval |= UA_Server_run(server, &running);
+    UA_Server_delete(server);
+    UA_ServerConfig_delete(config);
+    return (int)retval;
+}

include/ua_server_pubsub.h

@@ -37,6 +37,36 @@ extern "C" {
  * 
  * 3. The add[element] function returns the unique nodeId of the internally created element.
  *
+ * Take a look on the PubSub Tutorials for mor details about the API usage.::
+ *
+ *  +-----------+
+ *  | UA_Server |
+ *  +-----------+
+ *   |    |
+ *   |    |
+ *   |    |
+ *   |    |  +----------------------+
+ *   |    +->| UA_PubSubConnections |  UA_Server_addPubSubConnection
+ *   |       +----------------------+
+ *   |        |    |
+ *   |        |    |    +-----------------------+
+ *   |        |    +--->| UA_PubSubWriterGroups |
+ *   |        |         +-----------------------+
+ *   |        |
+ *   |        |         +-----------------------+
+ *   |        +-------->| UA_PubSubReaderGroups |
+ *   |                  +-----------------------+
+ *   |
+ *   |       +---------------------------+
+ *   +------>| UA_PubSubPublishedDataSet |  UA_Server_addPublishedDataSet
+ *           +---------------------------+
+ *
+ * Connections
+ * -----------
+ * The PubSub connections are the abstraction between the concrete transport protocol
+ * and the PubSub functionality. It is possible to create multiple connections with
+ * different transport protocols at runtime.
+ *
  * Take a look on the PubSub Tutorials for mor details about the API usage.
  * Connections
  * -----------
@@ -56,20 +86,100 @@ typedef struct{
     UA_Variant connectionTransportSettings;
 } UA_PubSubConnectionConfig;
 
+/**
+ * Connection handling
+ * ^^^^^^^^^^^^^^^^^^^
+ */
+
+UA_StatusCode
+UA_Server_addPubSubConnection(UA_Server *server, const UA_PubSubConnectionConfig *connectionConfig,
+                              UA_NodeId *connectionIdentifier);
+
+UA_StatusCode
+UA_Server_removePubSubConnection(UA_Server *server, UA_NodeId connectionIdentifier);
+
 UA_StatusCode
 UA_PubSubConnection_getConfig(UA_Server *server, UA_NodeId connectionIdentifier,
                               UA_PubSubConnectionConfig *config);
 
 /**
- * **Create and Remove Connections**
+ * PublishedDataSets
+ * -----------------
+ * The PublishedDataSets (PDS) are containers for the published information. The PDS contain
+ * the published variables and meta informations. The metainformations are commonly autogenerated
+ * or given as constant argument as part of the template functions. The template functions are standard
+ * defined and should only be used for configuration tools. You should normally
+ * create a empty PDS and call the functions to add new fields.
+ */
+
+typedef struct {
+    UA_StatusCode addResult;
+    size_t fieldAddResultsSize;
+    UA_StatusCode *fieldAddResults;
+    UA_ConfigurationVersionDataType configurationVersion;
+} UA_AddPublishedDataSetResult;
+
+typedef enum {
+    UA_PUBSUB_DATASET_PUBLISHEDITEMS,
+    UA_PUBSUB_DATASET_PUBLISHEDEVENTS,
+    UA_PUBSUB_DATASET_PUBLISHEDITEMS_TEMPLATE,
+    UA_PUBSUB_DATASET_PUBLISHEDEVENTS_TEMPLATE,
+} UA_PublishedDataSetType;
+
+/* The UA_PUBSUB_DATASET_PUBLISHEDITEMS has currently no additional members
+ * and thus no dedicated config structure.
+ */
+
+typedef struct {
+    UA_DataSetMetaDataType metaData;
+    size_t variablesToAddSize;
+    UA_PublishedVariableDataType *variablesToAdd;
+} UA_PublishedDataItemsTemplateConfig;
+
+typedef struct {
+    UA_NodeId eventNotfier;
+    UA_ContentFilter filter;
+} UA_PublishedEventConfig;
+
+typedef struct {
+    UA_DataSetMetaDataType metaData;
+    UA_NodeId eventNotfier;
+    size_t selectedFieldsSize;
+    UA_SimpleAttributeOperand *selectedFields;
+    UA_ContentFilter filter;
+} UA_PublishedEventTemplateConfig;
+
+/* Configuration structure for PubSubDataSet */
+typedef struct {
+    UA_String name;
+    UA_PublishedDataSetType publishedDataSetType;
+    union{
+        /* The UA_PUBSUB_DATASET_PUBLISHEDITEMS has currently no additional members
+         * and thus no dedicated config structure.*/
+        UA_PublishedDataItemsTemplateConfig itemsTemplate;
+        UA_PublishedEventConfig event;
+        UA_PublishedEventTemplateConfig eventTemplate;
+    } config;
+} UA_PublishedDataSetConfig;
+
+/**
+ * PublishedDataSet handling
+ * ^^^^^^^^^^^^^^^^^^^^^^^^^
  */
 
+UA_AddPublishedDataSetResult
+UA_Server_addPublishedDataSet(UA_Server *server, const UA_PublishedDataSetConfig *publishedDataSetConfig,
+                              UA_NodeId *pdsIdentifier);
+
 UA_StatusCode
-UA_Server_addPubSubConnection(UA_Server *server, const UA_PubSubConnectionConfig *connectionConfig,
-                              UA_NodeId *connectionIdentifier);
+UA_Server_removePublishedDataSet(UA_Server *server, UA_NodeId pdsIdentifier);
 
 UA_StatusCode
-UA_Server_removePubSubConnection(UA_Server *server, UA_NodeId connectionIdentifier);
+UA_PublishedDataSet_getConfig(UA_Server *server, UA_NodeId publishedDataSetIdentifier,
+                              UA_PublishedDataSetConfig *config);
+void
+UA_PublishedDataSetConfig_deleteMembers(UA_PublishedDataSetConfig *pdsConfig);
+
 
 #ifdef __cplusplus
 } // extern "C"

src/pubsub/ua_pubsub.c

@@ -5,6 +5,8 @@
  * Copyright (c) 2017-2018 Fraunhofer IOSB (Author: Andreas Ebner)
  */
 
+#include "ua_server_pubsub.h"
+#include "server/ua_server_internal.h"
 #include "ua_pubsub.h"
 #include "ua_pubsub_manager.h"
 
@@ -35,9 +37,6 @@ UA_PubSubConnectionConfig_copy(const UA_PubSubConnectionConfig *src, UA_PubSubCo
 /**
  * Get the current config of the Connection.
  *
- * @param server
- * @param connectionIdentifier
- * @param config
  * @return UA_STATUSCODE_GOOD on success
  */
 UA_StatusCode
@@ -46,7 +45,7 @@ UA_PubSubConnection_getConfig(UA_Server *server, UA_NodeId connectionIdentifier,
     if(!config)
         return UA_STATUSCODE_BADINVALIDARGUMENT;
 
-    UA_PubSubConnection *currentPubSubConnection = UA_PubSubManager_findConnectionbyId(server, connectionIdentifier);
+    UA_PubSubConnection *currentPubSubConnection = UA_PubSubConnection_findConnectionbyId(server, connectionIdentifier);
     if(!currentPubSubConnection)
         return UA_STATUSCODE_BADNOTFOUND;
 
@@ -57,6 +56,21 @@ UA_PubSubConnection_getConfig(UA_Server *server, UA_NodeId connectionIdentifier,
     return UA_STATUSCODE_GOOD;
 }
 
+/**
+ * Find a Connection by the connectionIdentifier.
+ *
+ * @return ptr to Connection or NULL if not found
+ */
+UA_PubSubConnection *
+UA_PubSubConnection_findConnectionbyId(UA_Server *server, UA_NodeId connectionIdentifier) {
+    for(size_t i = 0; i < server->pubSubManager.connectionsSize; i++){
+        if(UA_NodeId_equal(&connectionIdentifier, &server->pubSubManager.connections[i].identifier)){
+            return &server->pubSubManager.connections[i];
+        }
+    }
+    return NULL;
+}
+
 void
 UA_PubSubConnectionConfig_deleteMembers(UA_PubSubConnectionConfig *connectionConfig) {
     UA_String_deleteMembers(&connectionConfig->name);
@@ -80,3 +94,98 @@ UA_PubSubConnection_delete(UA_PubSubConnection *connection) {
     }
     UA_free(connection->config);
 }
+
+/**********************************************/
+/*               PublishedDataSet             */
+/**********************************************/
+
+UA_StatusCode
+UA_PublishedDataSetConfig_copy(const UA_PublishedDataSetConfig *src,
+                               UA_PublishedDataSetConfig *dst) {
+    UA_StatusCode retVal = UA_STATUSCODE_GOOD;
+    memcpy(dst, src, sizeof(UA_PublishedDataSetConfig));
+    retVal |= UA_String_copy(&src->name, &dst->name);
+    switch(src->publishedDataSetType){
+        case UA_PUBSUB_DATASET_PUBLISHEDITEMS:
+            //no additional items
+            break;
+        case UA_PUBSUB_DATASET_PUBLISHEDITEMS_TEMPLATE:
+            if (src->config.itemsTemplate.variablesToAddSize > 0){
+                dst->config.itemsTemplate.variablesToAdd = (UA_PublishedVariableDataType *) UA_calloc(
+                        src->config.itemsTemplate.variablesToAddSize, sizeof(UA_PublishedVariableDataType));
+            }
+            for(size_t i = 0; i < src->config.itemsTemplate.variablesToAddSize; i++){
+                retVal |= UA_PublishedVariableDataType_copy(&src->config.itemsTemplate.variablesToAdd[i], &dst->config.itemsTemplate.variablesToAdd[i]);
+            }
+            retVal |= UA_DataSetMetaDataType_copy(&src->config.itemsTemplate.metaData, &dst->config.itemsTemplate.metaData);
+            break;
+        default:
+            return UA_STATUSCODE_BADINVALIDARGUMENT;
+    }
+    return retVal;
+}
+
+/**
+ * Get the current config of the PublishedDataSetField.
+ *
+ * @return UA_STATUSCODE_GOOD on success
+ */
+UA_StatusCode
+UA_PublishedDataSet_getConfig(UA_Server *server, UA_NodeId publishedDataSetIdentifier,
+                              UA_PublishedDataSetConfig *config){
+    if(!config)
+        return UA_STATUSCODE_BADINVALIDARGUMENT;
+
+    UA_PublishedDataSet *currentPublishedDataSet = UA_PublishedDataSet_findPDSbyId(server,
+                                                                                         publishedDataSetIdentifier);
+    if(!currentPublishedDataSet)
+        return UA_STATUSCODE_BADNOTFOUND;
+
+    UA_PublishedDataSetConfig tmpPublishedDataSetConfig;
+    //deep copy of the actual config
+    UA_PublishedDataSetConfig_copy(&currentPublishedDataSet->config, &tmpPublishedDataSetConfig);
+    *config = tmpPublishedDataSetConfig;
+    return UA_STATUSCODE_GOOD;
+}
+
+/**
+ * Get PDS by the dataSetIdentifier.
+ *
+ * @return ptr to PDS or NULL if not found
+ */
+UA_PublishedDataSet *
+UA_PublishedDataSet_findPDSbyId(UA_Server *server, UA_NodeId identifier){
+    for(size_t i = 0; i < server->pubSubManager.publishedDataSetsSize; i++){
+        if(UA_NodeId_equal(&server->pubSubManager.publishedDataSets[i].identifier, &identifier)){
+            return &server->pubSubManager.publishedDataSets[i];
+        }
+    }
+    return NULL;
+}
+
+void UA_PublishedDataSetConfig_deleteMembers(UA_PublishedDataSetConfig *pdsConfig){
+    //delete pds config
+    UA_String_deleteMembers(&pdsConfig->name);
+    switch (pdsConfig->publishedDataSetType){
+        case UA_PUBSUB_DATASET_PUBLISHEDITEMS:
+            //no additional items
+            break;
+        case UA_PUBSUB_DATASET_PUBLISHEDITEMS_TEMPLATE:
+            if (pdsConfig->config.itemsTemplate.variablesToAddSize > 0){
+                for(size_t i = 0; i < pdsConfig->config.itemsTemplate.variablesToAddSize; i++){
+                    UA_PublishedVariableDataType_deleteMembers(&pdsConfig->config.itemsTemplate.variablesToAdd[i]);
+                }
+                UA_free(pdsConfig->config.itemsTemplate.variablesToAdd);
+            }
+            UA_DataSetMetaDataType_deleteMembers(&pdsConfig->config.itemsTemplate.metaData);
+            break;
+        default:
+            break;
+    }
+}
+
+void UA_PublishedDataSet_delete(UA_PublishedDataSet *publishedDataSet){
+    UA_PublishedDataSetConfig_deleteMembers(&publishedDataSet->config);
+    //delete PDS
+    UA_DataSetMetaDataType_deleteMembers(&publishedDataSet->dataSetMetaData);
+}

src/pubsub/ua_pubsub.h

@@ -18,26 +18,46 @@ extern "C" {
 #include "ua_server.h"
 #include "ua_server_pubsub.h"
 
-//forward declarations
-struct UA_PubSubConnection;
-typedef struct UA_PubSubConnection UA_PubSubConnection;
+/* The configuration structs (public part of PubSub entities) are defined in include/ua_plugin_pubsub.h */
+
+/**********************************************/
+/*            PublishedDataSet                */
+/**********************************************/
+typedef struct{
+    UA_PublishedDataSetConfig config;
+    UA_DataSetMetaDataType dataSetMetaData;
+    LIST_HEAD(UA_ListOfPubSubDataSetField, UA_PubSubDataSetField) fields;
+    UA_NodeId identifier;
+    UA_UInt16 fieldSize;
+    UA_UInt16 promotedFieldsCount;
+} UA_PublishedDataSet;
+
+UA_StatusCode
+UA_PublishedDataSetConfig_copy(const UA_PublishedDataSetConfig *src, UA_PublishedDataSetConfig *dst);
+UA_PublishedDataSet *
+UA_PublishedDataSet_findPDSbyId(UA_Server *server, UA_NodeId identifier);
+void
+UA_PublishedDataSet_delete(UA_PublishedDataSet *publishedDataSet);
 
 /**********************************************/
 /*               Connection                   */
 /**********************************************/
 //the connection config (public part of connection) object is defined in include/ua_plugin_pubsub.h
-struct UA_PubSubConnection{
+typedef struct{
     UA_PubSubConnectionConfig *config;
     //internal fields
     UA_PubSubChannel *channel;
     UA_NodeId identifier;
-};
+} UA_PubSubConnection;
 
 UA_StatusCode
 UA_PubSubConnectionConfig_copy(const UA_PubSubConnectionConfig *src, UA_PubSubConnectionConfig *dst);
-
-void UA_PubSubConnectionConfig_deleteMembers(UA_PubSubConnectionConfig *connectionConfig);
-void UA_PubSubConnection_delete(UA_PubSubConnection *connection);
+UA_PubSubConnection *
+UA_PubSubConnection_findConnectionbyId(UA_Server *server, UA_NodeId connectionIdentifier);
+void
+UA_PubSubConnectionConfig_deleteMembers(UA_PubSubConnectionConfig *connectionConfig);
+void
+UA_PubSubConnection_delete(UA_PubSubConnection *connection);
 
 #ifdef __cplusplus
 } // extern "C"

src/pubsub/ua_pubsub_manager.c

@@ -6,9 +6,10 @@
  */
 
 #include "ua_pubsub_manager.h"
-#include "ua_log_stdout.h"
 #include "server/ua_server_internal.h"
 
+#define UA_DATETIMESTAMP_2000 125911584000000000
+
 /**
  * Add new Connection to the current PubSub configuration.
  *
@@ -28,13 +29,13 @@ UA_Server_addPubSubConnection(UA_Server *server, const UA_PubSubConnectionConfig
             UA_PubSubConnectionConfig *tmpConnectionConfig = (UA_PubSubConnectionConfig *)
                     UA_calloc(1, sizeof(UA_PubSubConnectionConfig));
             if(!tmpConnectionConfig){
-                UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
+                UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
                              "PubSub Connection creation failed. Out of Memory.");
                 return UA_STATUSCODE_BADOUTOFMEMORY;
             }
             //deep copy the given connection config
             if(UA_PubSubConnectionConfig_copy(connectionConfig, tmpConnectionConfig) != UA_STATUSCODE_GOOD){
-                UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
+                UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
                              "PubSub Connection creation failed. Config copy problem.");
                 return UA_STATUSCODE_BADOUTOFMEMORY;
             }
@@ -45,7 +46,7 @@ UA_Server_addPubSubConnection(UA_Server *server, const UA_PubSubConnectionConfig
             if(!newConnectionsField) {
                 UA_PubSubConnectionConfig_deleteMembers(tmpConnectionConfig);
                 UA_free(tmpConnectionConfig);
-                UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
+                UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
                              "PubSub Connection creation failed. Out of Memory.");
                 return UA_STATUSCODE_BADOUTOFMEMORY;
             }
@@ -68,7 +69,7 @@ UA_Server_addPubSubConnection(UA_Server *server, const UA_PubSubConnectionConfig
                     UA_free(newConnectionsField);
                     server->pubSubManager.connections = NULL;
                 }
-                UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
+                UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
                              "PubSub Connection creation failed. Transport layer creation problem.");
                 return UA_STATUSCODE_BADINTERNALERROR;
             }
@@ -80,7 +81,7 @@ UA_Server_addPubSubConnection(UA_Server *server, const UA_PubSubConnectionConfig
             return UA_STATUSCODE_GOOD;
         }
     }
-    UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
+    UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
                  "PubSub Connection creation failed. Requested transport layer not found.");
     return UA_STATUSCODE_BADNOTFOUND;
 }
@@ -126,11 +127,123 @@ UA_StatusCode UA_Server_removePubSubConnection(UA_Server *server, UA_NodeId conn
     return UA_STATUSCODE_GOOD;
 }
 
+/**
+ * Add PublishedDataSet to the current PubSub configuration.
+ *
+ * @param server
+ * @param publishedDataSetConfig config of the new PDS
+ * @param pdsIdentifier nodeId of the generated PDS (NULL if not needed)
+ * @return UA_STATUSCODE_GOOD on success
+ */
+UA_AddPublishedDataSetResult
+UA_Server_addPublishedDataSet(UA_Server *server, const UA_PublishedDataSetConfig *publishedDataSetConfig,
+                              UA_NodeId *pdsIdentifier) {
+    if(!publishedDataSetConfig){
+        UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
+                     "PublishedDataSet creation failed. No config passed in.");
+        return (UA_AddPublishedDataSetResult) {UA_STATUSCODE_BADINVALIDARGUMENT, 0, NULL, {0, 0}};
+    }
+    if(publishedDataSetConfig->publishedDataSetType != UA_PUBSUB_DATASET_PUBLISHEDITEMS){
+        UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
+                     "PublishedDataSet creation failed. Unsupported PublishedDataSet type.");
+        return (UA_AddPublishedDataSetResult) {UA_STATUSCODE_BADINVALIDARGUMENT, 0, NULL, {0, 0}};
+    }
+    //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.");
+        return (UA_AddPublishedDataSetResult) {UA_STATUSCODE_BADINTERNALERROR, 0, NULL, {0, 0}};
+    }
+    //create new PDS and add to UA_PubSubManager
+    UA_PublishedDataSet *newPubSubDataSetField = (UA_PublishedDataSet *)
+            UA_realloc(server->pubSubManager.publishedDataSets,
+                       sizeof(UA_PublishedDataSet) * (server->pubSubManager.publishedDataSetsSize + 1));
+    if(!newPubSubDataSetField) {
+        UA_PublishedDataSetConfig_deleteMembers(&tmpPublishedDataSetConfig);
+        UA_LOG_ERROR(server->config.logger, UA_LOGCATEGORY_SERVER,
+                     "PublishedDataSet creation failed. Out of Memory.");
+        return (UA_AddPublishedDataSetResult) {UA_STATUSCODE_BADOUTOFMEMORY, 0, NULL, {0, 0}};
+    }
+    server->pubSubManager.publishedDataSets = newPubSubDataSetField;
+    UA_PublishedDataSet *newPubSubDataSet = &server->pubSubManager.publishedDataSets[server->pubSubManager.publishedDataSetsSize];
+    memset(newPubSubDataSet, 0, sizeof(UA_PublishedDataSet));
+    newPubSubDataSet->config = tmpPublishedDataSetConfig;
+    if(tmpPublishedDataSetConfig.publishedDataSetType == UA_PUBSUB_DATASET_PUBLISHEDITEMS_TEMPLATE){
+        //parse template config and add fields (later PubSub batch)
+    }
+    //generate unique nodeId
+    UA_PubSubManager_generateUniqueNodeId(server, &newPubSubDataSet->identifier);
+    if(pdsIdentifier != NULL){
+        UA_NodeId_copy(&newPubSubDataSet->identifier, pdsIdentifier);
+    }
+    server->pubSubManager.publishedDataSetsSize++;
+    UA_AddPublishedDataSetResult result = {UA_STATUSCODE_GOOD, 0, NULL,
+                                                 {UA_PubSubConfigurationVersionTimeDifference(),
+                                                  UA_PubSubConfigurationVersionTimeDifference()}};
+    return result;
+}
+
+/**
+ * Remove PublishedDataSet, identified by the NodeId. Deletion of PDS
+ * removes all contained and linked PDS Fields. Connected WriterGroups
+ * will be also removed.
+ *
+ * @param server
+ * @param pdsIdentifier
+ * @return UA_STATUSCODE_GOOD on success
+ */
+UA_StatusCode
+UA_Server_removePublishedDataSet(UA_Server *server, UA_NodeId pdsIdentifier) {
+    //search the identified PublishedDataSet and store the PDS index
+    UA_PublishedDataSet *publishedDataSet = NULL;
+    size_t publishedDataSetIndex;
+    for(publishedDataSetIndex = 0; publishedDataSetIndex < server->pubSubManager.publishedDataSetsSize; publishedDataSetIndex++){
+        if(UA_NodeId_equal(&server->pubSubManager.publishedDataSets[publishedDataSetIndex].identifier, &pdsIdentifier)){
+            publishedDataSet = &server->pubSubManager.publishedDataSets[publishedDataSetIndex];
+            break;
+        };
+    }
+    if(!publishedDataSet){
+        return UA_STATUSCODE_BADNOTFOUND;
+    }
+    UA_PublishedDataSet_delete(publishedDataSet);
+    server->pubSubManager.publishedDataSetsSize--;
+    //copy the last PDS to the removed PDS inside the allocated memory block
+    if(server->pubSubManager.publishedDataSetsSize != publishedDataSetIndex){
+        memcpy(&server->pubSubManager.publishedDataSets[publishedDataSetIndex],
+               &server->pubSubManager.publishedDataSets[server->pubSubManager.publishedDataSetsSize],
+               sizeof(UA_PublishedDataSet));
+    }
+    if(server->pubSubManager.publishedDataSetsSize <= 0){
+        UA_free(server->pubSubManager.publishedDataSets);
+        server->pubSubManager.publishedDataSets = NULL;
+    } else {
+        server->pubSubManager.publishedDataSets = (UA_PublishedDataSet *)
+                UA_realloc(server->pubSubManager.publishedDataSets, sizeof(UA_PublishedDataSet) * server->pubSubManager.publishedDataSetsSize);
+        if (!server->pubSubManager.publishedDataSets) {
+            return UA_STATUSCODE_BADINTERNALERROR;
+        }
+    }
+    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) {
+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);
@@ -146,30 +259,16 @@ void UA_PubSubManager_generateUniqueNodeId(UA_Server *server, UA_NodeId *nodeId)
  */
 void
 UA_PubSubManager_delete(UA_Server *server, UA_PubSubManager *pubSubManager) {
-    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "PubSub cleanup was called.");
+    UA_LOG_INFO(server->config.logger, UA_LOGCATEGORY_SERVER, "PubSub cleanup was called.");
     //remove Connections and WriterGroups
     while(pubSubManager->connectionsSize > 0){
         UA_Server_removePubSubConnection(server, pubSubManager->connections[pubSubManager->connectionsSize-1].identifier);
     }
+    while(pubSubManager->publishedDataSetsSize > 0){
+        UA_Server_removePublishedDataSet(server, pubSubManager->publishedDataSets[pubSubManager->publishedDataSetsSize-1].identifier);
+    }
     //free the currently configured transport layers
     for(size_t i = 0; i < server->config.pubsubTransportLayersSize; i++){
         UA_free(&server->config.pubsubTransportLayers[i]);
     }
 }
-
-/**
- * Find a Connection by the connectionIdentifier.
- *
- * @param server
- * @param connectionIdentifier
- * @return
- */
-UA_PubSubConnection *
-UA_PubSubManager_findConnectionbyId(UA_Server *server, UA_NodeId connectionIdentifier) {
-    for(size_t i = 0; i < server->pubSubManager.connectionsSize; i++){
-        if(UA_NodeId_equal(&connectionIdentifier, &server->pubSubManager.connections[i].identifier)){
-            return &server->pubSubManager.connections[i];
-        }
-    }
-    return NULL;
-}

src/pubsub/ua_pubsub_manager.h

@@ -19,14 +19,16 @@ typedef struct UA_PubSubManager{
     //Connections and PublishedDataSets can exist alone (own lifecycle) -> top level components
     size_t connectionsSize;
     UA_PubSubConnection *connections;
+    size_t publishedDataSetsSize;
+    UA_PublishedDataSet *publishedDataSets;
 } UA_PubSubManager;
 
 void UA_PubSubManager_delete(UA_Server *server, UA_PubSubManager *pubSubManager);
 
 void UA_PubSubManager_generateUniqueNodeId(UA_Server *server, UA_NodeId *nodeId);
 
-UA_PubSubConnection *
-UA_PubSubManager_findConnectionbyId(UA_Server *server, UA_NodeId connectionIdentifier);
+UA_UInt32
+UA_PubSubConfigurationVersionTimeDifference(void);
 
 #ifdef __cplusplus
 } // extern "C"

tests/CMakeLists.txt

@@ -174,6 +174,9 @@ if(UA_ENABLE_PUBSUB)
     add_executable(check_pubsub_connection_udp pubsub/check_pubsub_connection_udp.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-plugins>)
     target_link_libraries(check_pubsub_connection_udp ${LIBS})
     add_test(NAME check_pubsub_connection_udp COMMAND check_pubsub_connection_udp)
+    add_executable(check_pubsub_pds pubsub/check_pubsub_pds.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-plugins>)
+    target_link_libraries(check_pubsub_pds ${LIBS})
+    add_test(check_pubsub_pds ${TESTS_BINARY_DIR}/check_pubsub_pds)
 endif()
 
 add_executable(check_server_readspeed server/check_server_readspeed.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-testplugins>)

tests/pubsub/check_pubsub_pds.c

@@ -0,0 +1,144 @@
+/* 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 - 2018 Fraunhofer IOSB (Author: Andreas Ebner)
+ */
+
+#include <ua_server_pubsub.h>
+#include <src_generated/ua_types_generated_encoding_binary.h>
+#include <ua_types.h>
+#include "ua_config_default.h"
+#include "ua_network_pubsub_udp.h"
+#include "ua_server_internal.h"
+#include "check.h"
+
+UA_Server *server = NULL;
+UA_ServerConfig *config = NULL;
+
+static void setup(void) {
+    config = UA_ServerConfig_new_default();
+    config->pubsubTransportLayers = (UA_PubSubTransportLayer *) UA_malloc(sizeof(UA_PubSubTransportLayer));
+    if(!config->pubsubTransportLayers) {
+        UA_ServerConfig_delete(config);
+    }
+    config->pubsubTransportLayers[0] = UA_PubSubTransportLayerUDPMP();
+    config->pubsubTransportLayersSize++;
+    server = UA_Server_new(config);
+    UA_Server_run_startup(server);
+}
+
+static void teardown(void) {
+    UA_Server_run_shutdown(server);
+    UA_Server_delete(server);
+    UA_ServerConfig_delete(config);
+}
+
+START_TEST(AddPDSWithMinimalValidConfiguration){
+    UA_StatusCode retVal = UA_STATUSCODE_GOOD;
+    UA_PublishedDataSetConfig pdsConfig;
+    memset(&pdsConfig, 0, sizeof(UA_PublishedDataSetConfig));
+    pdsConfig.publishedDataSetType = UA_PUBSUB_DATASET_PUBLISHEDITEMS;
+    pdsConfig.name = UA_STRING("TEST PDS 1");
+    retVal |= UA_Server_addPublishedDataSet(server, &pdsConfig, NULL).addResult;
+    ck_assert_int_eq(server->pubSubManager.publishedDataSetsSize, 1);
+    ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
+    UA_NodeId newPDSNodeID;
+    retVal |= UA_Server_addPublishedDataSet(server, &pdsConfig, &newPDSNodeID).addResult;
+    ck_assert_int_eq(server->pubSubManager.publishedDataSetsSize, 2);
+    ck_assert_int_eq(newPDSNodeID.identifierType, UA_NODEIDTYPE_NUMERIC);
+    ck_assert_int_ne(newPDSNodeID.identifier.numeric, 0);
+    ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
+} END_TEST
+
+START_TEST(AddRemoveAddPDSWithMinimalValidConfiguration){
+    UA_StatusCode retVal = UA_STATUSCODE_GOOD;
+    UA_PublishedDataSetConfig pdsConfig;
+    memset(&pdsConfig, 0, sizeof(UA_PublishedDataSetConfig));
+    pdsConfig.publishedDataSetType = UA_PUBSUB_DATASET_PUBLISHEDITEMS;
+    pdsConfig.name = UA_STRING("TEST PDS 1");
+    UA_NodeId newPDSNodeID;
+    retVal |= UA_Server_addPublishedDataSet(server, &pdsConfig, &newPDSNodeID).addResult;
+    ck_assert_int_eq(server->pubSubManager.publishedDataSetsSize, 1);
+    ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
+    retVal |= UA_Server_removePublishedDataSet(server, newPDSNodeID);
+    ck_assert_int_eq(server->pubSubManager.publishedDataSetsSize, 0);
+    ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
+    retVal |= UA_Server_addPublishedDataSet(server, &pdsConfig, &newPDSNodeID).addResult;
+    ck_assert_int_eq(server->pubSubManager.publishedDataSetsSize, 1);
+    ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
+} END_TEST
+
+START_TEST(AddPDSWithNullConfig){
+    UA_StatusCode retVal = UA_STATUSCODE_GOOD;
+    retVal |= UA_Server_addPublishedDataSet(server, NULL, NULL).addResult;
+    ck_assert_int_eq(server->pubSubManager.publishedDataSetsSize, 0);
+    ck_assert_int_ne(retVal, UA_STATUSCODE_GOOD);
+} END_TEST
+
+START_TEST(AddPDSWithUnsupportedType){
+    UA_StatusCode retVal = UA_STATUSCODE_GOOD;
+    UA_PublishedDataSetConfig pdsConfig;
+    memset(&pdsConfig, 0, sizeof(UA_PublishedDataSetConfig));
+    pdsConfig.name = UA_STRING("TEST PDS 1");
+    pdsConfig.publishedDataSetType = UA_PUBSUB_DATASET_PUBLISHEDITEMS_TEMPLATE;
+    retVal |= UA_Server_addPublishedDataSet(server, &pdsConfig, NULL).addResult;
+    ck_assert_int_eq(server->pubSubManager.publishedDataSetsSize, 0);
+    ck_assert_int_ne(retVal, UA_STATUSCODE_GOOD);
+    pdsConfig.publishedDataSetType = UA_PUBSUB_DATASET_PUBLISHEDEVENTS;
+    ck_assert_int_eq(server->pubSubManager.publishedDataSetsSize, 0);
+    ck_assert_int_ne(retVal, UA_STATUSCODE_GOOD);
+    pdsConfig.publishedDataSetType = UA_PUBSUB_DATASET_PUBLISHEDEVENTS_TEMPLATE;
+    ck_assert_int_eq(server->pubSubManager.publishedDataSetsSize, 0);
+    ck_assert_int_ne(retVal, UA_STATUSCODE_GOOD);
+} END_TEST
+
+START_TEST(GetPDSConfigurationAndCompareValues){
+    UA_StatusCode retVal = UA_STATUSCODE_GOOD;
+    UA_PublishedDataSetConfig pdsConfig;
+    memset(&pdsConfig, 0, sizeof(UA_PublishedDataSetConfig));
+    pdsConfig.publishedDataSetType = UA_PUBSUB_DATASET_PUBLISHEDITEMS;
+    pdsConfig.name = UA_STRING("TEST PDS 1");
+    UA_NodeId pdsIdentifier;
+    retVal |= UA_Server_addPublishedDataSet(server, &pdsConfig, &pdsIdentifier).addResult;
+    ck_assert_int_eq(server->pubSubManager.publishedDataSetsSize, 1);
+    ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
+    UA_PublishedDataSetConfig pdsConfigCopy;
+    memset(&pdsConfigCopy, 0, sizeof(UA_PublishedDataSetConfig));
+        UA_PublishedDataSet_getConfig(server, pdsIdentifier, &pdsConfigCopy);
+    ck_assert_int_eq(UA_String_equal(&pdsConfig.name, &pdsConfigCopy.name), UA_TRUE);
+    UA_PublishedDataSetConfig_deleteMembers(&pdsConfigCopy);
+} END_TEST
+
+int main(void) {
+    TCase *tc_add_pubsub_pds_minimal_config = tcase_create("Create PubSub PublishedDataItem with minimal valid config");
+    tcase_add_checked_fixture(tc_add_pubsub_pds_minimal_config, setup, teardown);
+    tcase_add_test(tc_add_pubsub_pds_minimal_config, AddPDSWithMinimalValidConfiguration);
+    tcase_add_test(tc_add_pubsub_pds_minimal_config, AddRemoveAddPDSWithMinimalValidConfiguration);
+
+    TCase *tc_add_pubsub_pds_invalid_config = tcase_create("Create PubSub PublishedDataItem with minimal invalid config");
+    tcase_add_checked_fixture(tc_add_pubsub_pds_invalid_config, setup, teardown);
+    tcase_add_test(tc_add_pubsub_pds_invalid_config, AddPDSWithNullConfig);
+    tcase_add_test(tc_add_pubsub_pds_invalid_config, AddPDSWithUnsupportedType);
+
+    TCase *tc_add_pubsub_pds_handling_utils = tcase_create("PubSub PublishedDataSet handling");
+    tcase_add_checked_fixture(tc_add_pubsub_pds_handling_utils, setup, teardown);
+    tcase_add_test(tc_add_pubsub_pds_handling_utils, GetPDSConfigurationAndCompareValues);
+    //tcase_add_test(tc_add_pubsub_connections_maximal_config, GetMaximalConnectionConfigurationAndCompareValues);
+
+    Suite *s = suite_create("PubSub PublishedDataSets handling");
+    suite_add_tcase(s, tc_add_pubsub_pds_minimal_config);
+    suite_add_tcase(s, tc_add_pubsub_pds_invalid_config);
+    suite_add_tcase(s, tc_add_pubsub_pds_handling_utils);
+
+
+    //suite_add_tcase(s, tc_add_pubsub_connections_maximal_config);
+    //suite_add_tcase(s, tc_decode);
+
+    SRunner *sr = srunner_create(s);
+    srunner_set_fork_status(sr, CK_NOFORK);
+    srunner_run_all(sr,CK_NORMAL);
+    int number_failed = srunner_ntests_failed(sr);
+    srunner_free(sr);
+    return (number_failed == 0) ? EXIT_SUCCESS : EXIT_FAILURE;
+}

tools/schema/datatypes_minimal.txt

@@ -202,4 +202,16 @@ AggregateFilter
 SetTriggeringRequest
 SetTriggeringResponse
 KeyValuePair
-NetworkAddressUrlDataType
+NetworkAddressUrlDataType
+ConfigurationVersionDataType
+DataSetMetaDataType
+FieldMetaData
+DataSetFieldFlags
+StructureDescription
+EnumDescription
+StructureDefinition
+EnumDefinition
+StructureType
+StructureField
+EnumField
+PublishedVariableDataType