[FEATURE] Subscriber implementation

 - Add ReaderGroup, DataSetReader, TargetVariable functionality
 - Data filtered from DataSetReader gets updated in target variable
   available in information model
 - Mirrortype implementation not handled

Change-Id: I990b4552b979fdcc76c172a07f7b245fee7d03eb

examples/pubsub/tutorial_pubsub_subscribe.c

@@ -1,14 +1,15 @@
 /* This work is licensed under a Creative Commons CCZero 1.0 Universal License.
  * See http://creativecommons.org/publicdomain/zero/1.0/ for more information.
+ *
+ * Copyright (c) 2019 Kalycito Infotech Private Limited
  */
 
 /**
  * IMPORTANT ANNOUNCEMENT
- * The PubSub subscriber API is currently not finished. This examples can be used to
- * receive and print the values, which are published by the tutorial_pubsub_publish
- * example. The following code uses internal API which will be later replaced by the
- * higher-level PubSub subscriber API. */
-
+ * The PubSub Subscriber API is currently not finished. This example can be used
+ * to receive and display values that are published by tutorial_pubsub_publish
+ * example in the TargetVariables of Subscriber Information Model .
+ */
 #include <open62541/plugin/log_stdout.h>
 #include <open62541/plugin/pubsub_udp.h>
 #include <open62541/server.h>
@@ -16,7 +17,6 @@
 #include <open62541/types_generated.h>
 
 #include "ua_pubsub.h"
-#include "ua_pubsub_networkmessage.h"
 
 #ifdef UA_ENABLE_PUBSUB_ETH_UADP
 #include <open62541/plugin/pubsub_ethernet.h>
@@ -26,90 +26,172 @@
 #include <signal.h>
 #include <stdlib.h>
 
-UA_Boolean running = true;
-static void stopHandler(int sign) {
-    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "received ctrl-c");
-    running = false;
+UA_NodeId connectionIdentifier;
+UA_NodeId readerGroupIdentifier;
+UA_NodeId readerIdentifier;
+
+UA_DataSetReaderConfig readerConfig;
+
+static void fillTestDataSetMetaData(UA_DataSetMetaDataType *pMetaData);
+
+/* Add new connection to the server */
+static void
+addPubSubConnection(UA_Server *server, UA_String *transportProfile,
+                    UA_NetworkAddressUrlDataType *networkAddressUrl) {
+    if((server == NULL) && (transportProfile == NULL) &&
+        (networkAddressUrl == NULL)) {
+        return;
+    }
+
+    /* Configuration creation for the connection */
+    UA_PubSubConnectionConfig connectionConfig;
+    memset (&connectionConfig, 0, sizeof(UA_PubSubConnectionConfig));
+    connectionConfig.name = UA_STRING("UDPMC Connection 1");
+    connectionConfig.transportProfileUri = *transportProfile;
+    connectionConfig.enabled = UA_TRUE;
+    UA_Variant_setScalar(&connectionConfig.address, networkAddressUrl,
+                          &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]);
+    connectionConfig.publisherId.numeric = UA_UInt32_random ();
+    UA_Server_addPubSubConnection (server, &connectionConfig, &connectionIdentifier);
 }
 
+/* Add ReaderGroup to the created connection */
 static void
-subscriptionPollingCallback(UA_Server *server, UA_PubSubConnection *connection) {
-    UA_ByteString buffer;
-    if (UA_ByteString_allocBuffer(&buffer, 512) != UA_STATUSCODE_GOOD) {
-        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
-                     "Message buffer allocation failed!");
+addReaderGroup(UA_Server *server) {
+    if(server == NULL) {
         return;
     }
 
-    /* Receive the message. Blocks for 5ms */
-    UA_StatusCode retval =
-        connection->channel->receive(connection->channel, &buffer, NULL, 5);
-    if(retval != UA_STATUSCODE_GOOD || buffer.length == 0) {
-        /* Workaround!! Reset buffer length. Receive can set the length to zero.
-         * Then the buffer is not deleted because no memory allocation is
-         * assumed.
-         * TODO: Return an error code in 'receive' instead of setting the buf
-         * length to zero. */
-        buffer.length = 512;
-        UA_ByteString_clear(&buffer);
+    UA_ReaderGroupConfig readerGroupConfig;
+    memset (&readerGroupConfig, 0, sizeof(UA_ReaderGroupConfig));
+    readerGroupConfig.name = UA_STRING("ReaderGroup1");
+    UA_Server_addReaderGroup(server, connectionIdentifier, &readerGroupConfig,
+                                        &readerGroupIdentifier);
+}
+
+/* Add DataSetReader to the ReaderGroup */
+static void
+addDataSetReader(UA_Server *server) {
+    if(server == NULL) {
         return;
     }
 
-    /* Decode the message */
-    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
-                "Message length: %lu", (unsigned long) buffer.length);
-    UA_NetworkMessage networkMessage;
-    memset(&networkMessage, 0, sizeof(UA_NetworkMessage));
-    size_t currentPosition = 0;
-    UA_NetworkMessage_decodeBinary(&buffer, &currentPosition, &networkMessage);
-    UA_ByteString_clear(&buffer);
-
-    /* Is this the correct message type? */
-    if(networkMessage.networkMessageType != UA_NETWORKMESSAGE_DATASET)
-        goto cleanup;
-
-    /* At least one DataSetMessage in the NetworkMessage? */
-    if(networkMessage.payloadHeaderEnabled &&
-       networkMessage.payloadHeader.dataSetPayloadHeader.count < 1)
-        goto cleanup;
-
-    /* Is this a KeyFrame-DataSetMessage? */
-    UA_DataSetMessage *dsm = &networkMessage.payload.dataSetPayload.dataSetMessages[0];
-    if(dsm->header.dataSetMessageType != UA_DATASETMESSAGE_DATAKEYFRAME)
-        goto cleanup;
-
-    /* Loop over the fields and print well-known content types */
-    for(int i = 0; i < dsm->data.keyFrameData.fieldCount; i++) {
-        const UA_DataType *currentType = dsm->data.keyFrameData.dataSetFields[i].value.type;
-        if(currentType == &UA_TYPES[UA_TYPES_BYTE]) {
-            UA_Byte value = *(UA_Byte *)dsm->data.keyFrameData.dataSetFields[i].value.data;
-            UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
-                        "Message content: [Byte] \tReceived data: %i", value);
-        } else if (currentType == &UA_TYPES[UA_TYPES_DATETIME]) {
-            UA_DateTime value = *(UA_DateTime *)dsm->data.keyFrameData.dataSetFields[i].value.data;
-            UA_DateTimeStruct receivedTime = UA_DateTime_toStruct(value);
-            UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
-                        "Message content: [DateTime] \t"
-                        "Received date: %02i-%02i-%02i Received time: %02i:%02i:%02i",
-                        receivedTime.year, receivedTime.month, receivedTime.day,
-                        receivedTime.hour, receivedTime.min, receivedTime.sec);
-        }
+    memset (&readerConfig, 0, sizeof(UA_DataSetReaderConfig));
+    readerConfig.name = UA_STRING("DataSet Reader 1");
+    readerConfig.dataSetWriterId = 1;
+
+    /* Setting up Meta data configuration in DataSetReader */
+    fillTestDataSetMetaData(&readerConfig.dataSetMetaData);
+    UA_Server_addDataSetReader(server, readerGroupIdentifier, &readerConfig,
+                                          &readerIdentifier);
+}
+
+/* Set SubscribedDataSet type to TargetVariables data type
+ * Add subscribedvariables to the DataSetReader */
+static void addSubscribedVariables (UA_Server *server, UA_NodeId dataSetReaderId) {
+    if(server == NULL) {
+        return;
+    }
+
+    UA_NodeId folderId;
+    UA_String folderName = readerConfig.dataSetMetaData.name;
+    UA_ObjectAttributes oAttr = UA_ObjectAttributes_default;
+    UA_QualifiedName folderBrowseName;
+    if(folderName.length > 0) {
+        oAttr.displayName.locale = UA_STRING ("en-US");
+        oAttr.displayName.text = folderName;
+        folderBrowseName.namespaceIndex = 1;
+        folderBrowseName.name = folderName;
+    }
+    else {
+        oAttr.displayName = UA_LOCALIZEDTEXT ("en-US", "Subscribed Variables");
+        folderBrowseName = UA_QUALIFIEDNAME (1, "Subscribed Variables");
+    }
+
+    UA_Server_addObjectNode (server, UA_NODEID_NULL,
+                             UA_NODEID_NUMERIC (0, UA_NS0ID_OBJECTSFOLDER),
+                             UA_NODEID_NUMERIC (0, UA_NS0ID_ORGANIZES),
+                             folderBrowseName, UA_NODEID_NUMERIC (0,
+                             UA_NS0ID_BASEOBJECTTYPE), oAttr, NULL, &folderId);
+
+    UA_Server_DataSetReader_addTargetVariables (server, &folderId,
+                                               dataSetReaderId,
+                                               UA_PUBSUB_SDS_TARGET);
+}
+
+/* Define MetaData for TargetVariables */
+static void fillTestDataSetMetaData(UA_DataSetMetaDataType *pMetaData) {
+    if(pMetaData == NULL) {
+        return;
     }
 
- cleanup:
-    UA_NetworkMessage_clear(&networkMessage);
+    UA_DataSetMetaDataType_init (pMetaData);
+    UA_String strTmp = UA_STRING ("DataSet 1");
+    UA_String_copy (&strTmp, &pMetaData->name);
+
+    /* Static definition of number of fields size to 4 to create four different
+     * targetVariables of distinct datatype
+     * Currently the publisher sends only DateTime data type */
+    pMetaData->fieldsSize = 4;
+    pMetaData->fields = (UA_FieldMetaData*)UA_Array_new (pMetaData->fieldsSize,
+                         &UA_TYPES[UA_TYPES_FIELDMETADATA]);
+
+    /* DateTime DataType */
+    UA_FieldMetaData_init (&pMetaData->fields[0]);
+    UA_NodeId_copy (&UA_TYPES[UA_TYPES_DATETIME].typeId,
+                    &pMetaData->fields[0].dataType);
+    pMetaData->fields[0].builtInType = UA_NS0ID_DATETIME;
+    strTmp = UA_STRING ("DateTime");
+    UA_String_copy (&strTmp, &pMetaData->fields[0].name);
+    pMetaData->fields[0].valueRank = -1; /* scalar */
+
+    /* Int32 DataType */
+    UA_FieldMetaData_init (&pMetaData->fields[1]);
+    UA_NodeId_copy(&UA_TYPES[UA_TYPES_INT32].typeId,
+                   &pMetaData->fields[1].dataType);
+    pMetaData->fields[1].builtInType = UA_NS0ID_INT32;
+    strTmp = UA_STRING ("Int32");
+    UA_String_copy (&strTmp, &pMetaData->fields[1].name);
+    pMetaData->fields[1].valueRank = -1; /* scalar */
+
+    /* Int64 DataType */
+    UA_FieldMetaData_init (&pMetaData->fields[2]);
+    UA_NodeId_copy(&UA_TYPES[UA_TYPES_INT32].typeId,
+                   &pMetaData->fields[2].dataType);
+    pMetaData->fields[2].builtInType = UA_NS0ID_INT32;
+    strTmp = UA_STRING ("Int32Fast");
+    UA_String_copy (&strTmp, &pMetaData->fields[2].name);
+    pMetaData->fields[2].valueRank = -1; /* scalar */
+
+    /* Boolean DataType */
+    UA_FieldMetaData_init (&pMetaData->fields[3]);
+    UA_NodeId_copy (&UA_TYPES[UA_TYPES_BOOLEAN].typeId,
+                    &pMetaData->fields[3].dataType);
+    pMetaData->fields[3].builtInType = UA_NS0ID_BOOLEAN;
+    strTmp = UA_STRING ("BoolToggle");
+    UA_String_copy (&strTmp, &pMetaData->fields[3].name);
+    pMetaData->fields[3].valueRank = -1; /* scalar */
+}
+
+UA_Boolean running = true;
+static void stopHandler(int sign) {
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "received ctrl-c");
+    running = false;
 }
 
 static int
 run(UA_String *transportProfile, UA_NetworkAddressUrlDataType *networkAddressUrl) {
     signal(SIGINT, stopHandler);
     signal(SIGTERM, stopHandler);
-
+    /* Return value initialized to Status Good */
+    UA_StatusCode retval;
     UA_Server *server = UA_Server_new();
     UA_ServerConfig *config = UA_Server_getConfig(server);
     UA_ServerConfig_setMinimal(config, 4801, NULL);
 
-    /* Details about the PubSubTransportLayer can be found inside the
+    /* Add the PubSub network layer implementation to the server config.
+     * The TransportLayer is acting as factory to create new connections
+     * on runtime. Details about the PubSubTransportLayer can be found inside the
      * tutorial_pubsub_connection */
     config->pubsubTransportLayers = (UA_PubSubTransportLayer *)
         UA_calloc(2, sizeof(UA_PubSubTransportLayer));
@@ -117,6 +199,7 @@ run(UA_String *transportProfile, UA_NetworkAddressUrlDataType *networkAddressUrl
         UA_Server_delete(server);
         return EXIT_FAILURE;
     }
+
     config->pubsubTransportLayers[0] = UA_PubSubTransportLayerUDPMP();
     config->pubsubTransportLayersSize++;
 #ifdef UA_ENABLE_PUBSUB_ETH_UADP
@@ -124,75 +207,47 @@ run(UA_String *transportProfile, UA_NetworkAddressUrlDataType *networkAddressUrl
     config->pubsubTransportLayersSize++;
 #endif
 
-    UA_PubSubConnectionConfig connectionConfig;
-    memset(&connectionConfig, 0, sizeof(connectionConfig));
-    connectionConfig.name = UA_STRING("UADP Connection 1");
-    connectionConfig.transportProfileUri = *transportProfile;
-    connectionConfig.enabled = UA_TRUE;
-    UA_Variant_setScalar(&connectionConfig.address, networkAddressUrl,
-                         &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]);
-    UA_NodeId connectionIdent;
-    UA_StatusCode retval =
-        UA_Server_addPubSubConnection(server, &connectionConfig, &connectionIdent);
-    if(retval == UA_STATUSCODE_GOOD)
-        UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
-                    "The PubSub Connection was created successfully!");
-
-    /* The following lines register the listening on the configured multicast
-     * address and configure a repeated job, which is used to handle received
-     * messages. */
-    UA_PubSubConnection *connection =
-        UA_PubSubConnection_findConnectionbyId(server, connectionIdent);
-    if(connection != NULL) {
-        UA_StatusCode rv = connection->channel->regist(connection->channel, NULL, NULL);
-        if (rv == UA_STATUSCODE_GOOD) {
-            UA_UInt64 subscriptionCallbackId;
-            UA_Server_addRepeatedCallback(server, (UA_ServerCallback)subscriptionPollingCallback,
-                                          connection, 100, &subscriptionCallbackId);
-        } else {
-            UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "register channel failed: %s!",
-                           UA_StatusCode_name(rv));
-        }
-    }
-
-    retval |= UA_Server_run(server, &running);
+    /* API calls */
+    addPubSubConnection(server, transportProfile, networkAddressUrl);
+    addReaderGroup(server);
+    addDataSetReader(server);
+    addSubscribedVariables(server, readerIdentifier);
 
+    retval = UA_Server_run(server, &running);
     UA_Server_delete(server);
-    return retval == UA_STATUSCODE_GOOD ? EXIT_SUCCESS : EXIT_FAILURE;;
+    return retval == UA_STATUSCODE_GOOD ? EXIT_SUCCESS : EXIT_FAILURE;
 }
 
-
 static void
 usage(char *progname) {
     printf("usage: %s <uri> [device]\n", progname);
 }
 
 int main(int argc, char **argv) {
-    UA_String transportProfile =
-        UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-udp-uadp");
-    UA_NetworkAddressUrlDataType networkAddressUrl =
-        {UA_STRING_NULL , UA_STRING("opc.udp://224.0.0.22:4840/")};
-
-    if (argc > 1) {
-        if (strcmp(argv[1], "-h") == 0) {
+    UA_String transportProfile = UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-udp-uadp");
+    UA_NetworkAddressUrlDataType networkAddressUrl = {UA_STRING_NULL , UA_STRING("opc.udp://224.0.0.22:4840/")};
+    if(argc > 1) {
+        if(strcmp(argv[1], "-h") == 0) {
             usage(argv[0]);
             return EXIT_SUCCESS;
-        } else if (strncmp(argv[1], "opc.udp://", 10) == 0) {
+        } else if(strncmp(argv[1], "opc.udp://", 10) == 0) {
             networkAddressUrl.url = UA_STRING(argv[1]);
-        } else if (strncmp(argv[1], "opc.eth://", 10) == 0) {
+        } else if(strncmp(argv[1], "opc.eth://", 10) == 0) {
             transportProfile =
                 UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-eth-uadp");
-            if (argc < 3) {
+            if(argc < 3) {
                 printf("Error: UADP/ETH needs an interface name\n");
                 return EXIT_FAILURE;
             }
+
             networkAddressUrl.networkInterface = UA_STRING(argv[2]);
             networkAddressUrl.url = UA_STRING(argv[1]);
         } else {
-            printf("Error: unknown URI\n");
+            printf ("Error: unknown URI\n");
             return EXIT_FAILURE;
         }
     }
 
     return run(&transportProfile, &networkAddressUrl);
 }
+

include/open62541/server_pubsub.h

@@ -3,6 +3,7 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  *
  * Copyright (c) 2017-2018 Fraunhofer IOSB (Author: Andreas Ebner)
+ * Copyright (c) 2019 Kalycito Infotech Private Limited
  */
 
 #ifndef UA_SERVER_PUBSUB_H
@@ -238,8 +239,8 @@ typedef enum {
 typedef struct {
     UA_DataSetFieldType dataSetFieldType;
     union {
+        /* events need other config later */
         UA_DataSetVariableConfig variable;
-        //events need other config later
     } field;
 } UA_DataSetFieldConfig;
 
@@ -366,6 +367,100 @@ UA_Server_getDataSetWriterConfig(UA_Server *server, const UA_NodeId dsw,
 UA_StatusCode UA_EXPORT
 UA_Server_removeDataSetWriter(UA_Server *server, const UA_NodeId dsw);
 
+/**
+ * DataSetReader
+ * -------------
+ * DataSetReader can receive NetworkMessages with the DataSet
+ * of interest sent by the Publisher. DataSetReaders represent
+ * the configuration necessary to receive and process DataSetMessages
+ * on the Subscriber side */
+
+/* Parameters for PubSubSecurity */
+typedef struct {
+    UA_Int32 securityMode;          /* placeholder datatype 'MessageSecurityMode' */
+    UA_String securityGroupId;
+    size_t keyServersSize;
+    UA_Int32 *keyServers;
+} UA_PubSubSecurityParameters;
+
+/* Parameters for PubSub DataSetReader Configuration */
+typedef struct {
+    UA_String name;
+    UA_Variant publisherId;
+    UA_UInt16 writerGroupId;
+    UA_UInt16 dataSetWriterId;
+    UA_DataSetMetaDataType dataSetMetaData;
+    UA_DataSetFieldContentMask dataSetFieldContentMask;
+    UA_Double messageReceiveTimeout;
+    UA_PubSubSecurityParameters securityParameters;
+    UA_UadpDataSetReaderMessageDataType messageSettings;
+    UA_TargetVariablesDataType subscribedDataSetTarget;
+} UA_DataSetReaderConfig;
+
+/* Update configuration to the dataSetReader */
+UA_StatusCode
+UA_Server_DataSetReader_updateConfig(UA_Server *server, UA_NodeId dataSetReaderIdentifier,
+                                   UA_NodeId readerGroupIdentifier, const UA_DataSetReaderConfig *config);
+
+/* Get configuration of the dataSetReader */
+UA_StatusCode
+UA_Server_DataSetReader_getConfig(UA_Server *server, UA_NodeId dataSetReaderIdentifier,
+                                 UA_DataSetReaderConfig *config);
+
+/* Return Status Code after creating TargetVariables in Subscriber AddressSpace
+ * TargetVariables define a list of variable mappings between received DataSet fields
+ * and the TargetVariables in the Subscriber AddressSpace */
+UA_StatusCode
+UA_Server_DataSetReader_createTargetVariables(UA_Server *server, UA_NodeId dataSetReaderIdentifier,
+                                             UA_TargetVariablesDataType* targetVariables);
+
+/* To Do:Implementation of SubscribedDataSetMirrorType
+ * UA_StatusCode
+ * A_PubSubDataSetReader_createDataSetMirror(UA_Server *server, UA_NodeId dataSetReaderIdentifier,
+ * UA_SubscribedDataSetMirrorDataType* mirror) */
+
+/**
+ * ReaderGroup
+ * -----------
+ * All ReaderGroups are created within a PubSubConnection and automatically
+ * deleted if the connection is removed. */
+
+/* ReaderGroup configuration */
+typedef struct {
+    UA_String name;
+    UA_PubSubSecurityParameters securityParameters;
+} UA_ReaderGroupConfig;
+
+/* Add DataSetReader to the ReaderGroup */
+UA_StatusCode
+UA_Server_addDataSetReader(UA_Server *server, UA_NodeId readerGroupIdentifier,
+                                      const UA_DataSetReaderConfig *dataSetReaderConfig,
+                                      UA_NodeId *readerIdentifier);
+
+/* Remove DataSetReader from ReaderGroup */
+UA_StatusCode
+UA_Server_removeDataSetReader(UA_Server *server, UA_NodeId readerIdentifier);
+
+/* To Do: Update Configuration of ReaderGroup */
+UA_StatusCode
+UA_Server_ReaderGroup_updateConfig(UA_Server *server, UA_NodeId readerGroupIdentifier,
+                                  const UA_ReaderGroupConfig *config);
+
+/* Get configuraiton of ReaderGroup */
+UA_StatusCode
+UA_Server_ReaderGroup_getConfig(UA_Server *server, UA_NodeId readerGroupIdentifier,
+                               UA_ReaderGroupConfig *config);
+
+/* Add ReaderGroup to the created connection */
+UA_StatusCode
+UA_Server_addReaderGroup(UA_Server *server, UA_NodeId connectionIdentifier,
+                                   const UA_ReaderGroupConfig *readerGroupConfig,
+                                   UA_NodeId *readerGroupIdentifier);
+
+/* Remove ReaderGroup from connection */
+UA_StatusCode
+UA_Server_removeReaderGroup(UA_Server *server, UA_NodeId groupIdentifier);
+
 #endif /* UA_ENABLE_PUBSUB */
 
 _UA_END_DECLS

src/pubsub/ua_pubsub.h

@@ -3,6 +3,7 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  *
  * Copyright (c) 2017-2018 Fraunhofer IOSB (Author: Andreas Ebner)
+ * Copyright (c) 2019 Kalycito Infotech Private Limited
  */
 
 #ifndef UA_PUBSUB_H_
@@ -19,10 +20,14 @@ _UA_BEGIN_DECLS
 
 #ifdef UA_ENABLE_PUBSUB /* conditional compilation */
 
-//forward declarations
+/* forward declarations */
 struct UA_WriterGroup;
 typedef struct UA_WriterGroup UA_WriterGroup;
 
+/* Declaration for ReaderGroup */
+struct UA_ReaderGroup;
+typedef struct UA_ReaderGroup UA_ReaderGroup;
+
 /* The configuration structs (public part of PubSub entities) are defined in include/ua_plugin_pubsub.h */
 
 /**********************************************/
@@ -54,6 +59,8 @@ typedef struct{
     UA_PubSubChannel *channel;
     UA_NodeId identifier;
     LIST_HEAD(UA_ListOfWriterGroup, UA_WriterGroup) writerGroups;
+    LIST_HEAD(UA_ListOfPubSubReaderGroup, UA_ReaderGroup) readerGroups;
+    size_t readerGroupsSize;
 } UA_PubSubConnection;
 
 UA_StatusCode
@@ -138,6 +145,74 @@ UA_DataSetFieldConfig_copy(const UA_DataSetFieldConfig *src, UA_DataSetFieldConf
 UA_DataSetField *
 UA_DataSetField_findDSFbyId(UA_Server *server, UA_NodeId identifier);
 
+/**********************************************/
+/*               DataSetReader                */
+/**********************************************/
+
+/* SubscribedDataSetDataType Definition */
+typedef enum {
+    UA_PUBSUB_SDS_TARGET,
+    UA_PUBSUB_SDS_MIRROR
+}UA_SubscribedDataSetEnumType;
+
+/* DataSetReader Type definition */
+typedef struct UA_DataSetReader {
+    UA_DataSetReaderConfig config;
+    /* implementation defined fields */
+    UA_NodeId identifier;
+    UA_NodeId linkedReaderGroup;
+    LIST_ENTRY(UA_DataSetReader) listEntry;
+    UA_SubscribedDataSetEnumType subscribedDataSetType;
+    UA_TargetVariablesDataType subscribedDataSetTarget;
+    /* To Do UA_SubscribedDataSetMirrorDataType subscribedDataSetMirror */
+}UA_DataSetReader;
+
+/* Delete DataSetReader */
+void UA_DataSetReader_delete(UA_Server *server, UA_DataSetReader *dataSetReader);
+
+/* Process Network Message using DataSetReader */
+void UA_Server_DataSetReader_process(UA_Server *server, UA_DataSetReader *dataSetReader, UA_DataSetMessage* dataSetMsg);
+
+/* Copy the configuration of DataSetReader */
+UA_StatusCode UA_DataSetReaderConfig_copy(const UA_DataSetReaderConfig *src, UA_DataSetReaderConfig *dst);
+
+/* Add TargetVariables */
+UA_StatusCode
+UA_Server_DataSetReader_addTargetVariables(UA_Server* server, UA_NodeId* parentNode, UA_NodeId dataSetReaderIdentifier, UA_SubscribedDataSetEnumType sdsType);
+
+/**********************************************/
+/*                ReaderGroup                 */
+/**********************************************/
+/* ReaderGroup Type Definition*/
+
+struct UA_ReaderGroup {
+    UA_ReaderGroupConfig config;
+    UA_NodeId identifier;
+    UA_NodeId linkedConnection;
+    LIST_ENTRY(UA_ReaderGroup) listEntry;
+    LIST_HEAD(UA_ListOfPubSubDataSetReader, UA_DataSetReader) readers;
+    /* for simplified information access */
+    UA_UInt32 readersCount;
+    UA_UInt64 subscribeCallbackId;
+    UA_Boolean subscribeCallbackIsRegistered;
+};
+
+/* Delete ReaderGroup */
+void UA_Server_ReaderGroup_delete(UA_Server *server, UA_ReaderGroup *readerGroup);
+
+/* Copy configuration of ReaderGroup */
+UA_StatusCode
+UA_ReaderGroupConfig_copy(const UA_ReaderGroupConfig *src, UA_ReaderGroupConfig *dst);
+
+/* Process Network Message */
+UA_StatusCode
+UA_Server_processNetworkMessage(UA_Server *server, UA_NetworkMessage* pMsg, UA_PubSubConnection *pConnection);
+
+/* Prototypes for internal util functions - some functions maybe removed later
+ *(currently moved from public to internal)*/
+UA_ReaderGroup *UA_ReaderGroup_findRGbyId(UA_Server *server, UA_NodeId identifier);
+UA_DataSetReader *UA_ReaderGroup_findDSRbyId(UA_Server *server, UA_NodeId identifier);
+
 /*********************************************************/
 /*               PublishValues handling                  */
 /*********************************************************/
@@ -147,6 +222,15 @@ UA_WriterGroup_addPublishCallback(UA_Server *server, UA_WriterGroup *writerGroup
 void
 UA_WriterGroup_publishCallback(UA_Server *server, UA_WriterGroup *writerGroup);
 
+/*********************************************************/
+/*               SubscribeValues handling                */
+/*********************************************************/
+
+UA_StatusCode
+UA_ReaderGroup_addSubscribeCallback(UA_Server *server, UA_ReaderGroup *readerGroup);
+void
+UA_ReaderGroup_subscribeCallback(UA_Server *server, UA_ReaderGroup *readerGroup);
+
 #endif /* UA_ENABLE_PUBSUB */
 
 _UA_END_DECLS

src/pubsub/ua_pubsub_ns0.c

@@ -3,6 +3,7 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  *
  * Copyright (c) 2017-2018 Fraunhofer IOSB (Author: Andreas Ebner)
+ * Copyright (c) 2019 Kalycito Infotech Private Limited
  */
 
 #include "ua_pubsub_ns0.h"
@@ -354,7 +355,7 @@ addPubSubConnectionAction(UA_Server *server,
         //UA_PubSubConnection_addWriterGroup(server, UA_NODEID_NULL, NULL, NULL);
     }
     for(size_t i = 0; i < pubSubConnectionDataType.readerGroupsSize; i++){
-        //UA_PubSubConnection_addReaderGroup(server, NULL, NULL, NULL);
+        //UA_Server_addReaderGroup(server, NULL, NULL, NULL);
     }
     UA_NetworkAddressUrlDataType_deleteMembers(&networkAddressUrlDataType);
     //set ouput value
@@ -398,6 +399,49 @@ removeConnectionAction(UA_Server *server,
 }
 #endif
 
+/**********************************************/
+/*               DataSetReader                */
+/**********************************************/
+UA_StatusCode
+addDataSetReaderRepresentation(UA_Server *server, UA_DataSetReader *dataSetReader){
+    //TODO implement reader part
+    return UA_STATUSCODE_BADNOTIMPLEMENTED;
+}
+
+#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS
+static UA_StatusCode
+addDataSetReaderAction(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_StatusCode retVal = UA_STATUSCODE_BADNOTIMPLEMENTED;
+    //TODO implement reader part
+    return retVal;
+}
+#endif
+
+UA_StatusCode
+removeDataSetReaderRepresentation(UA_Server *server, UA_DataSetReader* dataSetReader){
+    //TODO implement reader part
+    return UA_STATUSCODE_BADNOTIMPLEMENTED;
+}
+
+#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS
+static UA_StatusCode
+removeDataSetReaderAction(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_StatusCode retVal = UA_STATUSCODE_BADNOTIMPLEMENTED;
+    //TODO implement reader part
+    return retVal;
+}
+#endif
+
 /*************************************************/
 /*                PublishedDataSet               */
 /*************************************************/
@@ -798,6 +842,29 @@ removeGroupAction(UA_Server *server,
 }
 #endif
 
+/**********************************************/
+/*               ReaderGroup                  */
+/**********************************************/
+UA_StatusCode
+addReaderGroupRepresentation(UA_Server *server, UA_ReaderGroup *readerGroup){
+    //TODO implement reader part
+    return UA_STATUSCODE_BADNOTIMPLEMENTED;
+}
+
+#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL_METHODS
+static UA_StatusCode
+addReaderGroupAction(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_StatusCode retVal = UA_STATUSCODE_GOOD;
+    //TODO implement reader part
+    return retVal;
+}
+#endif
+
 /**********************************************/
 /*               DataSetWriter                */
 /**********************************************/
@@ -921,6 +988,14 @@ writerGroupTypeDestructor(UA_Server *server,
     }
 }
 
+static void
+readerGroupTypeDestructor(UA_Server *server,
+                          const UA_NodeId *sessionId, void *sessionContext,
+                          const UA_NodeId *typeId, void *typeContext,
+                          const UA_NodeId *nodeId, void **nodeContext) {
+    UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_USERLAND, "ReaderGroup destructor called!");
+}
+
 static void
 dataSetWriterTypeDestructor(UA_Server *server,
                             const UA_NodeId *sessionId, void *sessionContext,
@@ -929,6 +1004,14 @@ dataSetWriterTypeDestructor(UA_Server *server,
     UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_USERLAND, "DataSetWriter destructor called!");
 }
 
+static void
+dataSetReaderTypeDestructor(UA_Server *server,
+                            const UA_NodeId *sessionId, void *sessionContext,
+                            const UA_NodeId *typeId, void *typeContext,
+                            const UA_NodeId *nodeId, void **nodeContext) {
+    UA_LOG_INFO(&server->config.logger, UA_LOGCATEGORY_USERLAND, "DataSetReader destructor called!");
+}
+
 static void
 publishedDataItemsTypeDestructor(UA_Server *server,
                             const UA_NodeId *sessionId, void *sessionContext,
@@ -985,9 +1068,12 @@ UA_Server_initPubSubNS0(UA_Server *server) {
     retVal |= UA_Server_setMethodNode_callback(server,
                                                UA_NODEID_NUMERIC(0, UA_NS0ID_PUBLISHEDDATAITEMSTYPE_REMOVEVARIABLES), removeVariablesAction);
     retVal |= UA_Server_setMethodNode_callback(server, UA_NODEID_NUMERIC(0, UA_NS0ID_PUBSUBCONNECTIONTYPE_ADDWRITERGROUP), addWriterGroupAction);
+    retVal |= UA_Server_setMethodNode_callback(server, UA_NODEID_NUMERIC(0, UA_NS0ID_PUBSUBCONNECTIONTYPE_ADDREADERGROUP), addReaderGroupAction);
     retVal |= UA_Server_setMethodNode_callback(server, UA_NODEID_NUMERIC(0, UA_NS0ID_PUBSUBCONNECTIONTYPE_REMOVEGROUP), removeGroupAction);
     retVal |= UA_Server_setMethodNode_callback(server, UA_NODEID_NUMERIC(0, UA_NS0ID_WRITERGROUPTYPE_ADDDATASETWRITER), addDataSetWriterAction);
     retVal |= UA_Server_setMethodNode_callback(server, UA_NODEID_NUMERIC(0, UA_NS0ID_WRITERGROUPTYPE_REMOVEDATASETWRITER), removeDataSetWriterAction);
+    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);
 
 #else
     retVal |= UA_Server_deleteReference(server, UA_NODEID_NUMERIC(0, UA_NS0ID_PUBLISHSUBSCRIBE), UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT), true,
@@ -1003,10 +1089,14 @@ UA_Server_initPubSubNS0(UA_Server *server) {
     UA_Server_setNodeTypeLifecycle(server, UA_NODEID_NUMERIC(0, UA_NS0ID_PUBSUBCONNECTIONTYPE), liveCycle);
     liveCycle.destructor = writerGroupTypeDestructor;
     UA_Server_setNodeTypeLifecycle(server, UA_NODEID_NUMERIC(0, UA_NS0ID_WRITERGROUPTYPE), liveCycle);
+    liveCycle.destructor = readerGroupTypeDestructor;
+    UA_Server_setNodeTypeLifecycle(server, UA_NODEID_NUMERIC(0, UA_NS0ID_READERGROUPTYPE), liveCycle);
     liveCycle.destructor = dataSetWriterTypeDestructor;
     UA_Server_setNodeTypeLifecycle(server, UA_NODEID_NUMERIC(0, UA_NS0ID_DATASETWRITERDATATYPE), liveCycle);
     liveCycle.destructor = publishedDataItemsTypeDestructor;
     UA_Server_setNodeTypeLifecycle(server, UA_NODEID_NUMERIC(0, UA_NS0ID_PUBLISHEDDATAITEMSTYPE), liveCycle);
+    liveCycle.destructor = dataSetReaderTypeDestructor;
+    UA_Server_setNodeTypeLifecycle(server, UA_NODEID_NUMERIC(0, UA_NS0ID_DATASETREADERDATATYPE), liveCycle);
 
     return retVal;
 }

src/pubsub/ua_pubsub_ns0.h

@@ -3,6 +3,7 @@
  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
  *
  * Copyright (c) 2017-2018 Fraunhofer IOSB (Author: Andreas Ebner)
+ * Copyright (c) 2019 Kalycito Infotech Private Limited
  */
 
 #ifndef UA_PUBSUB_NS0_H_
@@ -27,6 +28,9 @@ removePubSubConnectionRepresentation(UA_Server *server, UA_PubSubConnection *con
 UA_StatusCode
 addWriterGroupRepresentation(UA_Server *server, UA_WriterGroup *writerGroup);
 
+UA_StatusCode
+addReaderGroupRepresentation(UA_Server *server, UA_ReaderGroup *readerGroup);
+
 UA_StatusCode
 removeGroupRepresentation(UA_Server *server, UA_WriterGroup *writerGroup);
 
@@ -42,6 +46,12 @@ addPublishedDataItemsRepresentation(UA_Server *server, UA_PublishedDataSet *publ
 UA_StatusCode
 removePublishedDataSetRepresentation(UA_Server *server, UA_PublishedDataSet *publishedDataSet);
 
+UA_StatusCode
+addDataSetReaderRepresentation(UA_Server *server, UA_DataSetReader *dataSetReader);
+
+UA_StatusCode
+removeDataSetReaderRepresentation(UA_Server *server, UA_DataSetReader *dataSetReader);
+
 #endif /* UA_ENABLE_PUBSUB_INFORMATIONMODEL */
 
 _UA_END_DECLS

tools/schema/datatypes_minimal.txt

@@ -107,12 +107,12 @@ RegisterNodesRequest
 RegisterNodesResponse
 UnregisterNodesRequest
 UnregisterNodesResponse
-ServerState
-ServerStatusDataType
-BuildInfo
 Duration
 UtcTime
 LocaleId
-RedundancySupport
-ServerDiagnosticsSummaryDataType
 EnumValueType
+BuildInfo
+ServerStatusDataType
+ServerState
+ServerDiagnosticsSummaryDataType
+RedundancySupport

tools/schema/datatypes_pubsub.txt

@@ -33,3 +33,7 @@ BrokerConnectionTransportDataType
 BrokerWriterGroupTransportDataType
 BrokerDataSetWriterTransportDataType
 BrokerWriterGroupTransportType
+UadpDataSetReaderMessageDataType
+TargetVariablesDataType
+FieldTargetDataType
+OverrideValueHandling