GithubStar/open62541
1
0
Fork 0
mirror of https://github.com/open62541/open62541.git synced 2026-10-27 03:27:40 +08:00
Code Issues Packages Projects Releases Wiki Activity
Files
testing_offset_issue
open62541/examples/pubsub/tutorial_pubsub_mqtt_subscribe.c
T
Julius Pfrommer 5bf6f0c3e7 refactor(pubsub): Remove the old direct-access API 
The public Offset Table API should be used for realtime operations on
the NetworkMessages instead.
2025-01-19 22:33:50 +01:00

366 lines
15 KiB
C
Raw Permalink Blame History

/* 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) 2022 Fraunhofer IOSB (Author: Noel Graf)
*/
#include <open62541/server.h>
#include <open62541/server_pubsub.h>
#include <open62541/plugin/log_stdout.h>
#include <open62541/plugin/securitypolicy_default.h>
#include <stdio.h>
#define CONNECTION_NAME "MQTT Subscriber Connection"
#define TRANSPORT_PROFILE_URI_UADP "http://opcfoundation.org/UA-Profile/Transport/pubsub-mqtt-uadp"
#define TRANSPORT_PROFILE_URI_JSON "http://opcfoundation.org/UA-Profile/Transport/pubsub-mqtt-json"
#define MQTT_CLIENT_ID "TESTCLIENTPUBSUBMQTTSUBSCRIBE"
#define CONNECTIONOPTION_NAME "mqttClientId"
#define SUBSCRIBER_TOPIC "customTopic"
#define SUBSCRIBER_METADATAQUEUENAME "MetaDataTopic"
#define SUBSCRIBER_METADATAUPDATETIME 0
#define BROKER_ADDRESS_URL "opc.mqtt://127.0.0.1:1883"
// Uncomment the following line to enable MQTT login for the example
// #define EXAMPLE_USE_MQTT_LOGIN
#ifdef EXAMPLE_USE_MQTT_LOGIN
#define LOGIN_OPTION_COUNT 2
#define USERNAME_OPTION_NAME "mqttUsername"
#define PASSWORD_OPTION_NAME "mqttPassword"
#define MQTT_USERNAME "open62541user"
#define MQTT_PASSWORD "open62541"
#endif
// Uncomment the following line to enable MQTT via TLS for the example
//#define EXAMPLE_USE_MQTT_TLS
#ifdef EXAMPLE_USE_MQTT_TLS
#define TLS_OPTION_COUNT 2
#define USE_TLS_OPTION_NAME "mqttUseTLS"
#define MQTT_CA_FILE_PATH_OPTION_NAME "mqttCaFilePath"
#define CA_FILE_PATH "/path/to/server.cert"
#endif
#if defined(UA_ENABLE_ENCRYPTION_MBEDTLS) && !defined(UA_ENABLE_JSON_ENCODING)
#define UA_AES128CTR_SIGNING_KEY_LENGTH 32
#define UA_AES128CTR_KEY_LENGTH 16
#define UA_AES128CTR_KEYNONCE_LENGTH 4
UA_Byte signingKey[UA_AES128CTR_SIGNING_KEY_LENGTH] = {0};
UA_Byte encryptingKey[UA_AES128CTR_KEY_LENGTH] = {0};
UA_Byte keyNonce[UA_AES128CTR_KEYNONCE_LENGTH] = {0};
#endif
static UA_Boolean useJson = false;
UA_NodeId connectionIdent;
UA_NodeId subscribedDataSetIdent;
UA_NodeId readerGroupIdent;
UA_DataSetReaderConfig readerConfig;
static void fillTestDataSetMetaData(UA_DataSetMetaDataType *pMetaData);
static void
addPubSubConnection(UA_Server *server, char *addressUrl) {
/* Details about the connection configuration and handling are located
* in the pubsub connection tutorial */
UA_PubSubConnectionConfig connectionConfig;
memset(&connectionConfig, 0, sizeof(connectionConfig));
connectionConfig.name = UA_STRING(CONNECTION_NAME);
if(useJson) {
connectionConfig.transportProfileUri = UA_STRING(TRANSPORT_PROFILE_URI_JSON);
} else {
connectionConfig.transportProfileUri = UA_STRING(TRANSPORT_PROFILE_URI_UADP);
}
/* configure address of the mqtt broker (local on default port) */
UA_NetworkAddressUrlDataType networkAddressUrl = {UA_STRING_NULL , UA_STRING(addressUrl)};
UA_Variant_setScalar(&connectionConfig.address, &networkAddressUrl,
&UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]);
/* Changed to static publisherId from random generation to identify
* the publisher on Subscriber side */
connectionConfig.publisherId.idType = UA_PUBLISHERIDTYPE_UINT16;
connectionConfig.publisherId.id.uint16 = 2234;
UA_KeyValuePair connectionOptions[1];
UA_String mqttClientId = UA_STRING(MQTT_CLIENT_ID);
connectionOptions[0].key = UA_QUALIFIEDNAME(0, CONNECTIONOPTION_NAME);
UA_Variant_setScalar(&connectionOptions[0].value, &mqttClientId, &UA_TYPES[UA_TYPES_STRING]);
connectionConfig.connectionProperties.map = connectionOptions;
connectionConfig.connectionProperties.mapSize = 1;
UA_Server_addPubSubConnection(server, &connectionConfig, &connectionIdent);
}
/**
* **ReaderGroup**
*
* ReaderGroup is used to group a list of DataSetReaders. All ReaderGroups are
* created within a PubSubConnection and automatically deleted if the connection
* is removed. All network message related filters are only available in the DataSetReader. */
static void
addReaderGroup(UA_Server *server) {
UA_ReaderGroupConfig readerGroupConfig;
memset (&readerGroupConfig, 0, sizeof(UA_ReaderGroupConfig));
readerGroupConfig.name = UA_STRING("ReaderGroup1");
if(useJson)
readerGroupConfig.encodingMimeType = UA_PUBSUB_ENCODING_JSON;
/* configure the mqtt publish topic */
UA_BrokerDataSetReaderTransportDataType brokerTransportSettings;
memset(&brokerTransportSettings, 0, sizeof(UA_BrokerDataSetReaderTransportDataType));
/* Assign the Topic at which MQTT publish should happen */
/*ToDo: Pass the topic as argument from the reader group */
brokerTransportSettings.queueName = UA_STRING(SUBSCRIBER_TOPIC);
brokerTransportSettings.resourceUri = UA_STRING_NULL;
brokerTransportSettings.authenticationProfileUri = UA_STRING_NULL;
/* Choose the QOS Level for MQTT */
brokerTransportSettings.requestedDeliveryGuarantee = UA_BROKERTRANSPORTQUALITYOFSERVICE_BESTEFFORT;
/* Encapsulate config in transportSettings */
UA_ExtensionObject transportSettings;
memset(&transportSettings, 0, sizeof(UA_ExtensionObject));
transportSettings.encoding = UA_EXTENSIONOBJECT_DECODED;
transportSettings.content.decoded.type = &UA_TYPES[UA_TYPES_BROKERDATASETREADERTRANSPORTDATATYPE];
transportSettings.content.decoded.data = &brokerTransportSettings;
readerGroupConfig.transportSettings = transportSettings;
#if defined(UA_ENABLE_ENCRYPTION_MBEDTLS) && !defined(UA_ENABLE_JSON_ENCODING)
/* Encryption settings */
UA_ServerConfig *config = UA_Server_getConfig(server);
readerGroupConfig.securityMode = UA_MESSAGESECURITYMODE_SIGNANDENCRYPT;
readerGroupConfig.securityPolicy = &config->pubSubConfig.securityPolicies[0];
#endif
UA_Server_addReaderGroup(server, connectionIdent, &readerGroupConfig,
&readerGroupIdent);
#if defined(UA_ENABLE_ENCRYPTION_MBEDTLS) && !defined(UA_ENABLE_JSON_ENCODING)
/* Add the encryption key informaton */
UA_ByteString sk = {UA_AES128CTR_SIGNING_KEY_LENGTH, signingKey};
UA_ByteString ek = {UA_AES128CTR_KEY_LENGTH, encryptingKey};
UA_ByteString kn = {UA_AES128CTR_KEYNONCE_LENGTH, keyNonce};
// TODO security token not necessary for readergroup (extracted from security-header)
UA_Server_setReaderGroupEncryptionKeys(server, readerGroupIdent, 1, sk, ek, kn);
#endif
}
/**
* **DataSetReader**
*
* DataSetReader can receive NetworkMessages with the DataSetMessage
* of interest sent by the Publisher. DataSetReader provides
* the configuration necessary to receive and process DataSetMessages
* on the Subscriber side. DataSetReader must be linked with a
* SubscribedDataSet and be contained within a ReaderGroup. */
static void
addDataSetReader(UA_Server *server) {
memset (&readerConfig, 0, sizeof(UA_DataSetReaderConfig));
readerConfig.name = UA_STRING("DataSet Reader 1");
/* Parameters to filter which DataSetMessage has to be processed
* by the DataSetReader */
/* The following parameters are used to show that the data published by
* tutorial_pubsub_mqtt_publish.c is being subscribed and is being updated in
* the information model */
UA_UInt16 publisherIdentifier = 2234;
readerConfig.publisherId.idType = UA_PUBLISHERIDTYPE_UINT16;
readerConfig.publisherId.id.uint16 = publisherIdentifier;
readerConfig.writerGroupId = 100;
readerConfig.dataSetWriterId = 62541;
readerConfig.messageReceiveTimeout = 10;
/* Setting up Meta data configuration in DataSetReader */
fillTestDataSetMetaData(&readerConfig.dataSetMetaData);
UA_Server_addDataSetReader(server, readerGroupIdent, &readerConfig,
&subscribedDataSetIdent);
}
/**
* **SubscribedDataSet**
*
* Set SubscribedDataSet type to TargetVariables data type.
* Add subscribedvariables to the DataSetReader */
static void
addSubscribedVariables (UA_Server *server, UA_NodeId dataSetReaderId) {
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_NS0ID(OBJECTSFOLDER),
UA_NS0ID(ORGANIZES), folderBrowseName,
UA_NS0ID(BASEOBJECTTYPE), oAttr, NULL, &folderId);
/**
* **TargetVariables**
*
* The SubscribedDataSet option TargetVariables defines a list of Variable mappings between
* received DataSet fields and target Variables in the Subscriber AddressSpace.
* The values subscribed from the Publisher are updated in the value field of these variables */
/* Create the TargetVariables with respect to DataSetMetaData fields */
UA_FieldTargetDataType *targetVars = (UA_FieldTargetDataType *)
UA_calloc(readerConfig.dataSetMetaData.fieldsSize, sizeof(UA_FieldTargetDataType));
for(size_t i = 0; i < readerConfig.dataSetMetaData.fieldsSize; i++) {
/* Variable to subscribe data */
UA_VariableAttributes vAttr = UA_VariableAttributes_default;
UA_LocalizedText_copy(&readerConfig.dataSetMetaData.fields[i].description,
&vAttr.description);
vAttr.displayName.locale = UA_STRING("en-US");
vAttr.displayName.text = readerConfig.dataSetMetaData.fields[i].name;
vAttr.dataType = readerConfig.dataSetMetaData.fields[i].dataType;
UA_NodeId newNode;
UA_Server_addVariableNode(server, UA_NODEID_NUMERIC(1, (UA_UInt32)i + 50000),
folderId, UA_NS0ID(HASCOMPONENT),
UA_QUALIFIEDNAME(1, (char *)readerConfig.dataSetMetaData.fields[i].name.data),
UA_NS0ID(BASEDATAVARIABLETYPE),
vAttr, NULL, &newNode);
targetVars[i].attributeId = UA_ATTRIBUTEID_VALUE;
targetVars[i].targetNodeId = newNode;
}
UA_Server_DataSetReader_createTargetVariables(server, dataSetReaderId,
readerConfig.dataSetMetaData.fieldsSize,
targetVars);
UA_free(targetVars);
UA_free(readerConfig.dataSetMetaData.fields);
}
/**
* **DataSetMetaData**
*
* The DataSetMetaData describes the content of a DataSet. It provides the information necessary to decode
* DataSetMessages on the Subscriber side. DataSetMessages received from the Publisher are decoded into
* DataSet and each field is updated in the Subscriber based on datatype match of TargetVariable fields of Subscriber
* and PublishedDataSetFields of Publisher */
static void
fillTestDataSetMetaData(UA_DataSetMetaDataType *pMetaData) {
UA_DataSetMetaDataType_init (pMetaData);
pMetaData->name = UA_STRING ("DataSet 1");
/* 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;
pMetaData->fields[0].name = UA_STRING ("DateTime");
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;
pMetaData->fields[1].name = UA_STRING ("Int32");
pMetaData->fields[1].valueRank = -1; /* scalar */
/* Int64 DataType */
UA_FieldMetaData_init (&pMetaData->fields[2]);
UA_NodeId_copy(&UA_TYPES[UA_TYPES_INT64].typeId,
&pMetaData->fields[2].dataType);
pMetaData->fields[2].builtInType = UA_NS0ID_INT64;
pMetaData->fields[2].name = UA_STRING ("Int64");
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;
pMetaData->fields[3].name = UA_STRING ("BoolToggle");
pMetaData->fields[3].valueRank = -1; /* scalar */
}
static void usage(void) {
printf("Usage: tutorial_pubsub_mqtt_subscribe [--url <opc.mqtt://hostname:port>] "
"[--json]\n"
" Defaults are:\n"
" - Url: opc.mqtt://127.0.0.1:1883\n"
" - JSON: Off\n");
}
/**
* Followed by the main server code, making use of the above definitions */
int main(int argc, char **argv) {
char *addressUrl = BROKER_ADDRESS_URL;
/* Parse arguments */
for(int argpos = 1; argpos < argc; argpos++) {
if(strcmp(argv[argpos], "--help") == 0) {
usage();
return 0;
}
if(strcmp(argv[argpos], "--json") == 0) {
#ifdef UA_ENABLE_JSON_ENCODING
useJson = true;
#else
printf("Json encoding not enabled (UA_ENABLE_JSON_ENCODING)\n");
useJson = false;
#endif
continue;
}
if(strcmp(argv[argpos], "--url") == 0) {
if(argpos + 1 == argc) {
usage();
return -1;
}
argpos++;
addressUrl = argv[argpos];
continue;
}
}
/* Return value initialized to Status Good */
UA_Server *server = UA_Server_new();
#if defined(UA_ENABLE_ENCRYPTION_MBEDTLS)
/* Instantiate the PubSub SecurityPolicy */
UA_ServerConfig *config = UA_Server_getConfig(server);
config->pubSubConfig.securityPolicies = (UA_PubSubSecurityPolicy*)
UA_malloc(sizeof(UA_PubSubSecurityPolicy));
config->pubSubConfig.securityPoliciesSize = 1;
UA_PubSubSecurityPolicy_Aes128Ctr(config->pubSubConfig.securityPolicies,
config->logging);
#endif
addPubSubConnection(server, addressUrl);
addReaderGroup(server);
addDataSetReader(server);
addSubscribedVariables(server, subscribedDataSetIdent);
UA_Server_enableAllPubSubComponents(server);
UA_Server_runUntilInterrupt(server);
UA_Server_delete(server);
return 0;
}
Reference in New Issue View Git Blame Copy Permalink