Add pubsub message encoding

CMakeLists.txt

@@ -86,6 +86,9 @@ endif()
 option(UA_ENABLE_MULTITHREADING "Enable multithreading (experimental)" OFF)
 mark_as_advanced(UA_ENABLE_MULTITHREADING)
 
+option(UA_ENABLE_PUBSUB "Enable publish/subscribe (experimental)" OFF)
+mark_as_advanced(UA_ENABLE_PUBSUB)
+
 option(UA_ENABLE_STATUSCODE_DESCRIPTIONS "Enable conversion of StatusCode to human-readable error message" ON)
 mark_as_advanced(UA_ENABLE_STATUSCODE_DESCRIPTIONS)
 
@@ -439,14 +442,15 @@ set(default_plugin_sources ${PROJECT_SOURCE_DIR}/plugins/ua_network_tcp.c
 )
 
 if(UA_ENABLE_ENCRYPTION)
-    set(default_plugin_headers ${default_plugin_headers}
-            ${PROJECT_SOURCE_DIR}/plugins/ua_securitypolicy_basic128rsa15.h)
-    set(default_plugin_headers ${default_plugin_headers}
-            ${PROJECT_SOURCE_DIR}/plugins/ua_securitypolicy_basic256sha256.h)
-    set(default_plugin_sources ${default_plugin_sources}
-            ${PROJECT_SOURCE_DIR}/plugins/ua_securitypolicy_basic128rsa15.c)
-    set(default_plugin_sources ${default_plugin_sources}
-            ${PROJECT_SOURCE_DIR}/plugins/ua_securitypolicy_basic256sha256.c)
+    list(APPEND default_plugin_headers ${PROJECT_SOURCE_DIR}/plugins/ua_securitypolicy_basic128rsa15.h
+                                       ${PROJECT_SOURCE_DIR}/plugins/ua_securitypolicy_basic256sha256.h)
+    list(APPEND default_plugin_sources ${PROJECT_SOURCE_DIR}/plugins/ua_securitypolicy_basic128rsa15.c
+                                       ${PROJECT_SOURCE_DIR}/plugins/ua_securitypolicy_basic256sha256.c)
+endif()
+
+if(UA_ENABLE_PUBSUB)
+    list(APPEND internal_headers ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub_networkmessage.h)
+    list(APPEND lib_sources ${PROJECT_SOURCE_DIR}/src/pubsub/ua_pubsub_networkmessage.c)
 endif()
 
 if(UA_DEBUG_DUMP_PKGS)

src/pubsub/ua_pubsub_networkmessage.h

@@ -0,0 +1,225 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at http://mozilla.org/MPL/2.0/.
+ *
+ * Copyright (c) 2017 - 2018 Fraunhofer IOSB (Author: Tino Bischoff)
+ */
+
+#ifndef UA_NETWORKMESSAGE_H_
+#define UA_NETWORKMESSAGE_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "ua_types.h"
+#include "ua_types_generated.h"
+
+/* DataSet Payload Header */
+typedef struct {
+    UA_Byte count;
+    UA_UInt16* dataSetWriterIds;
+} UA_DataSetPayloadHeader;
+
+/* FieldEncoding Enum  */
+typedef enum {
+    UA_FIELDENCODING_VARIANT = 0, 
+    UA_FIELDENCODING_RAWDATA = 1,
+    UA_FIELDENCODING_DATAVALUE = 2
+} UA_FieldEncoding;
+
+/* DataSetMessage Type */
+typedef enum {
+    UA_DATASETMESSAGE_DATAKEYFRAME = 0,
+    UA_DATASETMESSAGE_DATADELTAFRAME = 1,
+    UA_DATASETMESSAGE_EVENT = 2, 
+    UA_DATASETMESSAGE_KEEPALIVE = 3
+} UA_DataSetMessageType;
+
+/* DataSetMessage Header */
+typedef struct {
+    UA_Boolean dataSetMessageValid;
+    UA_FieldEncoding fieldEncoding;
+    UA_Boolean dataSetMessageSequenceNrEnabled;
+    UA_Boolean timestampEnabled;
+    UA_Boolean statusEnabled;
+    UA_Boolean configVersionMajorVersionEnabled;
+    UA_Boolean configVersionMinorVersionEnabled;
+    UA_DataSetMessageType dataSetMessageType;
+    UA_Boolean picoSecondsIncluded;
+    UA_UInt16 dataSetMessageSequenceNr;
+    UA_UtcTime timestamp;
+    UA_UInt16 picoSeconds;
+    UA_UInt16 status;
+    UA_UInt32 configVersionMajorVersion;
+    UA_UInt32 configVersionMinorVersion;
+} UA_DataSetMessageHeader;
+
+UA_StatusCode
+UA_DataSetMessageHeader_encodeBinary(const UA_DataSetMessageHeader* src,
+                                     UA_Byte **bufPos, const UA_Byte *bufEnd);
+
+UA_StatusCode
+UA_DataSetMessageHeader_decodeBinary(const UA_ByteString *src, size_t *offset,
+                                     UA_DataSetMessageHeader* dst);
+
+size_t
+UA_DataSetMessageHeader_calcSizeBinary(const UA_DataSetMessageHeader* p);
+
+/**
+ * DataSetMessage
+ * ^^^^^^^^^^^^^^ */
+
+typedef struct {
+    UA_UInt16 fieldCount;
+    UA_DataValue* dataSetFields;
+} UA_DataSetMessage_DataKeyFrameData;
+
+typedef struct {
+    UA_UInt16 fieldIndex;
+    UA_DataValue fieldValue;
+} UA_DataSetMessage_DeltaFrameField;
+
+typedef struct {
+    UA_UInt16 fieldCount;
+    UA_DataSetMessage_DeltaFrameField* deltaFrameFields;
+} UA_DataSetMessage_DataDeltaFrameData;
+
+typedef struct {
+    UA_DataSetMessageHeader header;
+    union {
+        UA_DataSetMessage_DataKeyFrameData keyFrameData;
+        UA_DataSetMessage_DataDeltaFrameData deltaFrameData;
+    } data;
+} UA_DataSetMessage;
+
+UA_StatusCode
+UA_DataSetMessage_encodeBinary(const UA_DataSetMessage* src, UA_Byte **bufPos,
+                               const UA_Byte *bufEnd);
+
+UA_StatusCode
+UA_DataSetMessage_decodeBinary(const UA_ByteString *src, size_t *offset,
+                               UA_DataSetMessage* dst);
+
+size_t
+UA_DataSetMessage_calcSizeBinary(const UA_DataSetMessage* p);
+
+void UA_DataSetMessage_free(const UA_DataSetMessage* p);
+
+typedef struct {
+    UA_UInt16* sizes;
+    UA_DataSetMessage* dataSetMessages;
+} UA_DataSetPayload;
+
+typedef enum {
+    UA_PUBLISHERDATATYPE_BYTE = 0,
+    UA_PUBLISHERDATATYPE_UINT16 = 1,
+    UA_PUBLISHERDATATYPE_UINT32 = 2,
+    UA_PUBLISHERDATATYPE_UINT64 = 3,
+    UA_PUBLISHERDATATYPE_STRING = 4
+} UA_PublisherIdDatatype;
+
+typedef enum {
+    UA_NETWORKMESSAGE_DATASET = 0,
+    UA_NETWORKMESSAGE_DISCOVERY_REQUEST = 1,
+    UA_NETWORKMESSAGE_DISCOVERY_RESPONSE = 2
+} UA_NetworkMessageType;
+
+/**
+ * UA_NetworkMessageGroupHeader
+ * ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ */
+typedef struct {
+    UA_Boolean writerGroupIdEnabled;
+    UA_Boolean groupVersionEnabled;
+    UA_Boolean networkMessageNumberEnabled;
+    UA_Boolean sequenceNumberEnabled;
+    UA_UInt16 writerGroupId;
+    UA_UInt32 groupVersion; // spec: type "VersionTime"
+    UA_UInt16 networkMessageNumber;
+    UA_UInt16 sequenceNumber;
+} UA_NetworkMessageGroupHeader;
+
+/**
+ * UA_NetworkMessageSecurityHeader
+ * ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ */
+typedef struct {
+    UA_Boolean networkMessageSigned;
+    UA_Boolean networkMessageEncrypted;
+    UA_Boolean securityFooterEnabled;
+    UA_Boolean forceKeyReset;
+    UA_UInt32 securityTokenId;      // spec: IntegerId
+    UA_Byte nonceLength;
+    UA_ByteString messageNonce;
+    UA_UInt16 securityFooterSize;
+} UA_NetworkMessageSecurityHeader;
+
+/**
+ * UA_NetworkMessage
+ * ^^^^^^^^^^^^^^^^^ */
+typedef struct {
+    UA_Byte version;
+    UA_Boolean publisherIdEnabled;
+    UA_Boolean groupHeaderEnabled;
+    UA_Boolean payloadHeaderEnabled;
+    UA_PublisherIdDatatype publisherIdType;
+    UA_Boolean dataSetClassIdEnabled;
+    UA_Boolean securityEnabled;
+    UA_Boolean timestampEnabled;
+    UA_Boolean picosecondsEnabled;
+    UA_Boolean chunkMessage;
+    UA_Boolean promotedFieldsEnabled;
+    UA_NetworkMessageType networkMessageType;
+    union {
+        UA_Byte publisherIdByte;
+        UA_UInt16 publisherIdUInt16;
+        UA_UInt32 publisherIdUInt32;
+        UA_UInt64 publisherIdUInt64;
+        UA_Guid publisherIdGuid;
+        UA_String publisherIdString;
+    } publisherId;
+    UA_Guid dataSetClassId;
+
+    UA_NetworkMessageGroupHeader groupHeader;
+
+    union {
+        UA_DataSetPayloadHeader dataSetPayloadHeader;
+    } payloadHeader;
+    
+    UA_DateTime timestamp;
+    UA_UInt16 picoseconds;
+    UA_UInt16 promotedFieldsSize;
+    UA_Variant* promotedFields;	/* BaseDataType */
+    
+    UA_NetworkMessageSecurityHeader securityHeader;
+
+    union {
+        UA_DataSetPayload dataSetPayload;
+    } payload;
+    
+    UA_ByteString securityFooter;
+    UA_ByteString signature;
+} UA_NetworkMessage;
+
+UA_StatusCode
+UA_NetworkMessage_encodeBinary(const UA_NetworkMessage* src,
+                               UA_Byte **bufPos, const UA_Byte *bufEnd);
+
+UA_StatusCode
+UA_NetworkMessage_decodeBinary(const UA_ByteString *src, size_t *offset,
+                               UA_NetworkMessage* dst);
+
+size_t
+UA_NetworkMessage_calcSizeBinary(const UA_NetworkMessage* p);
+
+void
+UA_NetworkMessage_deleteMembers(UA_NetworkMessage* p);
+
+void
+UA_NetworkMessage_delete(UA_NetworkMessage* p);
+
+
+#ifdef __cplusplus
+} // extern "C"
+#endif
+
+#endif /* UA_NETWORKMESSAGE_H_ */

tests/CMakeLists.txt

@@ -11,6 +11,7 @@ include_directories(${PROJECT_SOURCE_DIR}/include)
 include_directories(${PROJECT_SOURCE_DIR}/deps)
 include_directories(${PROJECT_SOURCE_DIR}/src)
 include_directories(${PROJECT_SOURCE_DIR}/src/server)
+include_directories(${PROJECT_SOURCE_DIR}/src/pubsub)
 include_directories(${PROJECT_SOURCE_DIR}/plugins)
 include_directories(${PROJECT_BINARY_DIR}/src_generated)
 include_directories(${PROJECT_SOURCE_DIR}/tests/testing-plugins)
@@ -164,6 +165,12 @@ if(UA_ENABLE_DISCOVERY)
     add_test_valgrind(discovery ${TESTS_BINARY_DIR}/check_discovery)
 endif()
 
+if(UA_ENABLE_PUBSUB)
+    add_executable(check_pubsub_encoding pubsub/check_pubsub_encoding.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-testplugins>)
+    target_link_libraries(check_pubsub_encoding ${LIBS})
+    add_test_valgrind(pubsub_encoding ${TESTS_BINARY_DIR}/check_pubsub_encoding)
+endif()
+
 add_executable(check_server_readspeed server/check_server_readspeed.c $<TARGET_OBJECTS:open62541-object> $<TARGET_OBJECTS:open62541-testplugins>)
 target_link_libraries(check_server_readspeed ${LIBS})
 add_test_valgrind(server_readspeed ${TESTS_BINARY_DIR}/check_server_readspeed)