pubsub_subscribe_standalone.c 5.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135
  1. /* This work is licensed under a Creative Commons CCZero 1.0 Universal License.
  2. * See http://creativecommons.org/publicdomain/zero/1.0/ for more information.
  3. */
  4. /**
  5. * IMPORTANT ANNOUNCEMENT
  6. * The PubSub subscriber API is currently not finished. This examples can be used to receive
  7. * and print the values, which are published by the tutorial_pubsub_publish example.
  8. * The following code uses internal API which will be later replaced by the higher-level
  9. * PubSub subscriber API.
  10. */
  11. #include <open62541/plugin/log_stdout.h>
  12. #include <open62541/plugin/pubsub.h>
  13. #include <open62541/plugin/pubsub_udp.h>
  14. #include <open62541/server.h>
  15. #include <open62541/server_config_default.h>
  16. #include "ua_pubsub_networkmessage.h"
  17. #include <signal.h>
  18. #ifdef UA_ENABLE_PUBSUB_ETH_UADP
  19. #include <open62541/plugin/pubsub_ethernet.h>
  20. #endif
  21. UA_Boolean running = true;
  22. static void stopHandler(int sign) {
  23. UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
  24. "received ctrl-c");
  25. running = false;
  26. }
  27. static UA_StatusCode
  28. subscriberListen(UA_PubSubChannel *psc) {
  29. UA_ByteString buffer;
  30. UA_StatusCode retval = UA_ByteString_allocBuffer(&buffer, 512);
  31. if(retval != UA_STATUSCODE_GOOD) {
  32. UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
  33. "Message buffer allocation failed!");
  34. return retval;
  35. }
  36. /* Receive the message. Blocks for 100ms */
  37. retval = psc->receive(psc, &buffer, NULL, 100);
  38. if(retval != UA_STATUSCODE_GOOD || buffer.length == 0) {
  39. /* Workaround!! Reset buffer length. Receive can set the length to zero.
  40. * Then the buffer is not deleted because no memory allocation is
  41. * assumed.
  42. * TODO: Return an error code in 'receive' instead of setting the buf
  43. * length to zero. */
  44. buffer.length = 512;
  45. UA_ByteString_clear(&buffer);
  46. return UA_STATUSCODE_GOOD;
  47. }
  48. /* Decode the message */
  49. UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
  50. "Message length: %lu", (unsigned long) buffer.length);
  51. UA_NetworkMessage networkMessage;
  52. memset(&networkMessage, 0, sizeof(UA_NetworkMessage));
  53. size_t currentPosition = 0;
  54. UA_NetworkMessage_decodeBinary(&buffer, &currentPosition, &networkMessage);
  55. UA_ByteString_clear(&buffer);
  56. /* Is this the correct message type? */
  57. if(networkMessage.networkMessageType != UA_NETWORKMESSAGE_DATASET)
  58. goto cleanup;
  59. /* At least one DataSetMessage in the NetworkMessage? */
  60. if(networkMessage.payloadHeaderEnabled &&
  61. networkMessage.payloadHeader.dataSetPayloadHeader.count < 1)
  62. goto cleanup;
  63. /* Is this a KeyFrame-DataSetMessage? */
  64. for(size_t j = 0; j < networkMessage.payloadHeader.dataSetPayloadHeader.count; j++) {
  65. UA_DataSetMessage *dsm = &networkMessage.payload.dataSetPayload.dataSetMessages[j];
  66. if(dsm->header.dataSetMessageType != UA_DATASETMESSAGE_DATAKEYFRAME)
  67. continue;
  68. /* Loop over the fields and print well-known content types */
  69. for(int i = 0; i < dsm->data.keyFrameData.fieldCount; i++) {
  70. const UA_DataType *currentType = dsm->data.keyFrameData.dataSetFields[i].value.type;
  71. if(currentType == &UA_TYPES[UA_TYPES_BYTE]) {
  72. UA_Byte value = *(UA_Byte *)dsm->data.keyFrameData.dataSetFields[i].value.data;
  73. UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
  74. "Message content: [Byte] \tReceived data: %i", value);
  75. } else if (currentType == &UA_TYPES[UA_TYPES_DATETIME]) {
  76. UA_DateTime value = *(UA_DateTime *)dsm->data.keyFrameData.dataSetFields[i].value.data;
  77. UA_DateTimeStruct receivedTime = UA_DateTime_toStruct(value);
  78. UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
  79. "Message content: [DateTime] \t"
  80. "Received date: %02i-%02i-%02i Received time: %02i:%02i:%02i",
  81. receivedTime.year, receivedTime.month, receivedTime.day,
  82. receivedTime.hour, receivedTime.min, receivedTime.sec);
  83. }
  84. }
  85. }
  86. cleanup:
  87. UA_NetworkMessage_clear(&networkMessage);
  88. return retval;
  89. }
  90. int main(int argc, char **argv) {
  91. signal(SIGINT, stopHandler);
  92. signal(SIGTERM, stopHandler);
  93. UA_PubSubTransportLayer udpLayer = UA_PubSubTransportLayerUDPMP();
  94. UA_PubSubConnectionConfig connectionConfig;
  95. memset(&connectionConfig, 0, sizeof(connectionConfig));
  96. connectionConfig.name = UA_STRING("UADP Connection 1");
  97. connectionConfig.transportProfileUri =
  98. UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-udp-uadp");
  99. connectionConfig.enabled = UA_TRUE;
  100. UA_NetworkAddressUrlDataType networkAddressUrl =
  101. {UA_STRING_NULL , UA_STRING("opc.udp://224.0.0.22:4840/")};
  102. UA_Variant_setScalar(&connectionConfig.address, &networkAddressUrl,
  103. &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]);
  104. UA_PubSubChannel *psc =
  105. udpLayer.createPubSubChannel(&connectionConfig);
  106. psc->regist(psc, NULL, NULL);
  107. UA_StatusCode retval = UA_STATUSCODE_GOOD;
  108. while(running && retval == UA_STATUSCODE_GOOD)
  109. retval = subscriberListen(psc);
  110. psc->close(psc);
  111. return 0;
  112. }