ua_pubsub.h

Go to the documentation of this file.

/** This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 * Copyright (c) 2017-2019 Fraunhofer IOSB (Author: Andreas Ebner)
 * Copyright (c) 2019 Kalycito Infotech Private Limited
 * Copyright (c) 2020 Yannick Wallerer, Siemens AG
 * Copyright (c) 2020, 2022 Thomas Fischer, Siemens AG
 * Copyright (c) 2021 Fraunhofer IOSB (Author: Jan Hermes)
 * Copyright (c) 2022 Siemens AG (Author: Thomas Fischer)
 * Copyright (c) 2022 Fraunhofer IOSB (Author: Noel Graf)
 * Copyright (c) 2022 Linutronix GmbH (Author: Muddasir Shakil)
 */
 
#ifndef UA_PUBSUB_H_
#define UA_PUBSUB_H_
 
#define UA_INTERNAL
#include <open62541/server.h>
#include <open62541/server_pubsub.h>
 
#include "open62541_queue.h"
#include "ziptree.h"
#include "ua_pubsub_networkmessage.h"
 
#ifdef UA_ENABLE_PUBSUB_SKS
#include "ua_pubsub_keystorage.h"
#endif
 
_UA_BEGIN_DECLS
 
#ifdef UA_ENABLE_PUBSUB
 
/** Max number of underlying for sending and receiving */
#define UA_PUBSUB_MAXCHANNELS 8
 
struct UA_WriterGroup;
typedef struct UA_WriterGroup UA_WriterGroup;
 
struct UA_ReaderGroup;
typedef struct UA_ReaderGroup UA_ReaderGroup;
 
struct UA_SecurityGroup;
typedef struct UA_SecurityGroup UA_SecurityGroup;
 
 
/**            PublishedDataSet                */
 
 
typedef struct UA_PublishedDataSet {
    UA_PublishedDataSetConfig config;
    UA_DataSetMetaDataType dataSetMetaData;
    TAILQ_HEAD(, UA_DataSetField) fields;
    UA_UInt16 fieldSize;
    UA_NodeId identifier;
    UA_UInt16 promotedFieldsCount;
    UA_UInt16 configurationFreezeCounter;
    TAILQ_ENTRY(UA_PublishedDataSet) listEntry;
} 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);
 
UA_PublishedDataSet *
UA_PublishedDataSet_findPDSbyName(UA_Server *server, UA_String name);
 
UA_AddPublishedDataSetResult
UA_PublishedDataSet_create(UA_Server *server,
                           const UA_PublishedDataSetConfig *publishedDataSetConfig,
                           UA_NodeId *pdsIdentifier);
 
void
UA_PublishedDataSet_clear(UA_Server *server,
                          UA_PublishedDataSet *publishedDataSet);
 
UA_StatusCode
UA_PublishedDataSet_remove(UA_Server *server, UA_PublishedDataSet *publishedDataSet);
 
UA_StatusCode
getPublishedDataSetConfig(UA_Server *server, const UA_NodeId pds,
                          UA_PublishedDataSetConfig *config);
 
typedef struct UA_StandaloneSubscribedDataSet{
    UA_StandaloneSubscribedDataSetConfig config;
    UA_NodeId identifier;
    TAILQ_ENTRY(UA_StandaloneSubscribedDataSet) listEntry;
    UA_NodeId connectedReader;
} UA_StandaloneSubscribedDataSet;
 
UA_StatusCode
UA_StandaloneSubscribedDataSetConfig_copy(const UA_StandaloneSubscribedDataSetConfig *src,
                                          UA_StandaloneSubscribedDataSetConfig *dst);
UA_StandaloneSubscribedDataSet *
UA_StandaloneSubscribedDataSet_findSDSbyId(UA_Server *server, UA_NodeId identifier);
UA_StandaloneSubscribedDataSet *
UA_StandaloneSubscribedDataSet_findSDSbyName(UA_Server *server, UA_String identifier);
void
UA_StandaloneSubscribedDataSet_clear(UA_Server *server,
                                     UA_StandaloneSubscribedDataSet *subscribedDataSet);
 
#define UA_LOG_PDS_INTERNAL(LOGGER, LEVEL, PDS, MSG, ...)               \
    if(UA_LOGLEVEL <= UA_LOGLEVEL_##LEVEL) {                            \
        UA_String idStr = UA_STRING_NULL;                               \
        if(PDS)                                                         \
            UA_NodeId_print(&(PDS)->identifier, &idStr);                \
        UA_LOG_##LEVEL(LOGGER, UA_LOGCATEGORY_PUBSUB,                   \
                       "DataSet %.*s\t| " MSG "%.0s", (int)idStr.length, \
                       (char*)idStr.data, __VA_ARGS__);                 \
        UA_String_clear(&idStr);                                        \
    }
 
#define UA_LOG_TRACE_DATASET(LOGGER, PDS, ...)                          \
    UA_MACRO_EXPAND(UA_LOG_PDS_INTERNAL(LOGGER, TRACE, PDS, __VA_ARGS__, ""))
#define UA_LOG_DEBUG_DATASET(LOGGER, PDS, ...)                          \
    UA_MACRO_EXPAND(UA_LOG_PDS_INTERNAL(LOGGER, DEBUG, PDS, __VA_ARGS__, ""))
#define UA_LOG_INFO_DATASET(LOGGER, PDS, ...)                           \
    UA_MACRO_EXPAND(UA_LOG_PDS_INTERNAL(LOGGER, INFO, PDS, __VA_ARGS__, ""))
#define UA_LOG_WARNING_DATASET(LOGGER, PDS, ...)                        \
    UA_MACRO_EXPAND(UA_LOG_PDS_INTERNAL(LOGGER, WARNING, PDS, __VA_ARGS__, ""))
#define UA_LOG_ERROR_DATASET(LOGGER, PDS, ...)                          \
    UA_MACRO_EXPAND(UA_LOG_PDS_INTERNAL(LOGGER, ERROR, PDS, __VA_ARGS__, ""))
#define UA_LOG_FATAL_DATASET(LOGGER, PDS, ...)                          \
    UA_MACRO_EXPAND(UA_LOG_PDS_INTERNAL(LOGGER, FATAL, PDS, __VA_ARGS__, ""))
 
 
/**               Connection                   */
 
 
typedef struct UA_PubSubConnection {
    UA_PubSubComponentEnumType componentType;
 
    TAILQ_ENTRY(UA_PubSubConnection) listEntry;
    UA_NodeId identifier;
 
    /* The send/recv connections are only opened if the state is operational */
    UA_PubSubState state;
    UA_PubSubConnectionConfig config;
    UA_Boolean json; /* Extracted from the TransportProfileUrl */
 
    /* Channels belonging to the PubSubConnection. Send channels belong to
     * WriterGroups, recv channels belong to ReaderGroups. We only open channels 
     * if there is at least one WriterGroup/ReaderGroup respectively.
     *
     * Some channels belong exclusively to just one WriterGroup/ReaderGroup that
     * defines additional connection properties. For example an MQTT topic name
     * or QoS parameters. In that case a dedicated NetworkCallback is used that
     * takes this ReaderGroup/WriterGroup directly as context. */
    UA_ConnectionManager *cm;
    uintptr_t recvChannels[UA_PUBSUB_MAXCHANNELS];
    size_t recvChannelsSize;
    uintptr_t sendChannel;
 
    size_t writerGroupsSize;
    LIST_HEAD(, UA_WriterGroup) writerGroups;
 
    size_t readerGroupsSize;
    LIST_HEAD(, UA_ReaderGroup) readerGroups;
 
    UA_UInt16 configurationFreezeCounter;
 
    UA_DateTime silenceErrorUntil; /* Avoid generating too many logs */
 
    UA_Boolean deleteFlag; /* To be deleted - in addition to the PubSubState */
    UA_DelayedCallback dc; /* For delayed freeing */
} UA_PubSubConnection;
 
UA_StatusCode
UA_PubSubConnectionConfig_copy(const UA_PubSubConnectionConfig *src,
                               UA_PubSubConnectionConfig *dst);
 
UA_PubSubConnection *
UA_PubSubConnection_findConnectionbyId(UA_Server *server,
                                       UA_NodeId connectionIdentifier);
 
UA_StatusCode
UA_PubSubConnection_create(UA_Server *server,
                           const UA_PubSubConnectionConfig *connectionConfig,
                           UA_NodeId *connectionIdentifier);
 
void
UA_PubSubConnectionConfig_clear(UA_PubSubConnectionConfig *connectionConfig);
 
void
UA_PubSubConnection_delete(UA_Server *server, UA_PubSubConnection *c);
 
UA_StatusCode
UA_PubSubConnection_connect(UA_Server *server, UA_PubSubConnection *c,
                            UA_Boolean validate);
 
void
UA_PubSubConnection_disconnect(UA_PubSubConnection *c);
 
/** Returns either the eventloop configured in the connection or, in its absence,
 * for the server */
UA_EventLoop *
UA_PubSubConnection_getEL(UA_Server *server, UA_PubSubConnection *c);
 
UA_StatusCode
UA_PubSubConnection_setPubSubState(UA_Server *server,
                                   UA_PubSubConnection *connection,
                                   UA_PubSubState state,
                                   UA_StatusCode cause);
 
#define UA_LOG_CONNECTION_INTERNAL(LOGGER, LEVEL, CONNECTION, MSG, ...) \
    if(UA_LOGLEVEL <= UA_LOGLEVEL_##LEVEL) {                            \
        UA_String idStr = UA_STRING_NULL;                               \
        if(CONNECTION)                                                  \
            UA_NodeId_print(&(CONNECTION)->identifier, &idStr);         \
        UA_LOG_##LEVEL(LOGGER, UA_LOGCATEGORY_PUBSUB,                   \
                       "Connection %.*s\t| " MSG "%.0s", (int)idStr.length, \
                       (char*)idStr.data, __VA_ARGS__);                 \
        UA_String_clear(&idStr);                                        \
    }
 
#define UA_LOG_TRACE_CONNECTION(LOGGER, CONNECTION, ...)                \
    UA_MACRO_EXPAND(UA_LOG_CONNECTION_INTERNAL(LOGGER, TRACE, CONNECTION, __VA_ARGS__, ""))
#define UA_LOG_DEBUG_CONNECTION(LOGGER, CONNECTION, ...)                \
    UA_MACRO_EXPAND(UA_LOG_CONNECTION_INTERNAL(LOGGER, DEBUG, CONNECTION, __VA_ARGS__, ""))
#define UA_LOG_INFO_CONNECTION(LOGGER, CONNECTION, ...)                 \
    UA_MACRO_EXPAND(UA_LOG_CONNECTION_INTERNAL(LOGGER, INFO, CONNECTION, __VA_ARGS__, ""))
#define UA_LOG_WARNING_CONNECTION(LOGGER, CONNECTION, ...)              \
    UA_MACRO_EXPAND(UA_LOG_CONNECTION_INTERNAL(LOGGER, WARNING, CONNECTION, __VA_ARGS__, ""))
#define UA_LOG_ERROR_CONNECTION(LOGGER, CONNECTION, ...)                \
    UA_MACRO_EXPAND(UA_LOG_CONNECTION_INTERNAL(LOGGER, ERROR, CONNECTION, __VA_ARGS__, ""))
#define UA_LOG_FATAL_CONNECTION(LOGGER, CONNECTION, ...)                \
    UA_MACRO_EXPAND(UA_LOG_CONNECTION_INTERNAL(LOGGER, FATAL, CONNECTION, __VA_ARGS__, ""))
 
 
/**              DataSetWriter                 */
 
 
typedef struct UA_DataSetWriterSample {
    UA_Boolean valueChanged;
    UA_DataValue value;
} UA_DataSetWriterSample;
 
typedef struct UA_DataSetWriter {
    UA_PubSubComponentEnumType componentType;
    UA_DataSetWriterConfig config;
    LIST_ENTRY(UA_DataSetWriter) listEntry;
    UA_NodeId identifier;
    UA_NodeId linkedWriterGroup;
    UA_NodeId connectedDataSet;
    UA_ConfigurationVersionDataType connectedDataSetVersion;
    UA_PubSubState state;
 
    /* Deltaframes */
    UA_UInt16 deltaFrameCounter; /* count of sent deltaFrames */
    size_t lastSamplesCount;
    UA_DataSetWriterSample *lastSamples;
 
    UA_UInt16 actualDataSetMessageSequenceCount;
    UA_Boolean configurationFrozen;
} UA_DataSetWriter;
 
UA_StatusCode
UA_DataSetWriterConfig_copy(const UA_DataSetWriterConfig *src,
                            UA_DataSetWriterConfig *dst);
 
UA_DataSetWriter *
UA_DataSetWriter_findDSWbyId(UA_Server *server, UA_NodeId identifier);
 
UA_StatusCode
UA_DataSetWriter_setPubSubState(UA_Server *server,
                                UA_DataSetWriter *dataSetWriter,
                                UA_PubSubState state,
                                UA_StatusCode cause);
 
UA_StatusCode
UA_DataSetWriter_generateDataSetMessage(UA_Server *server,
                                        UA_DataSetMessage *dataSetMessage,
                                        UA_DataSetWriter *dataSetWriter);
 
UA_StatusCode
UA_DataSetWriter_prepareDataSet(UA_Server *server, UA_DataSetWriter *dsw,
                                UA_DataSetMessage *dsm);
 
void
UA_DataSetWriter_freezeConfiguration(UA_Server *server, UA_DataSetWriter *dsw);
 
void
UA_DataSetWriter_unfreezeConfiguration(UA_Server *server, UA_DataSetWriter *dsw);
 
UA_StatusCode
UA_DataSetWriter_create(UA_Server *server,
                        const UA_NodeId writerGroup, const UA_NodeId dataSet,
                        const UA_DataSetWriterConfig *dataSetWriterConfig,
                        UA_NodeId *writerIdentifier);
 
 
UA_StatusCode
UA_DataSetWriter_remove(UA_Server *server, UA_DataSetWriter *dataSetWriter);
 
#define UA_LOG_WRITER_INTERNAL(LOGGER, LEVEL, WRITER, MSG, ...)         \
    if(UA_LOGLEVEL <= UA_LOGLEVEL_##LEVEL) {                            \
        UA_String idStr = UA_STRING_NULL;                               \
        UA_String groupIdStr = UA_STRING_NULL;                          \
        if(WRITER) {                                                    \
            UA_NodeId_print(&(WRITER)->identifier, &idStr);             \
            UA_NodeId_print(&(WRITER)->linkedWriterGroup, &groupIdStr); \
        }                                                               \
        UA_LOG_##LEVEL(LOGGER, UA_LOGCATEGORY_PUBSUB,                   \
                       "WriterGroup %.*s\t| Writer %.*s\t| " MSG "%.0s", \
                       (int)groupIdStr.length, (char*)groupIdStr.data,  \
                       (int)idStr.length, (char*)idStr.data,            \
                       __VA_ARGS__);                                    \
        UA_String_clear(&idStr);                                        \
        UA_String_clear(&groupIdStr);                                   \
    }
 
#define UA_LOG_TRACE_WRITER(LOGGER, WRITER, ...)                        \
    UA_MACRO_EXPAND(UA_LOG_WRITER_INTERNAL(LOGGER, TRACE, WRITER, __VA_ARGS__, ""))
#define UA_LOG_DEBUG_WRITER(LOGGER, WRITER, ...)                        \
    UA_MACRO_EXPAND(UA_LOG_WRITER_INTERNAL(LOGGER, DEBUG, WRITER, __VA_ARGS__, ""))
#define UA_LOG_INFO_WRITER(LOGGER, WRITER, ...)                         \
    UA_MACRO_EXPAND(UA_LOG_WRITER_INTERNAL(LOGGER, INFO, WRITER, __VA_ARGS__, ""))
#define UA_LOG_WARNING_WRITER(LOGGER, WRITER, ...)                      \
    UA_MACRO_EXPAND(UA_LOG_WRITER_INTERNAL(LOGGER, WARNING, WRITER, __VA_ARGS__, ""))
#define UA_LOG_ERROR_WRITER(LOGGER, WRITER, ...)                        \
    UA_MACRO_EXPAND(UA_LOG_WRITER_INTERNAL(LOGGER, ERROR, WRITER, __VA_ARGS__, ""))
#define UA_LOG_FATAL_WRITER(LOGGER, WRITER, ...)                        \
    UA_MACRO_EXPAND(UA_LOG_WRITER_INTERNAL(LOGGER, FATAL, WRITER, __VA_ARGS__, ""))
 
 
/**               WriterGroup                  */
 
 
struct UA_WriterGroup {
    UA_PubSubComponentEnumType componentType;
    UA_WriterGroupConfig config;
    LIST_ENTRY(UA_WriterGroup) listEntry;
    UA_NodeId identifier;
 
    LIST_HEAD(, UA_DataSetWriter) writers;
    UA_UInt32 writersCount;
 
    UA_UInt64 publishCallbackId; /* registered if != 0 */
    UA_PubSubState state;
    UA_NetworkMessageOffsetBuffer bufferedMessage;
    UA_UInt16 sequenceNumber; /* Increased after every succressuly sent message */
    UA_Boolean configurationFrozen;
    UA_DateTime lastPublishTimeStamp;
 
    /* The ConnectionManager pointer is stored in the Connection. The channels
     * are either stored here or in the Connection, but never both. */
    UA_PubSubConnection *linkedConnection;
    uintptr_t sendChannel;
    UA_Boolean deleteFlag;
 
#ifdef UA_ENABLE_PUBSUB_ENCRYPTION
    UA_UInt32 securityTokenId;
    UA_UInt32 nonceSequenceNumber; /* To be part of the MessageNonce */
    void *securityPolicyContext;
#ifdef UA_ENABLE_PUBSUB_SKS
    UA_PubSubKeyStorage *keyStorage; /* non-owning pointer to keyStorage*/
#endif
#endif
};
 
UA_StatusCode
UA_WriterGroup_create(UA_Server *server, const UA_NodeId connection,
                      const UA_WriterGroupConfig *writerGroupConfig,
                      UA_NodeId *writerGroupIdentifier);
 
UA_StatusCode
UA_WriterGroup_remove(UA_Server *server, UA_WriterGroup *wg);
 
void
UA_WriterGroup_disconnect(UA_WriterGroup *wg);
 
UA_StatusCode
UA_WriterGroup_connect(UA_Server *server, UA_WriterGroup *wg,
                       UA_Boolean validate);
 
UA_StatusCode
setWriterGroupEncryptionKeys(UA_Server *server, const UA_NodeId writerGroup,
                             UA_UInt32 securityTokenId,
                             const UA_ByteString signingKey,
                             const UA_ByteString encryptingKey,
                             const UA_ByteString keyNonce);
 
UA_StatusCode
UA_WriterGroupConfig_copy(const UA_WriterGroupConfig *src,
                          UA_WriterGroupConfig *dst);
 
UA_WriterGroup *
UA_WriterGroup_findWGbyId(UA_Server *server, UA_NodeId identifier);
 
UA_StatusCode
UA_WriterGroup_freezeConfiguration(UA_Server *server, UA_WriterGroup *wg);
 
UA_StatusCode
UA_WriterGroup_unfreezeConfiguration(UA_Server *server, UA_WriterGroup *wg);
 
UA_StatusCode
UA_WriterGroup_setPubSubState(UA_Server *server,
                              UA_WriterGroup *writerGroup,
                              UA_PubSubState state,
                              UA_StatusCode cause);
 
UA_StatusCode
UA_WriterGroup_addPublishCallback(UA_Server *server, UA_WriterGroup *writerGroup);
 
void
UA_WriterGroup_publishCallback(UA_Server *server,
                               UA_WriterGroup *writerGroup);
 
UA_StatusCode
UA_WriterGroup_updateConfig(UA_Server *server, UA_WriterGroup *wg,
                            const UA_WriterGroupConfig *config);
 
#define UA_LOG_WRITERGROUP_INTERNAL(LOGGER, LEVEL, WRITERGROUP, MSG, ...) \
    if(UA_LOGLEVEL <= UA_LOGLEVEL_##LEVEL) {                            \
        UA_String idStr = UA_STRING_NULL;                               \
        if(WRITERGROUP)                                                 \
            UA_NodeId_print(&(WRITERGROUP)->identifier, &idStr);        \
        UA_LOG_##LEVEL(LOGGER, UA_LOGCATEGORY_PUBSUB,                   \
                       "WriterGroup %.*s\t| " MSG "%.0s",               \
                       (int)idStr.length, (char*)idStr.data,            \
                       __VA_ARGS__);                                    \
        UA_String_clear(&idStr);                                        \
    }
 
#define UA_LOG_TRACE_WRITERGROUP(LOGGER, WRITERGROUP, ...)              \
    UA_MACRO_EXPAND(UA_LOG_WRITERGROUP_INTERNAL(LOGGER, TRACE, WRITERGROUP, __VA_ARGS__, ""))
#define UA_LOG_DEBUG_WRITERGROUP(LOGGER, WRITERGROUP, ...)              \
    UA_MACRO_EXPAND(UA_LOG_WRITERGROUP_INTERNAL(LOGGER, DEBUG, WRITERGROUP, __VA_ARGS__, ""))
#define UA_LOG_INFO_WRITERGROUP(LOGGER, WRITERGROUP, ...)               \
    UA_MACRO_EXPAND(UA_LOG_WRITERGROUP_INTERNAL(LOGGER, INFO, WRITERGROUP, __VA_ARGS__, ""))
#define UA_LOG_WARNING_WRITERGROUP(LOGGER, WRITERGROUP, ...)            \
    UA_MACRO_EXPAND(UA_LOG_WRITERGROUP_INTERNAL(LOGGER, WARNING, WRITERGROUP, __VA_ARGS__, ""))
#define UA_LOG_ERROR_WRITERGROUP(LOGGER, WRITERGROUP, ...)              \
    UA_MACRO_EXPAND(UA_LOG_WRITERGROUP_INTERNAL(LOGGER, ERROR, WRITERGROUP, __VA_ARGS__, ""))
#define UA_LOG_FATAL_WRITERGROUP(LOGGER, WRITERGROUP, ...)              \
    UA_MACRO_EXPAND(UA_LOG_WRITERGROUP_INTERNAL(LOGGER, FATAL, WRITERGROUP, __VA_ARGS__, ""))
 
 
/**               DataSetField                 */
 
 
typedef struct UA_DataSetField {
    UA_DataSetFieldConfig config;
    TAILQ_ENTRY(UA_DataSetField) listEntry;
    UA_NodeId identifier;
    UA_NodeId publishedDataSet;     /* parent pds */
    UA_FieldMetaData fieldMetaData; /* contains the dataSetFieldId */
    UA_UInt64 sampleCallbackId;
    UA_Boolean sampleCallbackIsRegistered;
    UA_Boolean configurationFrozen;
} UA_DataSetField;
 
UA_StatusCode
UA_DataSetFieldConfig_copy(const UA_DataSetFieldConfig *src,
                           UA_DataSetFieldConfig *dst);
 
UA_DataSetField *
UA_DataSetField_findDSFbyId(UA_Server *server, UA_NodeId identifier);
 
UA_DataSetFieldResult
UA_DataSetField_remove(UA_Server *server, UA_DataSetField *currentField);
 
UA_DataSetFieldResult
UA_DataSetField_create(UA_Server *server, const UA_NodeId publishedDataSet,
                       const UA_DataSetFieldConfig *fieldConfig,
                       UA_NodeId *fieldIdentifier);
 
void
UA_PubSubDataSetField_sampleValue(UA_Server *server, UA_DataSetField *field,
                                  UA_DataValue *value);
 
 
/**               DataSetReader                */
 
 
/** DataSetReader Type definition */
typedef struct UA_DataSetReader {
    UA_PubSubComponentEnumType componentType;
    UA_DataSetReaderConfig config;
    UA_NodeId identifier;
    UA_NodeId linkedReaderGroup;
    LIST_ENTRY(UA_DataSetReader) listEntry;
 
    UA_PubSubState state; /* non std */
    UA_Boolean configurationFrozen;
    UA_NetworkMessageOffsetBuffer bufferedMessage;
 
#ifdef UA_ENABLE_PUBSUB_MONITORING
    /* MessageReceiveTimeout handling */
    UA_ServerCallback msgRcvTimeoutTimerCallback;
    UA_UInt64 msgRcvTimeoutTimerId;
    UA_Boolean msgRcvTimeoutTimerRunning;
#endif
    UA_DateTime lastHeartbeatReceived;
} UA_DataSetReader;
 
/** Process Network Message using DataSetReader */
void
UA_DataSetReader_process(UA_Server *server,
                         UA_ReaderGroup *readerGroup,
                         UA_DataSetReader *dataSetReader,
                         UA_DataSetMessage *dataSetMsg);
 
UA_StatusCode
UA_DataSetReader_checkIdentifier(UA_Server *server, UA_NetworkMessage *msg,
                                 UA_DataSetReader *reader,
                                 UA_ReaderGroupConfig readerGroupConfig);
 
UA_StatusCode
UA_DataSetReader_create(UA_Server *server, UA_NodeId readerGroupIdentifier,
                        const UA_DataSetReaderConfig *dataSetReaderConfig,
                        UA_NodeId *readerIdentifier);
 
UA_StatusCode
UA_DataSetReader_remove(UA_Server *server, UA_DataSetReader *dsr);
 
/** Copy the configuration of Target Variables */
UA_StatusCode UA_TargetVariables_copy(const UA_TargetVariables *src,
                                      UA_TargetVariables *dst);
 
/** Clear the Target Variables configuration */
void UA_TargetVariables_clear(UA_TargetVariables *subscribedDataSetTarget);
 
/** Copy the configuration of Field Target Variables */
UA_StatusCode UA_FieldTargetVariable_copy(const UA_FieldTargetVariable *src,
                                          UA_FieldTargetVariable *dst);
 
UA_StatusCode
DataSetReader_createTargetVariables(UA_Server *server, UA_DataSetReader *dsr,
                                    size_t targetVariablesSize,
                                    const UA_FieldTargetVariable *targetVariables);
 
UA_StatusCode
UA_DataSetReader_setPubSubState(UA_Server *server,
                                UA_DataSetReader *dataSetReader,
                                UA_PubSubState state,
                                UA_StatusCode cause);
 
#define UA_LOG_READER_INTERNAL(LOGGER, LEVEL, READER, MSG, ...)         \
    if(UA_LOGLEVEL <= UA_LOGLEVEL_##LEVEL) {                            \
        UA_String idStr = UA_STRING_NULL;                               \
        UA_String groupIdStr = UA_STRING_NULL;                          \
        if(READER) {                                                    \
            UA_NodeId_print(&(READER)->identifier, &idStr);             \
            UA_NodeId_print(&(READER)->linkedReaderGroup, &groupIdStr); \
        }                                                               \
        UA_LOG_##LEVEL(LOGGER, UA_LOGCATEGORY_PUBSUB,                   \
                       "ReaderGroup %.*s\t| Reader %.*s\t| " MSG "%.0s", \
                       (int)groupIdStr.length, (char*)groupIdStr.data,  \
                       (int)idStr.length, (char*)idStr.data,            \
                       __VA_ARGS__);                                    \
        UA_String_clear(&idStr);                                        \
        UA_String_clear(&groupIdStr);                                   \
    }
 
#define UA_LOG_TRACE_READER(LOGGER, READER, ...)                        \
    UA_MACRO_EXPAND(UA_LOG_READER_INTERNAL(LOGGER, TRACE, READER, __VA_ARGS__, ""))
#define UA_LOG_DEBUG_READER(LOGGER, READER, ...)                        \
    UA_MACRO_EXPAND(UA_LOG_READER_INTERNAL(LOGGER, DEBUG, READER, __VA_ARGS__, ""))
#define UA_LOG_INFO_READER(LOGGER, READER, ...)                         \
    UA_MACRO_EXPAND(UA_LOG_READER_INTERNAL(LOGGER, INFO, READER, __VA_ARGS__, ""))
#define UA_LOG_WARNING_READER(LOGGER, READER, ...)                      \
    UA_MACRO_EXPAND(UA_LOG_READER_INTERNAL(LOGGER, WARNING, READER, __VA_ARGS__, ""))
#define UA_LOG_ERROR_READER(LOGGER, READER, ...)                        \
    UA_MACRO_EXPAND(UA_LOG_READER_INTERNAL(LOGGER, ERROR, READER, __VA_ARGS__, ""))
#define UA_LOG_FATAL_READER(LOGGER, READER, ...)                        \
    UA_MACRO_EXPAND(UA_LOG_READER_INTERNAL(LOGGER, FATAL, READER, __VA_ARGS__, ""))
 
 
/**                ReaderGroup                 */
 
 
struct UA_ReaderGroup {
    UA_PubSubComponentEnumType componentType;
    UA_ReaderGroupConfig config;
    UA_NodeId identifier;
    LIST_ENTRY(UA_ReaderGroup) listEntry;
 
    LIST_HEAD(, UA_DataSetReader) readers;
    UA_UInt32 readersCount;
 
    UA_PubSubState state;
    UA_Boolean configurationFrozen;
 
    /* The ConnectionManager pointer is stored in the Connection. The channels 
     * are either stored here or in the Connection, but never both. */
    UA_PubSubConnection *linkedConnection;
    uintptr_t recvChannels[UA_PUBSUB_MAXCHANNELS];
    size_t recvChannelsSize;
    UA_Boolean deleteFlag;
 
#ifdef UA_ENABLE_PUBSUB_ENCRYPTION
    UA_UInt32 securityTokenId;
    UA_UInt32 nonceSequenceNumber; /* To be part of the MessageNonce */
    void *securityPolicyContext;
#ifdef UA_ENABLE_PUBSUB_SKS
    UA_PubSubKeyStorage *keyStorage;
#endif
#endif
};
 
UA_StatusCode
UA_ReaderGroup_create(UA_Server *server, UA_NodeId connectionId,
                      const UA_ReaderGroupConfig *rgc,
                      UA_NodeId *readerGroupId);
 
UA_StatusCode
UA_ReaderGroup_remove(UA_Server *server, UA_ReaderGroup *rg);
 
UA_StatusCode
UA_ReaderGroup_connect(UA_Server *server, UA_ReaderGroup *rg, UA_Boolean validate);
 
void
UA_ReaderGroup_disconnect(UA_ReaderGroup *rg);
 
UA_StatusCode
setReaderGroupEncryptionKeys(UA_Server *server, const UA_NodeId readerGroup,
                             UA_UInt32 securityTokenId,
                             const UA_ByteString signingKey,
                             const UA_ByteString encryptingKey,
                             const UA_ByteString keyNonce);
 
UA_StatusCode
UA_ReaderGroupConfig_copy(const UA_ReaderGroupConfig *src,
                          UA_ReaderGroupConfig *dst);
 
/** 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);
 
UA_StatusCode
UA_ReaderGroup_freezeConfiguration(UA_Server *server, UA_ReaderGroup *rg);
 
UA_StatusCode
UA_ReaderGroup_unfreezeConfiguration(UA_Server *server, UA_ReaderGroup *rg);
 
UA_StatusCode
UA_ReaderGroup_setPubSubState(UA_Server *server,
                              UA_ReaderGroup *readerGroup,
                              UA_PubSubState state,
                              UA_StatusCode cause);
 
UA_Boolean
UA_ReaderGroup_decodeAndProcessRT(UA_Server *server, UA_ReaderGroup *readerGroup,
                                    UA_ByteString *buf);
 
UA_Boolean
UA_ReaderGroup_process(UA_Server *server, UA_ReaderGroup *readerGroup,
                       UA_NetworkMessage *nm);
 
#define UA_LOG_READERGROUP_INTERNAL(LOGGER, LEVEL, RG, MSG, ...)        \
    if(UA_LOGLEVEL <= UA_LOGLEVEL_##LEVEL) {                            \
        UA_String idStr = UA_STRING_NULL;                               \
        if(RG)                                                          \
            UA_NodeId_print(&(RG)->identifier, &idStr);                 \
        UA_LOG_##LEVEL(LOGGER, UA_LOGCATEGORY_PUBSUB,                   \
                       "ReaderGroup %.*s\t| " MSG "%.0s", (int)idStr.length, \
                       (char*)idStr.data, __VA_ARGS__);                 \
        UA_String_clear(&idStr);                                        \
    }
 
#define UA_LOG_TRACE_READERGROUP(LOGGER, READERGROUP, ...)              \
    UA_MACRO_EXPAND(UA_LOG_READERGROUP_INTERNAL(LOGGER, TRACE, READERGROUP, __VA_ARGS__, ""))
#define UA_LOG_DEBUG_READERGROUP(LOGGER, READERGROUP, ...)              \
    UA_MACRO_EXPAND(UA_LOG_READERGROUP_INTERNAL(LOGGER, DEBUG, READERGROUP, __VA_ARGS__, ""))
#define UA_LOG_INFO_READERGROUP(LOGGER, READERGROUP, ...)               \
    UA_MACRO_EXPAND(UA_LOG_READERGROUP_INTERNAL(LOGGER, INFO, READERGROUP, __VA_ARGS__, ""))
#define UA_LOG_WARNING_READERGROUP(LOGGER, READERGROUP, ...)            \
    UA_MACRO_EXPAND(UA_LOG_READERGROUP_INTERNAL(LOGGER, WARNING, READERGROUP, __VA_ARGS__, ""))
#define UA_LOG_ERROR_READERGROUP(LOGGER, READERGROUP, ...)              \
    UA_MACRO_EXPAND(UA_LOG_READERGROUP_INTERNAL(LOGGER, ERROR, READERGROUP, __VA_ARGS__, ""))
#define UA_LOG_FATAL_READERGROUP(LOGGER, READERGROUP, ...)              \
    UA_MACRO_EXPAND(UA_LOG_READERGROUP_INTERNAL(LOGGER, FATAL, READERGROUP, __VA_ARGS__, ""))
 
 
/**               Reading Message handling                */
 
 
#ifdef UA_ENABLE_PUBSUB_ENCRYPTION
UA_StatusCode
verifyAndDecrypt(const UA_Logger *logger, UA_ByteString *buffer,
                 const size_t *currentPosition, const UA_NetworkMessage *nm,
                 UA_Boolean doValidate, UA_Boolean doDecrypt,
                 void *channelContext, UA_PubSubSecurityPolicy *securityPolicy);
 
UA_StatusCode
verifyAndDecryptNetworkMessage(const UA_Logger *logger, UA_ByteString *buffer,
                               size_t *currentPosition, UA_NetworkMessage *nm,
                               UA_ReaderGroup *readerGroup);
#endif
 
/** Takes a value (and not a pointer) to the buffer. The original buffer is
   const. Internally we may adjust the length during decryption. */
UA_StatusCode
decodeNetworkMessage(UA_Server *server, UA_ByteString *buffer, size_t *pos,
                     UA_NetworkMessage *nm, UA_PubSubConnection *connection);
 
#ifdef UA_ENABLE_PUBSUB_SKS
 
/**                    SecurityGroup                      */
 
struct UA_SecurityGroup {
    UA_String securityGroupId;
    UA_SecurityGroupConfig config;
    UA_PubSubKeyStorage *keyStorage;
    UA_NodeId securityGroupNodeId;
    UA_UInt64 callbackId;
    UA_DateTime baseTime;
#ifdef UA_ENABLE_PUBSUB_INFORMATIONMODEL
    UA_NodeId securityGroupFolderId;
#endif
    TAILQ_ENTRY(UA_SecurityGroup) listEntry;
};
 
UA_StatusCode
UA_SecurityGroupConfig_copy(const UA_SecurityGroupConfig *src,
                            UA_SecurityGroupConfig *dst);
 
/** finds the SecurityGroup within the server by SecurityGroup Name/Id*/
UA_SecurityGroup *
UA_SecurityGroup_findSGbyName(UA_Server *server, UA_String securityGroupName);
 
/** finds the SecurityGroup within the server by NodeId*/
UA_SecurityGroup *
UA_SecurityGroup_findSGbyId(UA_Server *server, UA_NodeId identifier);
 
void
UA_SecurityGroup_delete(UA_SecurityGroup *securityGroup);
 
void
removeSecurityGroup(UA_Server *server, UA_SecurityGroup *securityGroup);
 
#endif /* UA_ENABLE_PUBSUB_SKS */
 
 
/** PubSub Manager */
 
 
typedef struct UA_TopicAssign {
    UA_ReaderGroup *rgIdentifier;
    UA_String topic;
    TAILQ_ENTRY(UA_TopicAssign) listEntry;
} UA_TopicAssign;
 
typedef enum {
    UA_WRITER_GROUP = 0,
    UA_DATA_SET_WRITER = 1,
} UA_ReserveIdType;
 
typedef struct UA_ReserveId {
    UA_UInt16 id;
    UA_ReserveIdType reserveIdType;
    UA_String transportProfileUri;
    UA_NodeId sessionId;
    ZIP_ENTRY(UA_ReserveId) treeEntry;
} UA_ReserveId;
 
typedef ZIP_HEAD(UA_ReserveIdTree, UA_ReserveId) UA_ReserveIdTree;
 
typedef struct UA_PubSubManager {
    UA_UInt64 defaultPublisherId;
    /* Connections and PublishedDataSets can exist alone (own lifecycle) -> top
     * level components */
    size_t connectionsSize;
    TAILQ_HEAD(, UA_PubSubConnection) connections;
 
    size_t publishedDataSetsSize;
    TAILQ_HEAD(, UA_PublishedDataSet) publishedDataSets;
 
    size_t subscribedDataSetsSize;
    TAILQ_HEAD(, UA_StandaloneSubscribedDataSet) subscribedDataSets;
 
    size_t topicAssignSize;
    TAILQ_HEAD(, UA_TopicAssign) topicAssign;
 
    size_t reserveIdsSize;
    UA_ReserveIdTree reserveIds;
 
#ifdef UA_ENABLE_PUBSUB_SKS
    LIST_HEAD(, UA_PubSubKeyStorage) pubSubKeyList;
 
    size_t securityGroupsSize;
    TAILQ_HEAD(, UA_SecurityGroup) securityGroups;
#endif
 
#ifndef UA_ENABLE_PUBSUB_INFORMATIONMODEL
    UA_UInt32 uniqueIdCount;
#endif
} UA_PubSubManager;
 
UA_StatusCode
UA_PubSubManager_addPubSubTopicAssign(UA_Server *server, UA_ReaderGroup *readerGroup,
                                      UA_String topic);
 
UA_StatusCode
UA_PubSubManager_reserveIds(UA_Server *server, UA_NodeId sessionId, UA_UInt16 numRegWriterGroupIds,
                            UA_UInt16 numRegDataSetWriterIds, UA_String transportProfileUri,
                            UA_UInt16 **writerGroupIds, UA_UInt16 **dataSetWriterIds);
 
void
UA_PubSubManager_freeIds(UA_Server *server);
 
void
UA_PubSubManager_init(UA_Server *server, UA_PubSubManager *pubSubManager);
 
void
UA_PubSubManager_shutdown(UA_Server *server, UA_PubSubManager *pubSubManager);
 
void
UA_PubSubManager_delete(UA_Server *server, UA_PubSubManager *pubSubManager);
 
#ifndef UA_ENABLE_PUBSUB_INFORMATIONMODEL
void
UA_PubSubManager_generateUniqueNodeId(UA_PubSubManager *psm, UA_NodeId *nodeId);
#endif
 
#ifdef UA_ENABLE_PUBSUB_FILE_CONFIG
/** 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. */
UA_StatusCode
UA_PubSubManager_loadPubSubConfigFromByteString(UA_Server *server,
                                                const UA_ByteString buffer);
 
/** Saves the current PubSub configuration of a server in a byteString. */
UA_StatusCode
UA_PubSubManager_getEncodedPubSubConfiguration(UA_Server *server,
                                               UA_ByteString *buffer);
#endif
 
UA_Guid
UA_PubSubManager_generateUniqueGuid(UA_Server *server);
 
UA_UInt32
UA_PubSubConfigurationVersionTimeDifference(void);
 
 
/**      PubSub component monitoring              */
 
 
#ifdef UA_ENABLE_PUBSUB_MONITORING
 
UA_StatusCode
UA_PubSubManager_setDefaultMonitoringCallbacks(UA_PubSubMonitoringInterface *monitoringInterface);
 
#endif /* UA_ENABLE_PUBSUB_MONITORING */
 
#endif /* UA_ENABLE_PUBSUB */
 
_UA_END_DECLS
 
#endif /* UA_PUBSUB_H_ */