PubSub: Add standalone subscriber

examples/CMakeLists.txt

@@ -164,5 +164,6 @@ if(UA_ENABLE_PUBSUB)
         message(WARNING "PubSub subscriber tutorial (preview) can not be used with AMALGAMATION. Skipping tutorial_pubsub_subscribe.")
     else(NOT UA_ENABLE_AMALGAMATION)
         add_example(tutorial_pubsub_subscribe pubsub/tutorial_pubsub_subscribe.c)
+        add_example(pubsub_subscribe_standalone pubsub/pubsub_subscribe_standalone.c)
     endif()
 endif()

examples/pubsub/pubsub_subscribe_standalone.c

@@ -0,0 +1,131 @@
+/* This work is licensed under a Creative Commons CCZero 1.0 Universal License.
+ * See http://creativecommons.org/publicdomain/zero/1.0/ for more information.
+ */
+
+/**
+ * IMPORTANT ANNOUNCEMENT
+ * The PubSub subscriber API is currently not finished. This examples can be used to receive
+ * and print the values, which are published by the tutorial_pubsub_publish example.
+ * The following code uses internal API which will be later replaced by the higher-level
+ * PubSub subscriber API.
+ */
+
+#include <signal.h>
+#include "ua_pubsub_networkmessage.h"
+#include "ua_log_stdout.h"
+#include "ua_config_default.h"
+#include "ua_pubsub.h"
+#include "ua_network_pubsub_udp.h"
+#ifdef UA_ENABLE_PUBSUB_ETH_UADP
+#include "ua_network_pubsub_ethernet.h"
+#endif
+#include "src_generated/ua_types_generated.h"
+
+UA_Boolean running = true;
+static void stopHandler(int sign) {
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
+                "received ctrl-c");
+    running = false;
+}
+
+static UA_StatusCode
+subscriberListen(UA_PubSubChannel *psc) {
+    UA_ByteString buffer;
+    UA_StatusCode retval = UA_ByteString_allocBuffer(&buffer, 512);
+    if(retval != UA_STATUSCODE_GOOD) {
+        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER,
+                     "Message buffer allocation failed!");
+        return retval;
+    }
+
+    /* Receive the message. Blocks for 100ms */
+    retval = psc->receive(psc, &buffer, NULL, 100);
+    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_clear(&buffer);
+        return UA_STATUSCODE_GOOD;
+    }
+
+    /* Decode the message */
+    UA_LOG_INFO(UA_Log_Stdout, UA_LOGCATEGORY_USERLAND,
+                "Message length: %lu", (unsigned long) buffer.length);
+    UA_NetworkMessage networkMessage;
+    memset(&networkMessage, 0, sizeof(UA_NetworkMessage));
+    size_t currentPosition = 0;
+    UA_NetworkMessage_decodeBinary(&buffer, &currentPosition, &networkMessage);
+    UA_ByteString_clear(&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? */
+    for(size_t j = 0; j < networkMessage.payloadHeader.dataSetPayloadHeader.count; j++) {
+        UA_DataSetMessage *dsm = &networkMessage.payload.dataSetPayload.dataSetMessages[j];
+        if(dsm->header.dataSetMessageType != UA_DATASETMESSAGE_DATAKEYFRAME)
+            continue;
+
+        /* 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_clear(&networkMessage);
+    return retval;
+}
+
+int main(int argc, char **argv) {
+    signal(SIGINT, stopHandler);
+    signal(SIGTERM, stopHandler);
+
+    UA_PubSubTransportLayer udpLayer = UA_PubSubTransportLayerUDPMP();
+
+    UA_PubSubConnectionConfig connectionConfig;
+    memset(&connectionConfig, 0, sizeof(connectionConfig));
+    connectionConfig.name = UA_STRING("UADP Connection 1");
+    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_PubSubChannel *psc =
+        udpLayer.createPubSubChannel(&connectionConfig);
+    psc->regist(psc, NULL);
+
+    UA_StatusCode retval = UA_STATUSCODE_GOOD;
+    while(running && retval == UA_STATUSCODE_GOOD)
+        retval = subscriberListen(psc);
+
+    psc->close(psc);
+        
+    return 0;
+}