WIP: pubsub over amqp

[waheedejaz]

Up for the initial review.

Basic PubSub publish over AMQP is added, works with ay AMQP 1.0 broker

NOTE:
The build system is adapted to link the qpid proton lib. At the moment proton c doesn't successfully build using all the warnings enabled in the open62541 build. This will be fixed later.

#3849

[ccvca]

Maybe resolve the compile issues by using "libqpid-proton11-dev" and find_package()? Then there is no need to include all the sources in the open62541-project. mbedTLS is included the same way at the moment.

[jpfr]

Hey there!

Wow, this looks good!
You say that proton does not support all the compiler warnings that we enable by default.
I say that we don't care. We cannot maintain 28,000 lines of code of external dependency that we don't even know.

Most systems can install proton as a dependency.
So let's handle it like we handle mbedtls/openssl.

https://packages.debian.org/stable/libs/libqpid-proton-cpp12

We can have a closer look when the line-count when only the changes to open62541 itself are in the PR.

[waheedejaz]

Proton lib is linked using findPackage(). The previous commit is removed. PR rebased

[jpfr]

Thanks. We will review when the [WIP] is removed from the title.
Keep up the good work!

[basyskom-meerkoetter]

would this leak amqpCtx on the error path (potentially)?

[basyskom-meerkoetter]

also here. amqpCtx and either ua_connection or driver would leak?

[basyskom-meerkoetter]

leaks resources

[jackybek]

I tested your code but realised that the prefix opc.amqp:https://xxx.xxx.xxx.xxx:5672 was not changed to opc.tcp:https://xxx.xxx.xxx.xxx:5672 during runtime.
As compared to the implementation by kalycito on pubsub mqtt, when I pass in opc.mqtt:https://xxx.xxx.xxx.xxx:1883 the code changes the prefix to opc.tcp before sending to the mqtt broker.
This caused failure when open62541 tries to extract the host name from the URL. The host name returned was ‘empty string’.

To force the test to continue, i manually set the prefix to opc.tcp:https://....
a. ua_amqp_adaptor.c: UA_AmqpConnect() :
b. changes made:
UA_String myAddressUrl = UA_STRING("opc.tcp:https://192.168.1.22:5672");
UA_Connection connection = UA_ClientConnectionTCP_init( conf, myAddressUrl, 1000, NULL);
//UA_Connection connection = UA_ClientConnectionTCP_init( conf, address.url, 1000, NULL);
c. During testing, i got the following error:

[2021-04-30 07:25:59.616 (UTC+0800)] error/server ua_amqp_adaptor.c :: yieldAmqp: _write failed. ret_code=BadConnectionClosed
[2021-04-30 07:26:04.116 (UTC+0800)] info/userland received ctrl-c
[2021-04-30 07:26:04.116 (UTC+0800)] info/network Shutting down the TCP network lay

on closer look, the error is due to ctx->ua_connection.state returning a value of 1

In yieldAmqp function: inspecting ctx->ua_connection
ctx->ua_connection.state : 1

---

0 = UA_CONNECTIONSTATE_CLOSED : The socket has been closed and the connection will be deleted
1 = UA_CONNECTIONSTATE_OPENING : The socket is open, but the HEL/ACK handshake is not done
2 = UA_CONNECTIONSTATE_ESTABLISHED : The socket is open and the connection configured

---

ctx->ua_connection.channel : 0
ctx->ua_connection.sockfd : 5
ctx->ua_connection.openingDate : 0
ctx->ua_connection.handle :
......................................................
ctx->openLink : 1
ctx->sender_link_ready : 0
ctx->message : (null)
ctx->send_buffer (pn_rwbytes_t) : 0
ctx->message_buffer (pn_rwbytes_t) : 0
ctx->sequence_no : 0
ctx->acknowledged_no : 0

NB:
Other changes made to conduct the above tests:
open62541.c : UA_parseEndpointUrl() : add the check for "opc.amqp:https://"
Note: the code that follows is supposed to change the prefix to opc.tcp. It worked for opc.mqtt but it failed for opc.amqp

open62541.c : UA_ClientConnectionTCP_init() : add the following statement at the end of this function: connection.sockfd = socket(AF_INET, SOCK_STREAM, 0);