tutorial_pubsub_subscribe.c 7.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193
  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 "ua_pubsub_networkmessage.h"
  12. #include "ua_log_stdout.h"
  13. #include "ua_server.h"
  14. #include "ua_config_default.h"
  15. #include "ua_pubsub.h"
  16. #include "ua_network_pubsub_udp.h"
  17. #ifdef UA_ENABLE_PUBSUB_ETH_UADP
  18. #include "ua_network_pubsub_ethernet.h"
  19. #endif
  20. #include "src_generated/ua_types_generated.h"
  21. #include <stdio.h>
  22. #include <signal.h>
  23. UA_Boolean running = true;
  24. static void stopHandler(int sign) {
  25. UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "received ctrl-c");
  26. running = false;
  27. }
  28. static void
  29. subscriptionPollingCallback(UA_Server *server, UA_PubSubConnection *connection) {
  30. UA_ByteString buffer;
  31. if (UA_ByteString_allocBuffer(&buffer, 512) != UA_STATUSCODE_GOOD) {
  32. UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
  33. "Message buffer allocation failed!");
  34. return;
  35. }
  36. /* Receive the message. Blocks for 5ms */
  37. UA_StatusCode retval =
  38. connection->channel->receive(connection->channel, &buffer, NULL, 5);
  39. if(retval != UA_STATUSCODE_GOOD || buffer.length == 0) {
  40. /* Workaround!! Reset buffer length. Receive can set the length to zero.
  41. * Then the buffer is not deleted because no memory allocation is
  42. * assumed.
  43. * TODO: Return an error code in 'receive' instead of setting the buf
  44. * length to zero. */
  45. buffer.length = 512;
  46. UA_ByteString_clear(&buffer);
  47. return;
  48. }
  49. /* Decode the message */
  50. UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
  51. "Message length: %lu", (unsigned long) buffer.length);
  52. UA_NetworkMessage networkMessage;
  53. memset(&networkMessage, 0, sizeof(UA_NetworkMessage));
  54. size_t currentPosition = 0;
  55. UA_NetworkMessage_decodeBinary(&buffer, &currentPosition, &networkMessage);
  56. UA_ByteString_clear(&buffer);
  57. /* Is this the correct message type? */
  58. if(networkMessage.networkMessageType != UA_NETWORKMESSAGE_DATASET)
  59. goto cleanup;
  60. /* At least one DataSetMessage in the NetworkMessage? */
  61. if(networkMessage.payloadHeaderEnabled &&
  62. networkMessage.payloadHeader.dataSetPayloadHeader.count < 1)
  63. goto cleanup;
  64. /* Is this a KeyFrame-DataSetMessage? */
  65. UA_DataSetMessage *dsm = &networkMessage.payload.dataSetPayload.dataSetMessages[0];
  66. if(dsm->header.dataSetMessageType != UA_DATASETMESSAGE_DATAKEYFRAME)
  67. goto cleanup;
  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. cleanup:
  86. UA_NetworkMessage_clear(&networkMessage);
  87. }
  88. static int
  89. run(UA_String *transportProfile, UA_NetworkAddressUrlDataType *networkAddressUrl) {
  90. signal(SIGINT, stopHandler);
  91. signal(SIGTERM, stopHandler);
  92. UA_ServerConfig *config = UA_ServerConfig_new_minimal(4801, NULL);
  93. /* Details about the PubSubTransportLayer can be found inside the
  94. * tutorial_pubsub_connection */
  95. config->pubsubTransportLayers = (UA_PubSubTransportLayer *)
  96. UA_calloc(2, sizeof(UA_PubSubTransportLayer));
  97. if (!config->pubsubTransportLayers) {
  98. UA_ServerConfig_delete(config);
  99. return -1;
  100. }
  101. config->pubsubTransportLayers[0] = UA_PubSubTransportLayerUDPMP();
  102. config->pubsubTransportLayersSize++;
  103. #ifdef UA_ENABLE_PUBSUB_ETH_UADP
  104. config->pubsubTransportLayers[1] = UA_PubSubTransportLayerEthernet();
  105. config->pubsubTransportLayersSize++;
  106. #endif
  107. UA_Server *server = UA_Server_new(config);
  108. UA_PubSubConnectionConfig connectionConfig;
  109. memset(&connectionConfig, 0, sizeof(connectionConfig));
  110. connectionConfig.name = UA_STRING("UADP Connection 1");
  111. connectionConfig.transportProfileUri = *transportProfile;
  112. connectionConfig.enabled = UA_TRUE;
  113. UA_Variant_setScalar(&connectionConfig.address, networkAddressUrl,
  114. &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]);
  115. UA_NodeId connectionIdent;
  116. UA_StatusCode retval =
  117. UA_Server_addPubSubConnection(server, &connectionConfig, &connectionIdent);
  118. if(retval == UA_STATUSCODE_GOOD)
  119. UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
  120. "The PubSub Connection was created successfully!");
  121. /* The following lines register the listening on the configured multicast
  122. * address and configure a repeated job, which is used to handle received
  123. * messages. */
  124. UA_PubSubConnection *connection =
  125. UA_PubSubConnection_findConnectionbyId(server, connectionIdent);
  126. if(connection != NULL) {
  127. UA_StatusCode rv = connection->channel->regist(connection->channel, NULL);
  128. if (rv == UA_STATUSCODE_GOOD) {
  129. UA_UInt64 subscriptionCallbackId;
  130. UA_Server_addRepeatedCallback(server, (UA_ServerCallback)subscriptionPollingCallback,
  131. connection, 100, &subscriptionCallbackId);
  132. } else {
  133. UA_LOG_WARNING(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "register channel failed: %s!",
  134. UA_StatusCode_name(rv));
  135. }
  136. }
  137. retval |= UA_Server_run(server, &running);
  138. UA_Server_delete(server);
  139. UA_ServerConfig_delete(config);
  140. return (int)retval;
  141. }
  142. static void
  143. usage(char *progname) {
  144. printf("usage: %s <uri> [device]\n", progname);
  145. }
  146. int main(int argc, char **argv) {
  147. UA_String transportProfile =
  148. UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-udp-uadp");
  149. UA_NetworkAddressUrlDataType networkAddressUrl =
  150. {UA_STRING_NULL , UA_STRING("opc.udp://224.0.0.22:4840/")};
  151. if (argc > 1) {
  152. if (strcmp(argv[1], "-h") == 0) {
  153. usage(argv[0]);
  154. return 0;
  155. } else if (strncmp(argv[1], "opc.udp://", 10) == 0) {
  156. networkAddressUrl.url = UA_STRING(argv[1]);
  157. } else if (strncmp(argv[1], "opc.eth://", 10) == 0) {
  158. transportProfile =
  159. UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-eth-uadp");
  160. if (argc < 3) {
  161. printf("Error: UADP/ETH needs an interface name\n");
  162. return 1;
  163. }
  164. networkAddressUrl.networkInterface = UA_STRING(argv[2]);
  165. networkAddressUrl.url = UA_STRING(argv[1]);
  166. } else {
  167. printf("Error: unknown URI\n");
  168. return 1;
  169. }
  170. }
  171. return run(&transportProfile, &networkAddressUrl);
  172. }