github/open62541
7
0
Fork 0
mirror of https://github.com/open62541/open62541.git synced 2025-06-03 04:00:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve PubSub subscriber example (#1811)

Browse Source 
* PubSub: Fix memleaks in subscriber tutorial

* PubSub: Stack-allocate in the subscriber tutorial

* PubSub: In subscriber tutorial, evaluate receive error code; quit early

* PubSub: In subscriber example, use logging API instead of printf

* PubSub: Reduce the max indendation of the subscriber example; Add comments

* PubSub: Use short blocking wait times in subscriber example

* PubSub: Reduce the max line length in the subscriber example
This commit is contained in:
Julius Pfrommer 2018-05-28 20:11:01 +02:00 committed by GitHub
parent a9a2af2d37
commit b040314739
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 83 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
examples/pubsub/tutorial_pubsub_subscribe.c
View File

@ -10,7 +10,6 @@
* PubSub subscriber API.
*/
#include <signal.h>
#include <stdio.h>
#include "ua_pubsub_networkmessage.h"
#include "ua_log_stdout.h"
#include "ua_server.h"
@ -29,42 +28,68 @@ 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!");
UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
"Message buffer allocation failed!");
return;
}
connection->channel->receive(connection->channel, &buffer, NULL, 300000);
if (buffer.length > 0) {
UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND, "Message received:");
UA_NetworkMessage *actualNetworkMessage = (UA_NetworkMessage *) UA_calloc(1, sizeof(UA_NetworkMessage));
size_t currentPosition = 0;
if (!actualNetworkMessage) {
UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "Networkmessage allocation failed!");
UA_ByteString_deleteMembers(&buffer);
return;
}
UA_NetworkMessage_decodeBinary(&buffer, &currentPosition, actualNetworkMessage);
printf("Message length: %zu\n", buffer.length);
if (actualNetworkMessage->networkMessageType == UA_NETWORKMESSAGE_DATASET) {
if ((actualNetworkMessage->payloadHeaderEnabled && (actualNetworkMessage->payloadHeader.dataSetPayloadHeader.count >= 1)) ||
(!actualNetworkMessage->payloadHeaderEnabled)) {
if (actualNetworkMessage->payload.dataSetPayload.dataSetMessages[0].header.dataSetMessageType ==
UA_DATASETMESSAGE_DATAKEYFRAME) {
for (int i = 0; i < actualNetworkMessage->payload.dataSetPayload.dataSetMessages[0].data.keyFrameData.fieldCount; i++) {
const UA_DataType *currentType = actualNetworkMessage->payload.dataSetPayload.dataSetMessages[0].data.keyFrameData.dataSetFields[i].value.type;
if (currentType == &UA_TYPES[UA_TYPES_BYTE]) {
printf("Message content: [Byte] \n\tReceived data: %i\n",
*((UA_Byte *) actualNetworkMessage->payload.dataSetPayload.dataSetMessages[0].data.keyFrameData.dataSetFields[i].value.data));
} else if (currentType == &UA_TYPES[UA_TYPES_DATETIME]) {
UA_DateTimeStruct receivedTime = UA_DateTime_toStruct(
*((UA_DateTime *) actualNetworkMessage->payload.dataSetPayload.dataSetMessages[0].data.keyFrameData.dataSetFields[i].value.data));
printf("Message content: [DateTime] \n\tReceived date: %02i-%02i-%02i Received time: %02i:%02i:%02i\n", receivedTime.year, receivedTime.month,
receivedTime.day, receivedTime.hour, receivedTime.min, receivedTime.sec);
}
}
}
}
/* 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_deleteMembers(&buffer);
return;
}
/* Decode the message */
UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
"Message length: %zu", buffer.length);
UA_NetworkMessage networkMessage;
memset(&networkMessage, 0, sizeof(UA_NetworkMessage));
size_t currentPosition = 0;
UA_NetworkMessage_decodeBinary(&buffer, &currentPosition, &networkMessage);
UA_ByteString_deleteMembers(&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);
}
}
cleanup:
UA_NetworkMessage_deleteMembers(&networkMessage);
}
int main(void) {
@ -72,8 +97,10 @@ int main(void) {
signal(SIGTERM, stopHandler);
UA_ServerConfig *config = UA_ServerConfig_new_minimal(4801, NULL);
/* Details about the PubSubTransportLayer can be found inside the tutorial_pubsub_connection */
config->pubsubTransportLayers = (UA_PubSubTransportLayer *) UA_malloc(sizeof(UA_PubSubTransportLayer));
/* Details about the PubSubTransportLayer can be found inside the
* tutorial_pubsub_connection */
config->pubsubTransportLayers = (UA_PubSubTransportLayer *)
UA_malloc(sizeof(UA_PubSubTransportLayer));
if (!config->pubsubTransportLayers) {
UA_ServerConfig_delete(config);
return -1;
@ -85,26 +112,34 @@ int main(void) {
UA_PubSubConnectionConfig connectionConfig;
memset(&connectionConfig, 0, sizeof(connectionConfig));
connectionConfig.name = UA_STRING("UDP-UADP Connection 1");
connectionConfig.transportProfileUri = UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-udp-uadp");
connectionConfig.transportProfileUri =
UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-udp-uadp");
connectionConfig.enabled = UA_TRUE;
UA_NetworkAddressUrlDataType networkAddressUrl = {UA_STRING_NULL , UA_STRING("opc.udp://224.0.0.22:4840/")};
UA_Variant_setScalar(&connectionConfig.address, &networkAddressUrl, &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]);
UA_NetworkAddressUrlDataType networkAddressUrl =
{UA_STRING_NULL , UA_STRING("opc.udp://224.0.0.22:4840/")};
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!");
}
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) {
/* 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);
if (rv == UA_STATUSCODE_GOOD) {
UA_UInt64 subscriptionCallbackId;
UA_Server_addRepeatedCallback(server, (UA_ServerCallback)subscriptionPollingCallback, connection, 5, &subscriptionCallbackId);
UA_Server_addRepeatedCallback(server, (UA_ServerCallback)subscriptionPollingCallback,
connection, 100, &subscriptionCallbackId);
} else {
UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "register channel failed: 0x%x!", rv);
UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "register channel failed: %s!",
UA_StatusCode_name(rv));
}
}