feat(pubsub): pubsub configuration from binary file (#3806)

CMakeLists.txt

@@ -264,6 +264,8 @@ option(UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS "Enable PubSub informationmodel
 mark_as_advanced(UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS)
 option(UA_ENABLE_PUBSUB_DELTAFRAMES "Enable sending of delta frames with only the changes" OFF)
 mark_as_advanced(UA_ENABLE_PUBSUB_DELTAFRAMES)
+option(UA_ENABLE_PUBSUB_FILE_CONFIG "Enable loading PubSub Config from file extension" OFF)
+mark_as_advanced(UA_ENABLE_PUBSUB_FILE_CONFIG) 
 #RT and Transport PubSub settings
 option(UA_ENABLE_PUBSUB_ETH_UADP "Enable publish/subscribe UADP over Ethernet" OFF)
 mark_as_advanced(UA_ENABLE_PUBSUB_ETH_UADP)
@@ -302,6 +304,12 @@ if(UA_ENABLE_PUBSUB_INFORMATIONMODEL)
     endif()
 endif()
 
+if(UA_ENABLE_PUBSUB_FILE_CONFIG)
+    if(NOT UA_ENABLE_PUBSUB)
+        message(FATAL_ERROR "PubSub needs to be enabled")
+    endif()
+endif()
+
 if(UA_ENABLE_PUBSUB_CUSTOM_PUBLISH_HANDLING)
     if(NOT UA_ENABLE_PUBSUB)
         message(FATAL_ERROR "Custom publish callback handling cannot be used with PubSub function disabled")
@@ -743,7 +751,8 @@ set(internal_headers ${PROJECT_SOURCE_DIR}/deps/open62541_queue.h
                      ${PROJECT_SOURCE_DIR}/src/server/ua_server_async.h
                      ${PROJECT_SOURCE_DIR}/src/server/ua_server_internal.h
                      ${PROJECT_SOURCE_DIR}/src/server/ua_services.h
-                     ${PROJECT_SOURCE_DIR}/src/client/ua_client_internal.h)
+                     ${PROJECT_SOURCE_DIR}/src/client/ua_client_internal.h
+                     ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub_config.h)
 
 # TODO: make client optional
 set(lib_sources ${PROJECT_SOURCE_DIR}/src/ua_types.c
@@ -792,7 +801,9 @@ set(lib_sources ${PROJECT_SOURCE_DIR}/src/ua_types.c
                 # dependencies
                 ${PROJECT_SOURCE_DIR}/deps/libc_time.c
                 ${PROJECT_SOURCE_DIR}/deps/pcg_basic.c
-                ${PROJECT_SOURCE_DIR}/deps/base64.c)
+                ${PROJECT_SOURCE_DIR}/deps/base64.c
+                
+                ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub_config.c)
 
 set(default_plugin_headers ${PROJECT_SOURCE_DIR}/plugins/include/open62541/plugin/accesscontrol_default.h
                            ${PROJECT_SOURCE_DIR}/plugins/include/open62541/plugin/pki_default.h

examples/CMakeLists.txt

@@ -219,4 +219,7 @@ if(UA_ENABLE_PUBSUB)
             add_example(tutorial_pubsub_mqtt_publish pubsub/tutorial_pubsub_mqtt_publish.c)
         endif()
     endif()
+    if(UA_ENABLE_PUBSUB_FILE_CONFIG)
+        add_example(server_pubsub_file_configuration pubsub/server_pubsub_file_configuration.c)
+    endif()
 endif()

examples/common.h

@@ -35,3 +35,23 @@ loadFile(const char *const path) {
 
     return fileContents;
 }
+
+static UA_INLINE UA_StatusCode
+writeFile(const char* const path, const UA_ByteString buffer) {
+    FILE *fp = NULL;
+
+    fp = fopen(path, "wb");
+    if(fp == NULL) 
+        return UA_STATUSCODE_BADINTERNALERROR;
+
+    for(UA_UInt32 bufIndex = 0; bufIndex < buffer.length; bufIndex++) {
+        int retVal = fputc(buffer.data[bufIndex], fp);
+        if(retVal == EOF) {
+            fclose(fp);
+            return UA_STATUSCODE_BADINTERNALERROR;
+        }
+    }
+
+    fclose(fp);
+    return UA_STATUSCODE_GOOD;
+}

examples/pubsub/server_pubsub_file_configuration.c

@@ -0,0 +1,152 @@
+/* This work is licensed under a Creative Commons CCZero 1.0 Universal License.
+ * See http://creativecommons.org/publicdomain/zero/1.0/ for more information. */
+
+/* Includes */
+#include <open62541/plugin/log_stdout.h>
+#include <open62541/plugin/pubsub.h>
+#include <open62541/server.h>
+#include <open62541/server_config_default.h>
+
+#include "ua_pubsub_config.h"
+
+#include <signal.h>
+
+#include "common.h"
+
+/* Global variables */
+volatile UA_Boolean g_running = true;
+
+/* Signal handler */
+static void stopHandler(int signum) {
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "Exiting...\n");
+    g_running = false;
+}
+
+/* Function to give user information about correct usage */
+static void usage_info(void) {
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "USAGE: ./server_pubsub_file_configuration [port] [name of UA_Binary_Config_File]");
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "Alternatively, Bin-files can be loaded via configuration method calls.");
+}
+
+int main(int argc, char** argv) {
+    signal(SIGINT, stopHandler);
+    signal(SIGTERM, stopHandler);
+
+    UA_Boolean loadPubSubFromFile = UA_FALSE;
+    UA_UInt16 port = 4840;
+
+    /* 1. Check arguments and set name of PubSub configuration file*/
+    switch(argc) {
+        case 2:
+            port = (unsigned short)atoi(argv[1]);
+            break;
+        case 3:
+            port = (unsigned short)atoi(argv[1]);
+            loadPubSubFromFile = UA_TRUE;
+            break;
+
+        default:
+            usage_info(); 
+    }
+
+    /* 2. Initialize Server */
+    UA_Server *server = UA_Server_new();
+    UA_ServerConfig *config = UA_Server_getConfig(server);
+    UA_ServerConfig_setMinimal(config, port, NULL); /* creates server on default port 4840 */
+
+    /* 3. Add variable nodes to the server */
+    UA_VariableAttributes attr;
+    UA_NodeId parentReferenceNodeId = UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES);
+    UA_NodeId pubSubVariableObjectId = UA_NODEID_STRING(1, "PubSubObject");
+    
+    UA_ObjectAttributes oAttr = UA_ObjectAttributes_default;
+    oAttr.displayName = UA_LOCALIZEDTEXT("en-US", "PubSubVariables");
+    UA_Server_addObjectNode(server, pubSubVariableObjectId,
+                            UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
+                            parentReferenceNodeId,
+                            UA_QUALIFIEDNAME(1, "PubSubVariables"), UA_NODEID_NUMERIC(0, UA_NS0ID_BASEOBJECTTYPE),
+                            oAttr, NULL, NULL);
+
+    attr = UA_VariableAttributes_default;
+    UA_Boolean myBool = UA_TRUE;
+    UA_Variant_setScalar(&attr.value, &myBool, &UA_TYPES[UA_TYPES_BOOLEAN]);
+    attr.description = UA_LOCALIZEDTEXT("en-US","BoolToggle");
+    attr.displayName = UA_LOCALIZEDTEXT("en-US","BoolToggle");
+    attr.dataType = UA_TYPES[UA_TYPES_BOOLEAN].typeId;
+    attr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE;
+    UA_NodeId myBoolNodeId = UA_NODEID_STRING(1, "BoolToggle");
+    UA_QualifiedName myBoolName = UA_QUALIFIEDNAME(1, "BoolToggle");
+    UA_Server_addVariableNode(server, myBoolNodeId, pubSubVariableObjectId,
+                              parentReferenceNodeId, myBoolName,
+                              UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE), attr, NULL, NULL);
+
+    attr = UA_VariableAttributes_default;
+    UA_Int32 myInteger = 0;
+    UA_Variant_setScalar(&attr.value, &myInteger, &UA_TYPES[UA_TYPES_INT32]);
+    attr.description = UA_LOCALIZEDTEXT("en-US","Int32");
+    attr.displayName = UA_LOCALIZEDTEXT("en-US","Int32");
+    attr.dataType = UA_TYPES[UA_TYPES_INT32].typeId;
+    attr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE;
+    UA_NodeId myIntegerNodeId = UA_NODEID_STRING(1, "Int32");
+    UA_QualifiedName myIntegerName = UA_QUALIFIEDNAME(1, "Int32");
+    UA_Server_addVariableNode(server, myIntegerNodeId, pubSubVariableObjectId,
+                              parentReferenceNodeId, myIntegerName,
+                              UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE), attr, NULL, NULL);
+
+    attr = UA_VariableAttributes_default;
+    UA_Int32 myIntegerFast = 24;
+    UA_Variant_setScalar(&attr.value, &myIntegerFast, &UA_TYPES[UA_TYPES_INT32]);
+    attr.description = UA_LOCALIZEDTEXT("en-US","Int32Fast");
+    attr.displayName = UA_LOCALIZEDTEXT("en-US","Int32Fast");
+    attr.dataType = UA_TYPES[UA_TYPES_INT32].typeId;
+    attr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE;
+    UA_NodeId myIntegerFastNodeId = UA_NODEID_STRING(1, "Int32Fast");
+    UA_QualifiedName myIntegerFastName = UA_QUALIFIEDNAME(1, "Int32Fast");
+    UA_Server_addVariableNode(server, myIntegerFastNodeId, pubSubVariableObjectId,
+                              parentReferenceNodeId, myIntegerFastName,
+                              UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE), attr, NULL, NULL);
+
+    attr = UA_VariableAttributes_default;
+    UA_DateTime myDate = UA_DateTime_now() + UA_DateTime_localTimeUtcOffset();
+    UA_Variant_setScalar(&attr.value, &myDate, &UA_TYPES[UA_TYPES_DATETIME]);
+    attr.description = UA_LOCALIZEDTEXT("en-US","DateTime");
+    attr.displayName = UA_LOCALIZEDTEXT("en-US","DateTime");
+    attr.dataType = UA_TYPES[UA_TYPES_DATETIME].typeId;
+    attr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE;
+    UA_NodeId myDateNodeId = UA_NODEID_STRING(1, "DateTime");
+    UA_QualifiedName myDateName = UA_QUALIFIEDNAME(1, "DateTime");
+    UA_Server_addVariableNode(server, myDateNodeId, pubSubVariableObjectId,
+                              parentReferenceNodeId, myDateName,
+                              UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE), attr, NULL, NULL);
+    /* 4. load configuration from file */
+    if(loadPubSubFromFile) {
+        UA_ByteString configuration = loadFile(argv[2]);
+        UA_PubSubManager_loadPubSubConfigFromByteString(server, configuration);
+    }
+
+    /* 5. start server */
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "Starting server...");
+    UA_StatusCode statusCode = UA_Server_run(server, &g_running);
+    if(statusCode != UA_STATUSCODE_GOOD) {
+        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "Server stopped. Status code: 0x%x\n", statusCode);
+        return(-1);
+    }
+
+    if(loadPubSubFromFile) {
+        /* 6. save current configuration to file */
+        UA_ByteString buffer = UA_BYTESTRING_NULL;
+        statusCode = UA_PubSubManager_getEncodedPubSubConfiguration(server, &buffer);
+        if(statusCode == UA_STATUSCODE_GOOD)
+            statusCode = writeFile(argv[2], buffer);
+        
+        if(statusCode != UA_STATUSCODE_GOOD) 
+            UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "Saving PubSub configuration to file failed. StatusCode: 0x%x\n", statusCode);
+        
+        UA_ByteString_deleteMembers(&buffer);
+    }
+
+    UA_Server_delete(server);
+
+    return 0;
+}
+/******************************************************************************************************/

include/open62541/config.h.in

@@ -27,6 +27,7 @@
 #cmakedefine UA_ENABLE_NODEMANAGEMENT
 #cmakedefine UA_ENABLE_SUBSCRIPTIONS
 #cmakedefine UA_ENABLE_PUBSUB
+#cmakedefine UA_ENABLE_PUBSUB_FILE_CONFIG
 #cmakedefine UA_ENABLE_PUBSUB_ETH_UADP
 #cmakedefine UA_ENABLE_PUBSUB_ETH_UADP_ETF
 #cmakedefine UA_ENABLE_PUBSUB_ETH_UADP_XDP

include/open62541/server_pubsub.h

@@ -110,6 +110,8 @@ _UA_BEGIN_DECLS
  *  The PubSub messages differentiate between keyframe (all published values contained) and deltaframe (only changed values contained) messages.
  *  Deltaframe messages creation consumes some additional ressources and can be disabled with this flag. Disabled by default.
  *  Compile the human-readable name of the StatusCodes into the binary. Disabled by default.
+ * **UA_ENABLE_PUBSUB_FILE_CONFIG**
+ *  Enable loading OPC UA PubSub configuration from File/ByteString. Enabling PubSub informationmodel methods also will add a method to the Publish/Subscribe object which allows configuring PubSub at runtime.
  * **UA_ENABLE_PUBSUB_INFORMATIONMODEL**
  *  Enable the information model representation of the PubSub configuration. For more details take a look at the following section `PubSub Information Model Representation`. Disabled by default.
  * **UA_ENABLE_PUBSUB_CUSTOM_PUBLISH_HANDLING**

src/pubsub/ua_pubsub.h

@@ -4,6 +4,8 @@
  *
  * Copyright (c) 2017-2018 Fraunhofer IOSB (Author: Andreas Ebner)
  * Copyright (c) 2019 Kalycito Infotech Private Limited
+ * Copyright (c) 2020 Yannick Wallerer, Siemens AG
+ * Copyright (c) 2020 Thomas Fischer, Siemens AG
  */
 
 #ifndef UA_PUBSUB_H_
@@ -63,6 +65,7 @@ typedef struct UA_PubSubConnection{
     UA_PubSubChannel *channel;
     UA_NodeId identifier;
     LIST_HEAD(UA_ListOfWriterGroup, UA_WriterGroup) writerGroups;
+    size_t writerGroupsSize;
     LIST_HEAD(UA_ListOfPubSubReaderGroup, UA_ReaderGroup) readerGroups;
     size_t readerGroupsSize;
     TAILQ_ENTRY(UA_PubSubConnection) listEntry;

src/pubsub/ua_pubsub_config.h

@@ -0,0 +1,52 @@
+/* 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) 2020 Yannick Wallerer, Siemens AG
+ * Copyright (c) 2020 Thomas Fischer, Siemens AG
+ */
+
+#ifdef UA_ENABLE_PUBSUB_FILE_CONFIG
+
+#ifndef UA_PUBSUB_CONFIG_H_
+#define UA_PUBSUB_CONFIG_H_
+
+#include <open62541/types_generated.h>  /* Defines UA data types */
+#include <open62541/server.h>           /* Defines UA_Server     */
+
+/* UA_PubSubManager_loadPubSubConfigFromByteString() */
+/**
+ * @brief       Decodes the information from the ByteString. If the decoded content is a PubSubConfiguration in a UABinaryFileDataType-object               
+ *              it will overwrite the current PubSub configuration from the server.
+ * 
+ * @param       server      [bi]    Pointer to Server object that shall be configured
+ * @param       buffer      [in]    Relative path and name of the file that contains the PubSub configuration
+ * 
+ * @return      UA_STATUSCODE_GOOD on success
+ */
+UA_StatusCode 
+UA_PubSubManager_loadPubSubConfigFromByteString
+(
+    /*[bi]*/    UA_Server *server, 
+    /*[in]*/    const UA_ByteString buffer
+);
+
+/* UA_PubSubManager_getEncodedPubSubConfiguration() */
+/**
+ * @brief       Saves the current PubSub configuration of a server in a byteString.
+ * 
+ * @param       server  [in]    Pointer to server object, that contains the PubSubConfiguration
+ * @param       buffer  [out]    Pointer to a byteString object
+ *
+ * @return      UA_STATUSCODE_GOOD on success
+ */
+UA_StatusCode
+UA_PubSubManager_getEncodedPubSubConfiguration
+(
+    /*[bi]*/    UA_Server *server, 
+    /*[out]*/   UA_ByteString *buffer
+);
+
+#endif /* UA_PUBSUB_CONFIG_H_ */
+
+#endif /* UA_ENABLE_PUBSUB_FILE_CONFIG */

src/pubsub/ua_pubsub_ns0.c

@@ -4,11 +4,17 @@
  *
  * Copyright (c) 2017-2018 Fraunhofer IOSB (Author: Andreas Ebner)
  * Copyright (c) 2019 Kalycito Infotech Private Limited
+ * Copyright (c) 2020 Yannick Wallerer, Siemens AG
+ * Copyright (c) 2020 Thomas Fischer, Siemens AG
  */
 
 #include <open62541/types.h>
 #include "ua_pubsub_ns0.h"
 
+#ifdef UA_ENABLE_PUBSUB_FILE_CONFIG
+#include "ua_pubsub_config.h"
+#endif
+
 #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL /* conditional compilation */
 
 typedef struct{
@@ -1080,6 +1086,124 @@ publishedDataItemsTypeDestructor(UA_Server *server,
         UA_free(childContext);
 }
 
+/*************************************/
+/*         PubSub configurator       */
+/*************************************/
+
+/* UA_loadPubSubConfigMethodCallback() */
+/**
+ * @brief   callback function that will be executed when the method "PubSub configurator (replace config)" is called.
+ */
+#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS 
+#ifdef UA_ENABLE_PUBSUB_FILE_CONFIG
+static UA_StatusCode
+UA_loadPubSubConfigMethodCallback(UA_Server *server,
+                                  const UA_NodeId *sessionId, void *sessionHandle,
+                                  const UA_NodeId *methodId, void *methodContext,
+                                  const UA_NodeId *objectId, void *objectContext,
+                                  size_t inputSize, const UA_Variant *input,
+                                  size_t outputSize, UA_Variant *output) {
+    if(inputSize == 1) {
+        UA_ByteString *inputStr = (UA_ByteString*)input->data;
+        return UA_PubSubManager_loadPubSubConfigFromByteString(server, *inputStr);
+    } else if(inputSize > 1) {
+        return UA_STATUSCODE_BADTOOMANYARGUMENTS;
+    } else {
+        return UA_STATUSCODE_BADARGUMENTSMISSING;
+    }
+}
+#endif
+#endif /*UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS*/
+
+
+/* UA_addLoadPubSubConfigMethod() */
+/**
+ * @brief       Adds method node to server. This method is used to load binary files for PubSub 
+ *              configuration and delete / replace old PubSub configurations.
+ * 
+ * @param       server      [bi]    UA_Server object that shall contain the method.
+ * 
+ * @return      UA_STATUSCODE_GOOD on success
+ */
+#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS
+#ifdef UA_ENABLE_PUBSUB_FILE_CONFIG
+static UA_StatusCode 
+UA_addLoadPubSubConfigMethod(UA_Server *server) {
+    UA_Argument inputArgument;
+    UA_Argument_init(&inputArgument);
+    inputArgument.description = UA_LOCALIZEDTEXT("en-US", "PubSub config binfile");
+    inputArgument.name = UA_STRING("BinFile");
+    inputArgument.dataType = UA_TYPES[UA_TYPES_BYTESTRING].typeId;
+    inputArgument.valueRank = UA_VALUERANK_SCALAR;
+
+    UA_MethodAttributes configAttr = UA_MethodAttributes_default;
+    configAttr.description = UA_LOCALIZEDTEXT("en-US","Load binary configuration file");
+    configAttr.displayName = UA_LOCALIZEDTEXT("en-US","LoadPubSubConfigurationFile");
+    configAttr.executable = true;
+    configAttr.userExecutable = true;
+    UA_StatusCode retVal = UA_Server_addMethodNode(server, UA_NODEID_NULL,
+                                                   UA_NODEID_NUMERIC(0, UA_NS0ID_PUBLISHSUBSCRIBE),
+                                                   UA_NODEID_NUMERIC(0, UA_NS0ID_HASORDEREDCOMPONENT),
+                                                   UA_QUALIFIEDNAME(1, "PubSub configuration"),
+                                                   configAttr, &UA_loadPubSubConfigMethodCallback,
+                                                   1, &inputArgument, 0, NULL, NULL, NULL);
+    return retVal;
+}
+#endif
+#endif /*UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS*/
+
+
+/* UA_deletePubSubConfigMethodCallback() */
+/**
+ * @brief   callback function that will be executed when the method "PubSub configurator (delete config)" is called.
+ */
+#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS
+#ifdef UA_ENABLE_PUBSUB_FILE_CONFIG
+static UA_StatusCode
+UA_deletePubSubConfigMethodCallback(UA_Server *server,
+                                    const UA_NodeId *sessionId, void *sessionHandle,
+                                    const UA_NodeId *methodId, void *methodContext,
+                                    const UA_NodeId *objectId, void *objectContext,
+                                    size_t inputSize, const UA_Variant *input,
+                                    size_t outputSize, UA_Variant *output) {
+    UA_PubSubManager_delete(server, &(server->pubSubManager));
+    
+    return UA_STATUSCODE_GOOD;
+}
+#endif
+#endif /*UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS*/
+
+
+/* UA_addDeletePubSubConfigMethod() */
+/**
+ * @brief       Adds method node to server. This method is used to delete the current PubSub configuration.
+ * 
+ * @param       server      [bi]    UA_Server object that shall contain the method.
+ * 
+ * @return      UA_STATUSCODE_GOOD on success
+ */
+#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS
+#ifdef UA_ENABLE_PUBSUB_FILE_CONFIG
+static UA_StatusCode 
+UA_addDeletePubSubConfigMethod(UA_Server *server) {
+    UA_MethodAttributes configAttr = UA_MethodAttributes_default;
+    configAttr.description = UA_LOCALIZEDTEXT("en-US","Delete current PubSub configuration");
+    configAttr.displayName = UA_LOCALIZEDTEXT("en-US","DeletePubSubConfiguration");
+    configAttr.executable = true;
+    configAttr.userExecutable = true;
+    UA_StatusCode retVal = UA_Server_addMethodNode(server, UA_NODEID_NULL,
+                                                   UA_NODEID_NUMERIC(0, UA_NS0ID_PUBLISHSUBSCRIBE),
+                                                   UA_NODEID_NUMERIC(0, UA_NS0ID_HASORDEREDCOMPONENT),
+                                                   UA_QUALIFIEDNAME(1, "Delete PubSub config"),
+                                                   configAttr, &UA_deletePubSubConfigMethodCallback,
+                                                   0, NULL, 0, NULL, NULL, NULL);
+    return retVal;
+}
+#endif
+#endif /*UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS*/
+
+
+
 UA_StatusCode
 UA_Server_initPubSubNS0(UA_Server *server) {
     UA_StatusCode retVal = UA_STATUSCODE_GOOD;
@@ -1127,6 +1251,11 @@ UA_Server_initPubSubNS0(UA_Server *server) {
     retVal |= UA_Server_setMethodNode_callback(server, UA_NODEID_NUMERIC(0, UA_NS0ID_READERGROUPTYPE_ADDDATASETREADER), addDataSetReaderAction);
     retVal |= UA_Server_setMethodNode_callback(server, UA_NODEID_NUMERIC(0, UA_NS0ID_READERGROUPTYPE_REMOVEDATASETREADER), removeDataSetReaderAction);
 
+#ifdef UA_ENABLE_PUBSUB_FILE_CONFIG
+    retVal |= UA_addLoadPubSubConfigMethod(server);
+    retVal |= UA_addDeletePubSubConfigMethod(server);
+#endif
+
 #else
     retVal |= UA_Server_deleteReference(server, UA_NODEID_NUMERIC(0, UA_NS0ID_PUBLISHSUBSCRIBE), UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT), true,
                                         UA_EXPANDEDNODEID_NUMERIC(0, UA_NS0ID_PUBLISHSUBSCRIBE_ADDCONNECTION),

src/pubsub/ua_pubsub_writer.c

@@ -5,6 +5,8 @@
  * Copyright (c) 2017-2019 Fraunhofer IOSB (Author: Andreas Ebner)
  * Copyright (c) 2019 Fraunhofer IOSB (Author: Julius Pfrommer)
  * Copyright (c) 2019 Kalycito Infotech Private Limited
+ * Copyright (c) 2020 Yannick Wallerer, Siemens AG
+ * Copyright (c) 2020 Thomas Fischer, Siemens AG
  */
 
 #include <open62541/server_pubsub.h>
@@ -201,6 +203,7 @@ UA_Server_addWriterGroup(UA_Server *server, const UA_NodeId connection,
     }
     newWriterGroup->config = tmpWriterGroupConfig;
     LIST_INSERT_HEAD(&currentConnectionContext->writerGroups, newWriterGroup, listEntry);
+    currentConnectionContext->writerGroupsSize++;
 #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
     addWriterGroupRepresentation(server, newWriterGroup);
 #endif
@@ -232,6 +235,8 @@ UA_Server_removeWriterGroup(UA_Server *server, const UA_NodeId writerGroup) {
         //unregister the publish callback
         UA_PubSubManager_removeRepeatedPubSubCallback(server, wg->publishCallbackId);
     }
+
+    connection->writerGroupsSize--;
 #ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
     removeGroupRepresentation(server, wg);
 #endif

tests/CMakeLists.txt

@@ -419,6 +419,11 @@ if(UA_ENABLE_PUBSUB)
             add_test_valgrind(pubsub_connection_mqtt ${TESTS_BINARY_DIR}/check_pubsub_connection_mqtt)
         endif()
     endif()
+    if(UA_ENABLE_PUBSUB_FILE_CONFIG)
+        add_executable(check_pubsub_configuration pubsub/check_pubsub_configuration.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-plugins>)
+        target_link_libraries(check_pubsub_configuration ${LIBS})
+        add_test_valgrind(pubsub_configuration ${TESTS_BINARY_DIR}/check_pubsub_configuration)
+    endif()
 endif()
 
 add_executable(check_server_readspeed server/check_server_readspeed.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-testplugins>)

tests/common.h

@@ -0,0 +1,37 @@
+/* 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 <open62541/types.h>
+#include <open62541/types_generated_handling.h>
+
+/* loadFile parses the certificate file.
+ *
+ * @param  path               specifies the file name given in argv[]
+ * @return Returns the file content after parsing */
+static UA_INLINE UA_ByteString
+loadFile(const char *const path) {
+    UA_ByteString fileContents = UA_STRING_NULL;
+
+    /* Open the file */
+    FILE *fp = fopen(path, "rb");
+    if(!fp) {
+        errno = 0; /* We read errno also from the tcp layer... */
+        return fileContents;
+    }
+
+    /* Get the file length, allocate the data and read */
+    fseek(fp, 0, SEEK_END);
+    fileContents.length = (size_t)ftell(fp);
+    fileContents.data = (UA_Byte *)UA_malloc(fileContents.length * sizeof(UA_Byte));
+    if(fileContents.data) {
+        fseek(fp, 0, SEEK_SET);
+        size_t read = fread(fileContents.data, sizeof(UA_Byte), fileContents.length, fp);
+        if(read != fileContents.length)
+            UA_ByteString_clear(&fileContents);
+    } else {
+        fileContents.length = 0;
+    }
+    fclose(fp);
+
+    return fileContents;
+}

tests/pubsub/check_pubsub_configuration.c

@@ -0,0 +1,111 @@
+/* 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) 2020 Siemens AG (Author: Thomas Fischer)
+ */
+
+#include <open62541/plugin/pubsub_udp.h>
+#include <open62541/server_config_default.h>
+#include <open62541/server_pubsub.h>
+#include "../common.h"
+
+#include "open62541/types_generated_encoding_binary.h"
+
+#include "ua_pubsub.h"
+#include "pubsub/ua_pubsub_config.h"
+#include "ua_server_internal.h"
+
+#include <check.h>
+
+UA_Server *server = NULL;
+
+static void setup(void) {
+    server = UA_Server_new();
+    UA_ServerConfig *config = UA_Server_getConfig(server);
+    UA_ServerConfig_setDefault(config);
+
+    UA_Server_run_startup(server);
+}
+
+static void teardown(void) {
+    UA_Server_run_shutdown(server);
+    UA_Server_delete(server);
+}
+
+START_TEST(AddPublisherUsingBinaryFile) {
+    UA_ByteString publisherConfiguration = loadFile("../tests/pubsub/check_publisher_configuration.bin");
+    UA_StatusCode retVal = UA_PubSubManager_loadPubSubConfigFromByteString(server, publisherConfiguration);
+    ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
+    UA_PubSubConnection *connection;
+    UA_WriterGroup *writerGroup;
+    UA_DataSetWriter *dataSetWriter;
+    size_t connectionCount = 0;
+    size_t writerGroupCount = 0;
+    size_t dataSetWriterCount = 0;
+    TAILQ_FOREACH(connection, &server->pubSubManager.connections, listEntry) {
+        connectionCount++;
+        char* expectedConnectionName = "UADP Connection 1";
+        ck_assert_str_eq(expectedConnectionName, (char*)connection->config->name.data);
+        LIST_FOREACH(writerGroup, &connection->writerGroups, listEntry){
+            writerGroupCount++;
+            char* expectedWgName = "Demo WriterGroup";
+            ck_assert_str_eq(expectedWgName, (char*)writerGroup->config.name.data);
+            LIST_FOREACH(dataSetWriter, &writerGroup->writers, listEntry){
+                dataSetWriterCount++;
+                char* expectedWriterName = "Demo DataSetWriter";
+                ck_assert_str_eq(expectedWriterName, (char*)dataSetWriter->config.name.data);
+            }
+        }
+    }
+    ck_assert_int_eq(connectionCount, 1);
+    ck_assert_int_eq(writerGroupCount, 1);
+    ck_assert_int_eq(dataSetWriterCount, 1);
+} END_TEST
+
+START_TEST(AddSubscriberUsingBinaryFile) {
+    UA_ByteString subscriberConfiguration = loadFile("../tests/pubsub/check_subscriber_configuration.bin");
+    UA_StatusCode retVal = UA_PubSubManager_loadPubSubConfigFromByteString(server, subscriberConfiguration);
+    ck_assert_int_eq(retVal, UA_STATUSCODE_GOOD);
+    UA_PubSubConnection *connection;
+    UA_ReaderGroup *readerGroup;
+    UA_DataSetReader *dataSetReader;
+    size_t connectionCount = 0;
+    size_t readerGroupCount = 0;
+    size_t dataSetReaderCount = 0;
+    TAILQ_FOREACH(connection, &server->pubSubManager.connections, listEntry) {
+        connectionCount++;
+        char* expectedConnectionName = "UDPMC Connection 1";
+        ck_assert_str_eq(expectedConnectionName, (char*)connection->config->name.data);
+        LIST_FOREACH(readerGroup, &connection->readerGroups, listEntry){
+            readerGroupCount++;
+            char* expectedRgName = "ReaderGroup1";
+            ck_assert_str_eq(expectedRgName, (char*)readerGroup->config.name.data);
+            LIST_FOREACH(dataSetReader, &readerGroup->readers, listEntry){
+                dataSetReaderCount++;
+                char* expectedReaderName = "DataSet Reader 1";
+                ck_assert_str_eq(expectedReaderName, (char*)dataSetReader->config.name.data);
+            }
+        }
+    }
+    ck_assert_int_eq(connectionCount, 1);
+    ck_assert_int_eq(readerGroupCount, 1);
+    ck_assert_int_eq(dataSetReaderCount, 1);
+} END_TEST
+
+int main(void) {
+    TCase *tc_pubsub_file_configuration = tcase_create("File Configuration");
+    tcase_add_checked_fixture(tc_pubsub_file_configuration, setup, teardown);
+    tcase_add_test(tc_pubsub_file_configuration, AddPublisherUsingBinaryFile);
+    tcase_add_test(tc_pubsub_file_configuration, AddSubscriberUsingBinaryFile);
+
+    Suite *s = suite_create("PubSub file configuration");
+    suite_add_tcase(s, tc_pubsub_file_configuration);
+
+    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_pubsub.txt

@@ -40,3 +40,14 @@ JsonDataSetReaderMessageDataType
 TargetVariablesDataType
 FieldTargetDataType
 OverrideValueHandling
+PubSubConfigurationDataType
+PublishedDataSetDataType
+PublishedDataItemsDataType
+UABinaryFileDataType
+OverrideValueHandling
+PermissionType
+RolePermissionType
+SubscribedDataSetMirrorDataType
+PublishedEventsDataType
+DatagramConnectionTransportDataType
+DatagramWriterGroupTransportDataType