/* This work is licensed under a Creative Commons CCZero 1.0 Universal License. * See http://creativecommons.org/publicdomain/zero/1.0/ for more information. */ /** * .. _pubsub-tutorial: * * Working with Publish/Subscribe * ------------------------------ * * Work in progress: This Tutorial will be continuously extended during the next * PubSub batches. More details about the PubSub extension and corresponding * open62541 API are located here: :ref:`pubsub`. * * Publishing Fields * ^^^^^^^^^^^^^^^^^ * The PubSub MQTT publish example demonstrates the simplest way to publish * information from the information model over MQTT using the UADP (or * JSON) encoding. To receive information the subscribe functionality of MQTT is * used. A periodical call to yield is necessary to update the mqtt stack. * * **Connection handling** * PubSubConnections can be created and deleted on runtime. More details about * the system preconfiguration and connection can be found in * ``tutorial_pubsub_connection.c``. * * MQTT via TLS * ^^^^^^^^^^^^ * Defining EXAMPLE_USE_MQTT_TLS enables TLS for the MQTT connection. This is * experimental and currently only available with OpenSSL. * * https://test.mosquitto.org/ offers a public MQTT broker with TLS support. * The access via port 8883 can be used with this example. Download the file * ``mosquitto.org.crt`` and point the CA_FILE_PATH define to its location. * * If the server requires a client certificate, the two options ``mqttClientCertPath`` * and ``mqttClientKeyPath`` must be added in ``addPubSubConnection()``. * * To use a directory containing hashed certificates generated by ``c_rehash``, * use the ``mqttCaPath`` option instead of ``mqttCaFilePath``. If none of these two * options is provided, the system's default certificate location will be used. */ #include #include #include #include #define CONNECTION_NAME "MQTT Publisher 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 "TESTCLIENTPUBSUBMQTT" #define CONNECTIONOPTION_NAME "mqttClientId" #define PUBLISHER_TOPIC "customTopic" #define PUBLISHER_METADATAQUEUENAME "MetaDataTopic" #define PUBLISHER_METADATAUPDATETIME 0 #define BROKER_ADDRESS_URL "opc.mqtt://127.0.0.1:1883" #define PUBLISH_INTERVAL 500 // 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_PUBSUB_ENCRYPTION) && !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 #ifdef UA_ENABLE_JSON_ENCODING static UA_Boolean useJson = true; #else static UA_Boolean useJson = false; #endif static UA_NodeId connectionIdent; static UA_NodeId publishedDataSetIdent; static UA_NodeId writerGroupIdent; 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); } connectionConfig.enabled = true; /* 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.publisherIdType = UA_PUBLISHERIDTYPE_UINT16; connectionConfig.publisherId.uint16 = 2234; /* configure options, set mqtt client id */ /* #ifdef EXAMPLE_USE_MQTT_LOGIN */ /* + LOGIN_OPTION_COUNT */ /* #endif */ /* #ifdef EXAMPLE_USE_MQTT_TLS */ /* + TLS_OPTION_COUNT */ /* #endif */ 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]); /* #ifdef EXAMPLE_USE_MQTT_LOGIN */ /* connectionOptions[connectionOptionIndex].key = UA_QUALIFIEDNAME(0, USERNAME_OPTION_NAME); */ /* UA_String mqttUsername = UA_STRING(MQTT_USERNAME); */ /* UA_Variant_setScalar(&connectionOptions[connectionOptionIndex++].value, &mqttUsername, &UA_TYPES[UA_TYPES_STRING]); */ /* connectionOptions[connectionOptionIndex].key = UA_QUALIFIEDNAME(0, PASSWORD_OPTION_NAME); */ /* UA_String mqttPassword = UA_STRING(MQTT_PASSWORD); */ /* UA_Variant_setScalar(&connectionOptions[connectionOptionIndex++].value, &mqttPassword, &UA_TYPES[UA_TYPES_STRING]); */ /* #endif */ /* #ifdef EXAMPLE_USE_MQTT_TLS */ /* connectionOptions[connectionOptionIndex].key = UA_QUALIFIEDNAME(0, USE_TLS_OPTION_NAME); */ /* UA_Boolean mqttUseTLS = true; */ /* UA_Variant_setScalar(&connectionOptions[connectionOptionIndex++].value, &mqttUseTLS, &UA_TYPES[UA_TYPES_BOOLEAN]); */ /* connectionOptions[connectionOptionIndex].key = UA_QUALIFIEDNAME(0, MQTT_CA_FILE_PATH_OPTION_NAME); */ /* UA_String mqttCaFile = UA_STRING(CA_FILE_PATH); */ /* UA_Variant_setScalar(&connectionOptions[connectionOptionIndex++].value, &mqttCaFile, &UA_TYPES[UA_TYPES_STRING]); */ /* #endif */ connectionConfig.connectionProperties.map = connectionOptions; connectionConfig.connectionProperties.mapSize = 1; UA_Server_addPubSubConnection(server, &connectionConfig, &connectionIdent); } /** * **PublishedDataSet handling** * The PublishedDataSet (PDS) and PubSubConnection are the toplevel entities and * can exist alone. The PDS contains the collection of the published fields. All * other PubSub elements are directly or indirectly linked with the PDS or * connection. */ static void addPublishedDataSet(UA_Server *server) { /* The PublishedDataSetConfig contains all necessary public * information for the creation of a new PublishedDataSet */ UA_PublishedDataSetConfig publishedDataSetConfig; memset(&publishedDataSetConfig, 0, sizeof(UA_PublishedDataSetConfig)); publishedDataSetConfig.publishedDataSetType = UA_PUBSUB_DATASET_PUBLISHEDITEMS; publishedDataSetConfig.name = UA_STRING("Demo PDS"); /* Create new PublishedDataSet based on the PublishedDataSetConfig. */ UA_Server_addPublishedDataSet(server, &publishedDataSetConfig, &publishedDataSetIdent); } /** * **DataSetField handling** * The DataSetField (DSF) is part of the PDS and describes exactly one published field. */ static void addDataSetField(UA_Server *server) { /* Add a field to the previous created PublishedDataSet */ UA_DataSetFieldConfig dataSetFieldConfig; memset(&dataSetFieldConfig, 0, sizeof(UA_DataSetFieldConfig)); dataSetFieldConfig.dataSetFieldType = UA_PUBSUB_DATASETFIELD_VARIABLE; dataSetFieldConfig.field.variable.fieldNameAlias = UA_STRING("Server localtime"); dataSetFieldConfig.field.variable.promotedField = UA_FALSE; dataSetFieldConfig.field.variable.publishParameters.publishedVariable = UA_NODEID_NUMERIC(0, UA_NS0ID_SERVER_SERVERSTATUS_CURRENTTIME); dataSetFieldConfig.field.variable.publishParameters.attributeId = UA_ATTRIBUTEID_VALUE; UA_Server_addDataSetField(server, publishedDataSetIdent, &dataSetFieldConfig, NULL); } /** * **WriterGroup handling** * The WriterGroup (WG) is part of the connection and contains the primary configuration * parameters for the message creation. */ static UA_StatusCode addWriterGroup(UA_Server *server, char *topic, int interval) { UA_StatusCode retval = UA_STATUSCODE_GOOD; /* Now we create a new WriterGroupConfig and add the group to the existing PubSubConnection. */ UA_WriterGroupConfig writerGroupConfig; memset(&writerGroupConfig, 0, sizeof(UA_WriterGroupConfig)); writerGroupConfig.name = UA_STRING("Demo WriterGroup"); writerGroupConfig.publishingInterval = interval; writerGroupConfig.enabled = UA_FALSE; writerGroupConfig.writerGroupId = 100; UA_UadpWriterGroupMessageDataType *writerGroupMessage; /* decide whether to use JSON or UADP encoding*/ #ifdef UA_ENABLE_JSON_ENCODING UA_JsonWriterGroupMessageDataType *Json_writerGroupMessage; if(useJson) { writerGroupConfig.encodingMimeType = UA_PUBSUB_ENCODING_JSON; writerGroupConfig.messageSettings.encoding = UA_EXTENSIONOBJECT_DECODED; writerGroupConfig.messageSettings.content.decoded.type = &UA_TYPES[UA_TYPES_JSONWRITERGROUPMESSAGEDATATYPE]; /* The configuration flags for the messages are encapsulated inside the * message- and transport settings extension objects. These extension * objects are defined by the standard. e.g. * UadpWriterGroupMessageDataType */ Json_writerGroupMessage = UA_JsonWriterGroupMessageDataType_new(); /* Change message settings of writerGroup to send PublisherId, * DataSetMessageHeader, SingleDataSetMessage and DataSetClassId in PayloadHeader * of NetworkMessage */ Json_writerGroupMessage->networkMessageContentMask = (UA_JsonNetworkMessageContentMask)(UA_JSONNETWORKMESSAGECONTENTMASK_NETWORKMESSAGEHEADER | (UA_JsonNetworkMessageContentMask)UA_JSONNETWORKMESSAGECONTENTMASK_DATASETMESSAGEHEADER | (UA_JsonNetworkMessageContentMask)UA_JSONNETWORKMESSAGECONTENTMASK_SINGLEDATASETMESSAGE | (UA_JsonNetworkMessageContentMask)UA_JSONNETWORKMESSAGECONTENTMASK_PUBLISHERID | (UA_JsonNetworkMessageContentMask)UA_JSONNETWORKMESSAGECONTENTMASK_DATASETCLASSID); writerGroupConfig.messageSettings.content.decoded.data = Json_writerGroupMessage; } else #endif { writerGroupConfig.encodingMimeType = UA_PUBSUB_ENCODING_UADP; writerGroupConfig.messageSettings.encoding = UA_EXTENSIONOBJECT_DECODED; writerGroupConfig.messageSettings.content.decoded.type = &UA_TYPES[UA_TYPES_UADPWRITERGROUPMESSAGEDATATYPE]; /* The configuration flags for the messages are encapsulated inside the * message- and transport settings extension objects. These extension * objects are defined by the standard. e.g. * UadpWriterGroupMessageDataType */ writerGroupMessage = UA_UadpWriterGroupMessageDataType_new(); /* Change message settings of writerGroup to send PublisherId, * WriterGroupId in GroupHeader and DataSetWriterId in PayloadHeader * of NetworkMessage */ writerGroupMessage->networkMessageContentMask = (UA_UadpNetworkMessageContentMask)(UA_UADPNETWORKMESSAGECONTENTMASK_PUBLISHERID | (UA_UadpNetworkMessageContentMask)UA_UADPNETWORKMESSAGECONTENTMASK_GROUPHEADER | (UA_UadpNetworkMessageContentMask)UA_UADPNETWORKMESSAGECONTENTMASK_WRITERGROUPID | (UA_UadpNetworkMessageContentMask)UA_UADPNETWORKMESSAGECONTENTMASK_PAYLOADHEADER); writerGroupConfig.messageSettings.content.decoded.data = writerGroupMessage; } #if defined(UA_ENABLE_PUBSUB_ENCRYPTION) && !defined(UA_ENABLE_JSON_ENCODING) /* Encryption settings */ UA_ServerConfig *config = UA_Server_getConfig(server); writerGroupConfig.securityMode = UA_MESSAGESECURITYMODE_SIGNANDENCRYPT; writerGroupConfig.securityPolicy = &config->pubSubConfig.securityPolicies[0]; #endif /* configure the mqtt publish topic */ UA_BrokerWriterGroupTransportDataType brokerTransportSettings; memset(&brokerTransportSettings, 0, sizeof(UA_BrokerWriterGroupTransportDataType)); /* Assign the Topic at which MQTT publish should happen */ /*ToDo: Pass the topic as argument from the writer group */ brokerTransportSettings.queueName = UA_STRING(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_BROKERWRITERGROUPTRANSPORTDATATYPE]; transportSettings.content.decoded.data = &brokerTransportSettings; writerGroupConfig.transportSettings = transportSettings; retval = UA_Server_addWriterGroup(server, connectionIdent, &writerGroupConfig, &writerGroupIdent); if (retval == UA_STATUSCODE_GOOD) UA_Server_setWriterGroupOperational(server, writerGroupIdent); #ifdef UA_ENABLE_JSON_ENCODING if (useJson) { UA_JsonWriterGroupMessageDataType_delete(Json_writerGroupMessage); } #endif if (!useJson && writerGroupMessage) { UA_UadpWriterGroupMessageDataType_delete(writerGroupMessage); } #if defined(UA_ENABLE_PUBSUB_ENCRYPTION) && !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}; retval = UA_Server_setWriterGroupEncryptionKeys(server, writerGroupIdent, 1, sk, ek, kn); if (retval!= UA_STATUSCODE_GOOD) { UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,"SV_PubSub.c : addWriterGroup : UA_Server_setWriterGroupEncryptionKeys : " "failure %s", UA_StatusCode_name(retval)); } #endif return retval; } /** * **DataSetWriter handling** * A DataSetWriter (DSW) is the glue between the WG and the PDS. The DSW is * linked to exactly one PDS and contains additional information for the * message generation. */ static void addDataSetWriter(UA_Server *server, char *topic) { /* We need now a DataSetWriter within the WriterGroup. This means we must * create a new DataSetWriterConfig and add call the addWriterGroup function. */ UA_NodeId dataSetWriterIdent; UA_DataSetWriterConfig dataSetWriterConfig; memset(&dataSetWriterConfig, 0, sizeof(UA_DataSetWriterConfig)); dataSetWriterConfig.name = UA_STRING("Demo DataSetWriter"); dataSetWriterConfig.dataSetWriterId = 62541; dataSetWriterConfig.keyFrameCount = 10; #ifdef UA_ENABLE_JSON_ENCODING UA_JsonDataSetWriterMessageDataType jsonDswMd; UA_ExtensionObject messageSettings; if(useJson) { /* JSON config for the dataSetWriter */ jsonDswMd.dataSetMessageContentMask = (UA_JsonDataSetMessageContentMask) (UA_JSONDATASETMESSAGECONTENTMASK_DATASETWRITERID | UA_JSONDATASETMESSAGECONTENTMASK_SEQUENCENUMBER | UA_JSONDATASETMESSAGECONTENTMASK_STATUS | UA_JSONDATASETMESSAGECONTENTMASK_METADATAVERSION | UA_JSONDATASETMESSAGECONTENTMASK_TIMESTAMP); messageSettings.encoding = UA_EXTENSIONOBJECT_DECODED; messageSettings.content.decoded.type = &UA_TYPES[UA_TYPES_JSONDATASETWRITERMESSAGEDATATYPE]; messageSettings.content.decoded.data = &jsonDswMd; dataSetWriterConfig.messageSettings = messageSettings; } #endif /*TODO: Modify MQTT send to add DataSetWriters broker transport settings */ /*TODO: Pass the topic as argument from the writer group */ /*TODO: Publish Metadata to metaDataQueueName */ /* configure the mqtt publish topic */ UA_BrokerDataSetWriterTransportDataType brokerTransportSettings; memset(&brokerTransportSettings, 0, sizeof(UA_BrokerDataSetWriterTransportDataType)); /* Assign the Topic at which MQTT publish should happen */ brokerTransportSettings.queueName = UA_STRING(topic); brokerTransportSettings.resourceUri = UA_STRING_NULL; brokerTransportSettings.authenticationProfileUri = UA_STRING_NULL; brokerTransportSettings.metaDataQueueName = UA_STRING(PUBLISHER_METADATAQUEUENAME); brokerTransportSettings.metaDataUpdateTime = PUBLISHER_METADATAUPDATETIME; /* 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_BROKERDATASETWRITERTRANSPORTDATATYPE]; transportSettings.content.decoded.data = &brokerTransportSettings; dataSetWriterConfig.transportSettings = transportSettings; UA_Server_addDataSetWriter(server, writerGroupIdent, publishedDataSetIdent, &dataSetWriterConfig, &dataSetWriterIdent); } /** * That's it! You're now publishing the selected fields. Open a packet * inspection tool of trust e.g. wireshark and take a look on the outgoing * packages. The following graphic figures out the packages created by this * tutorial. * * .. figure:: ua-wireshark-pubsub.png * :figwidth: 100 % * :alt: OPC UA PubSub communication in wireshark * * The open62541 subscriber API will be released later. If you want to process * the the datagrams, take a look on the ua_network_pubsub_networkmessage.c * which already contains the decoding code for UADP messages. * * It follows the main server code, making use of the above definitions. */ static void usage(void) { printf("Usage: tutorial_pubsub_mqtt_publish [--url ] " "[--topic ] " "[--freq " "[--json]\n" " Defaults are:\n" " - Url: opc.mqtt://127.0.0.1:1883\n" " - Topic: customTopic\n" " - Frequency: 500\n" " - JSON: Off\n"); } int main(int argc, char **argv) { /* TODO: Change to secure mqtt port:8883 */ char *addressUrl = BROKER_ADDRESS_URL; char *topic = PUBLISHER_TOPIC; int interval = PUBLISH_INTERVAL; /* Parse arguments */ for(int argpos = 1; argpos < argc; argpos++) { if(strcmp(argv[argpos], "--help") == 0) { usage(); return 0; } if(strcmp(argv[argpos], "--json") == 0) { useJson = true; continue; } if(strcmp(argv[argpos], "--url") == 0) { if(argpos + 1 == argc) { usage(); return -1; } argpos++; addressUrl = argv[argpos]; continue; } if(strcmp(argv[argpos], "--topic") == 0) { if(argpos + 1 == argc) { usage(); return -1; } argpos++; topic = argv[argpos]; continue; } if(strcmp(argv[argpos], "--freq") == 0) { if(argpos + 1 == argc) { usage(); return -1; } argpos++; if(sscanf(argv[argpos], "%d", &interval) != 1) { usage(); return -1; } if(interval <= 10) { UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "Publication interval too small"); return -1; } continue; } usage(); return -1; } UA_Server *server = UA_Server_new(); #if defined(UA_ENABLE_PUBSUB_ENCRYPTION) /* 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); addPublishedDataSet(server); addDataSetField(server); UA_StatusCode retval = addWriterGroup(server, topic, interval); if(UA_STATUSCODE_GOOD != retval) { UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "Error Name = %s", UA_StatusCode_name(retval)); UA_Server_delete(server); return EXIT_SUCCESS; } addDataSetWriter(server, topic); UA_Server_runUntilInterrupt(server); UA_Server_delete(server); return 0; }