azure / azure-uamqp-c Goto Github PK

Hi all,

I'm trying to use amqp over websocket exploiting the example provided in this lib. I am able to connect and send messages to my IoTHub broker with this protocol.
I did't chenge the parameters for
WSIO_CONFIG ws_io_config = { IOT_HUB_HOST, 443, "AMQPWSB10", "/$iothub/websocket", true };

Firstly I was wandering the meaning of these parameters and if they can be used also for communication with other brokers (such as apollo).

Then I tried to put wrong values in those parameters. But the result is a seh exception with code 0xc00000fd. It seems that if the client is not able to connect to the server due to those parameters, a SEH exception is thrown. Is this an admissible behavior? Is there a way to catch the connection failed before this exception is thrown? (myabe with a callback)

Thank you in advance
Nicolo'

Hi all,

I'm tying to send 100 messages to my apollo broker. Each iteration my client performs: SSL initialization, connection to broker, disconnection from broker. The connection is over websockets and SSL.
The problem is that at a random iteration, the program stucks during the do_work while loop.
Without websockets, the problem does not appear.
I'm using the latest master.
Here is the code that produces the problem:

#include "stdafx.h"

#include <stdlib.h>
#include <stdio.h>
#include <stdbool.h>
#include "azure_c_shared_utility/platform.h"
#include "azure_c_shared_utility/tlsio.h"
#include "azure_uamqp_c/message_sender.h"
#include "azure_uamqp_c/message.h"
#include "azure_uamqp_c/messaging.h"
#include "azure_uamqp_c/amqpalloc.h"
#include "azure_uamqp_c/saslclientio.h"
#include "azure_uamqp_c/sasl_plain.h"
#include "azure_c_shared_utility/wsio.h"
#include "azure_uamqp_c/cbs.h"

#if _WIN32
#include "windows.h"
#endif

static bool message_sent = false;
static bool error_opening_sender = false;
static int counter = 0;

static void on_message_send_complete(void* context, MESSAGE_SEND_RESULT send_result)
{
    (void)send_result;
    (void)context;
    printf("iteration: %d, message sent!\n", counter);
    message_sent = true;
}

int main(int argc, char** argv)
{
    int result;
    (void)argc;
    (void)argv;

    for (counter = 0; counter < 100; counter++)
    {
        amqpalloc_set_memory_tracing_enabled(true);

        if (platform_init() != 0)
        {
            result = -1;
        }
        else
        {
            XIO_HANDLE sasl_io;
            CONNECTION_HANDLE connection;
            SESSION_HANDLE session;
            LINK_HANDLE link;
            MESSAGE_SENDER_HANDLE message_sender;
            MESSAGE_HANDLE message;

            size_t last_memory_used = 0;

            /* create SASL PLAIN handler */
            SASL_PLAIN_CONFIG sasl_plain_config = { USERNAME, PASSWORD, NULL };
            SASL_MECHANISM_HANDLE sasl_mechanism_handle = saslmechanism_create(saslplain_get_interface(), &sasl_plain_config);
            XIO_HANDLE underlying_io;

            /* create the WSIO */
            WSIO_CONFIG wsio_config;
            wsio_config.hostname = BROKER_HOST;
            wsio_config.port = PORT;
            wsio_config.protocol = "amqp";
            wsio_config.resource_name = "/";
            wsio_config.underlying_io_interface = NULL;
            wsio_config.underlying_io_parameters = NULL;
            const IO_INTERFACE_DESCRIPTION* tlsio_interface = wsio_get_interface_description();
            underlying_io = xio_create(tlsio_interface, &wsio_config);

            /* set certificates*/
            std::string trustStore = getTrustStoreApollo();
            xio_setoption(underlying_io, "TrustedCerts", trustStore.c_str());

            /* create the SASL client IO using the TLS IO */
            SASLCLIENTIO_CONFIG sasl_io_config;
            sasl_io_config.underlying_io = underlying_io;
            sasl_io_config.sasl_mechanism = sasl_mechanism_handle;
            sasl_io = xio_create(saslclientio_get_interface_description(), &sasl_io_config);

            /* create the connection, session and link */
            connection = connection_create(sasl_io, BROKER_HOST, "some", NULL, NULL);
            session = session_create(connection, NULL, NULL);
            session_set_incoming_window(session, 2147483647);
            session_set_outgoing_window(session, 65536);

            AMQP_VALUE source = messaging_create_source("ingress");
            AMQP_VALUE target = messaging_create_target("queue://testQueue");
            link = link_create(session, "sender-link", role_sender, source, target);
            link_set_snd_settle_mode(link, sender_settle_mode_unsettled);
            (void)link_set_max_message_size(link, 65536);

            amqpvalue_destroy(source);
            amqpvalue_destroy(target);

            message = message_create();
            unsigned char hello[] = { 'H', 'e', 'l', 'l', 'o' };
            BINARY_DATA binary_data;
            binary_data.bytes = hello;
            binary_data.length = sizeof(hello);
            message_add_body_amqp_data(message, binary_data);

            /* create a message sender */
            message_sender = messagesender_create(link, NULL, NULL);
            if (messagesender_open(message_sender) == 0)
            {
                uint32_t i;

                (void)messagesender_send(message_sender, message, on_message_send_complete, message);

                message_destroy(message);

                message_sent = false;
                while (true)
                {
                    connection_dowork(connection);

                    if (message_sent)
                        break;
                }

            }
            else
            {
                message_destroy(message);
            }

            messagesender_destroy(message_sender);
            link_destroy(link);
            session_destroy(session);
            connection_destroy(connection);
            xio_destroy(sasl_io);
            xio_destroy(underlying_io);
            saslmechanism_destroy(sasl_mechanism_handle);
            platform_deinit();

            result = 0;
        }
    }

    return result;
}

Thank you
Nicolo'

@dcristoloveanu
I have tried to cross compile uamqp lib, but I have got following issue:

azure-uamqp-c-070db5c06ec9677c7963c0f7be06d812029b1295/azure-c-shared-utility/c/src/crt_abstractions.c:12:
azure-uamqp-c-070db5c06ec9677c7963c0f7be06d812029b1295/azure-c-shared-utility/c/inc/crt_abstractions.h:67: error: two or more data types in declaration specifiers
azure-uamqp-c-070db5c06ec9677c7963c0f7be06d812029b1295/azure-c-shared-utility/c/inc/crt_abstractions.h:67: warning: useless type name in empty declaration
In file included from azure-uamqp-c-070db5c06ec9677c7963c0f7be06d812029b1295/azure-c-shared-utility/c/src/crt_abstractions.c:12:
azure-uamqp-c-070db5c06ec9677c7963c0f7be06d812029b1295/azure-c-shared-utility/c/inc/crt_abstractions.h:122:2: error: #error unknown (or C89) compiler, provide ISNAN with the same meaning as isnan in C99 standard
azure-uamqp-c-070db5c06ec9677c7963c0f7be06d812029b1295/azure-c-shared-utility/c/inc/crt_abstractions.h:142:2: error: #error unknown (or C89) compiler, provide INT64_PRINTF with the same meaning as PRIdN in C99 standard
azure-uamqp-c-070db5c06ec9677c7963c0f7be06d812029b1295/azure-c-shared-utility/c/src/crt_abstractions.c:243:2: error: #error for STDC_VERSION undefined (assumed C89), provide own implementation of sprintf_s
make[2]: *** [azure-c-shared-utility/c/CMakeFiles/aziotsharedutil.dir/src/crt_abstractions.c.o] Error 1
make[1]: *** [azure-c-shared-utility/c/CMakeFiles/aziotsharedutil.dir/all] Error 2

Do you have any idea about this?

Thanks and Kind regards

Consider removing composite type from the typesystem, as a composite type is nothing else than a describe type with a list.

Hey guys,

I had some problems when I tried to encode a float with amqpvalue_create_float and figured out that the case statement for AMQP_TYPE_FLOAT is missing in the switch of the function amqpvalue_encode. Is this intended or an issue?

Best regards,
Mario

https://github.com/Azure/azure-uamqp-c/blob/master/build_all/packaging/windows/rebuild_nugets.cmd
line75 & 79 source should be uamqp_x64?

It would be very nice to have an E2E test that runs an uAMQP client and an uAMQP based server (both raw TCP) and checks that at least one message can be send both directions.

Hi all,

I'm trying to send a message with a payload of 5mb to an IoTHub.
The library tells that the message has been sent successfully and no error is flagged in any callback. But the server does not receive the message because of its limitations (messages bigger then 64kb are dropped by the server).
I set

link_set_snd_settle_mode(linkHandle, sender_settle_mode_settled);
link_set_max_message_size(linkHandle, 5000000);

I supposed that the option sender_settle_mode_settled should force the library to tell if the message has been delivered to the endpoint but I am not able to catch any state that indicates that the message has not been delivered. in fact the send result state in the callback of the messagesender_send if finally set to MESSAGE_SEND_OK.

Is there a way to know if the message has been delivered to the endpoint?

It would be very nice to have an E2E test that sends at least one message to an IoT hub proving that this scenario still works for each build.

There is no need to run by default all the tests of all the submodules.
Thus it would be very nice to have a cmake option that would say run_all_tests.
If this options is OFF then run_unittests, run_int_tests and run_e2e_tests should be disabled before including any of the submodules. That way all the tests in ctest, umock, c shared utility et all would not be run unless explicitly asked for.

Hey guys,

we are currently testing our usage of the implementation with the broker of opcfoundation-prototyping.servicebus.windows.net. In his attach message, this broker sends a max-message-size of 65536. When the sent data exceeds this value, an link-error "max-message-size-exceeded" should arise according to the AMQP specification. In that case it is necessary to implement a segmentation mechanism in the application using your AMQP stack.

Currently the max-message-size is stored nowhere in the link structure such that it can be read by the application. Is this a mistake or are there other ways to access the max-message-size information sent by the broker?

Best regards,
Mario

This leak shows up when running the message_sender_sample.
It occurs because the realloc to 0 returns a NULL pointer but does not free the original block.

I have fixed with an if-else that checks for this case but is that the right fix?
Should amqpalloc_realloc be fixed?

It would be much nicer and cleaner to have an CONST_AMQP_VALUE module. This would be refcounted and read only.

That way any chances of modifying inadvertently underlying data due to having refcounting would not exist.

Currently the E2E test for IoTHub simply relies on using a certain device name.
That should be changed to create a device every time the test is run.

There are several places where const AMQP_VALUE is used. See amqp_frame codec for example.
These usages should be changed to AMQP_VALUE, as there is no meaning to applying const in this case.

TLS server IO in order to allow a server to use the TLS layer

A minor thing, but should be done at some point.

This is regarding the need for calling amqpvalue_create_message_annotations before passing the AMQP_VALUE to message_set_message_annotations().

In order to write tests for one SASL mechanism in the SASL init frame, a SASL server IO is needed.
This depends on header IO being rewritten first.

• With tests
• Multiple headers should be possible to be detected (SASL, AMQP). This is needed to implement SASL on server side.

I noticed that in the samples all IO operations (connection, send a message, authenticate, create message sender) are performed by a "connection_dowork" function that schedules the internal operations. This function is always enclosed in an endless while loop that breaks because of the flags set by the callbacks.

The drawback of this pattern is that each operation performed by the "connection_dowork" function within the while loop stresses the CPU (100% usage). In fact no asynchronous operation is performed by the "connection_dowork" function to prevent from stressing the CPU.

Is this the way this library is meant to be used? Are there any alternatives to use asynchronous operations in order to avoid stressing the CPU in the endless while loop?

Thanks in advance
Nicolò

Are you planning to use azure-uamqp-c for producton instead of qpid-proton-c?

Is it supposed to work without memory tracing?

If I try to build azure-uamqp-c with clang, I get lots of warnings (and if I fix some of them, more appear).

azure-uamqp-c should compile cleanly with clang.

I get crashes in trace_free when I access the library simultaneously from multiple threads.
It looks like it is because trace_free is not thread safe.
Does this sound like a correct diagnosis?

when running receive sample for amqp, the cpu usage is high. I try to debug with strace
~ # strace -c -p 7713
Process 7713 attached - interrupt to quit
^CProcess 7713 detached
% time seconds usecs/call calls errors syscall

68.44 0.093948 1 104773 104773 recv
31.56 0.043331 0 104773 times

100.00 0.137279 209546 104773 total

I could see that the recv in socketio_berkeley.c
received = recv(socket_io_instance->socket, recv_bytes, sizeof(recv_bytes), 0);
is called permanently and so causing high cpu usage.
Any help is appreciated!
Thanks

I clone the repo and built the lib according to the defined guidelines.
I took the send sample and broke it out into individual methods init(), send(), dispose().
The init() sets up the connections.
The send() sends a message.
The dispose() cleans up.
I then wrote a c++ library that wrapped the c code and tried to publish 100 messages using a single send() not a batch send() non async for either since message order is very important to me, and I am getting rates of around 20 messages per second which is quite slow. Any suggestions? Could this be due to type of event hubs plan I am paying for? Or does the ack actually take that long? I am using a partition key with the message.

I get this error when receiving messages from ActiveMQ:

azure-uamqp-c\src\link.c Func:_link_frame_received Line:323 Could not get the more field from the transfer performative

Note that the same version of ActiveMQ worked fine with an earlier version of uAQMP so whatever is going wrong is related to changes in the uAQMP library.

The headers in the message are as follows:

message-id: 1234567890:10
subject: ua-data
content-type: application/opcua+uajson
user-id: uAMQP
group-id: MyDataSetWriter
group-sequence: 10
ua-class-id: MyDataSetClass
ua-metadata-node-name: MetaDataTopicName

The message body is a JSON document < 1000 bytes.

Any hints or suggestions?

Each of these should give a different error message, and they should not terminate the SESSION.

		if ((attach_get_name(attach_handle, &name) != 0) ||
			(attach_get_role(attach_handle, &role) != 0) ||
			(attach_get_source(attach_handle, &source) != 0) ||
			(attach_get_target(attach_handle, &target) != 0))
		{
			end_session_with_error(session_instance, "amqp:decode-error", "Cannot get link name from ATTACH frame");
		}

Hey guys,

according to the AMQP specifiction (http://www.amqp.org/sites/amqp.org/files/amqp.pdf) at page 54, the field max-frame-size is not mandatory. In that case the AMQP-Type NULL is used for the transmission of this value. However, the function get_max_frame_size in amqp_definitions.c returns an error in that case.

I think the if statement should be the other way round: when the type is NULL, the max-frame-size should be set to 4294967295u, otherwise an error shall be returned.

Hi doing the following

git clone [email protected]:Azure/azure-uamqp-c.git --recursive
mkdir build-uamqp
cd build-uamqp
cmake ../azure-uamqp-c/ -Duse_wsio=ON -DcompileOption_C="-std=c99"

Lead to numerous compile errors, apprently in the shared utility test framework

/home/ubuntu/workspace/azure-uamqp-c/azure-c-shared-utility/c/testtools/ctest/inc/ctest.h:206:64: warning: invalid conversion from ‘void (*)(CONCRETE_IO_HANDLE) {aka void (*)(void*)}’ to ‘const void*’ [-fpermissive]
     type##_ToString(expectedString, sizeof(expectedString), (A)); \

Adding -DcompileOption_CXX="-fpermissive" to cmake command line hide problem, but this sound weird to me...

Hi all!

I'm trying to use azure-uamqp-c with an Apollo broker (https://activemq.apache.org/apollo/) running on my machine.

First question: is this even possible?

If this is the case, I'm unable to find the right way to set the parameters for the amqp connection.
I was able to create a simple working example in C# by using the AMQP.Net Lite library (https://github.com/Azure/amqpnetlite) to send a message to the Apollo broker:

        // Connect to the Apollo broker on localhost:5672
        var address = new Address("amqp://admin:[email protected]:5672");
        var connection = new Connection(address);
        // Begin an AMQP session
        var session = new Session(connection);
        // Create an AMQP link to send data and attach it to the "testQueue" queue in Apollo
        var sender = new SenderLink(session, "senderLink", "queue://testQueue");
        // Create a new message to be sent
        var message = new Message("Test message from AMQP Net Lite");
        // Send the message
        await sender.SendAsync(message);
        sendTime = DateTime.Now;

I'm at a loss as to where specify the "queue://testQueue" parameter in uamqp.
In the uaqmp examples linking is carried out using the "sender-link" string, which I assume is roughly the same as the "senderLink" string in the sample C# code. Am I correct?

Thanks in andvance
Nicola

Composite type in amqpvalue should have unit tests (currently it is not covered).

I am trying to listen to the IoT Hub device messenger endpoint using the listener sample. I have removed the connection strings but you get the idea. There is no error just no messages recieved:

#include <stdlib.h>
#ifdef _CRTDBG_MAP_ALLOC
#include <crtdbg.h>
#endif
#include <stdio.h>
#include <stdbool.h>
#include "azure_c_shared_utility/platform.h"
#include "azure_c_shared_utility/tlsio.h"
#include "azure_uamqp_c/message_receiver.h"
#include "azure_uamqp_c/message.h"
#include "azure_uamqp_c/messaging.h"
#include "azure_uamqp_c/amqpalloc.h"
#include "azure_uamqp_c/saslclientio.h"
#include "azure_uamqp_c/sasl_plain.h"
#include "azure_uamqp_c/consolelogger.h"

/* This sample connects to an Event Hub, authenticates using SASL PLAIN (key name/key) and then it received all messages for partition 0 */
/* Replace the below settings with your own.*/

#define EH_HOST "<<<Replace with your own EH host (like myeventhub.servicebus.windows.net)>>>"
#define EH_KEY_NAME "<<<Replace with your own key name>>>"
#define EH_KEY "<<<Replace with your own key>>>"

static AMQP_VALUE on_message_received(const void* context, MESSAGE_HANDLE message)
{
    (void)message;
    (void)context;

    printf("Message received.\r\n");

    return messaging_delivery_accepted();
}

int main(int argc, char** argv)
{
    int result;
    XIO_HANDLE sasl_io = NULL;
    CONNECTION_HANDLE connection = NULL;
    SESSION_HANDLE session = NULL;
    LINK_HANDLE link = NULL;
    MESSAGE_RECEIVER_HANDLE message_receiver = NULL;

    amqpalloc_set_memory_tracing_enabled(true);

    if (platform_init() != 0)
    {
        result = -1;
    }
    else
    {
        size_t last_memory_used = 0;

        /* create SASL plain handler */
        SASL_PLAIN_CONFIG sasl_plain_config = { EH_KEY_NAME, EH_KEY, NULL };
        SASL_MECHANISM_HANDLE sasl_mechanism_handle = saslmechanism_create(saslplain_get_interface(), &sasl_plain_config);
        XIO_HANDLE tls_io;

        /* create the TLS IO */
        TLSIO_CONFIG tls_io_config = { EH_HOST, 5671 };
        const IO_INTERFACE_DESCRIPTION* tlsio_interface = platform_get_default_tlsio();
        tls_io = xio_create(tlsio_interface, &tls_io_config, NULL);

        /* create the SASL client IO using the TLS IO */
        SASLCLIENTIO_CONFIG sasl_io_config = { tls_io, sasl_mechanism_handle };
        sasl_io = xio_create(saslclientio_get_interface_description(), &sasl_io_config, NULL);

        /* create the connection, session and link */
        connection = connection_create(sasl_io, EH_HOST, "whatever", NULL, NULL);
        session = session_create(connection, NULL, NULL);

        /* set incoming window to 100 for the session */
        session_set_incoming_window(session, 100);

                /* THIS IS WHERE I ADDED MY CHANGES */
        AMQP_VALUE source = messaging_create_source("amqps://" EH_HOST "/myhubname");
        AMQP_VALUE target = messaging_create_target("/messages/events");
        link = link_create(session, "receiver-link", role_receiver, source, target);
        link_set_rcv_settle_mode(link, receiver_settle_mode_first);
        amqpvalue_destroy(source);
        amqpvalue_destroy(target);

        /* create a message receiver */
        message_receiver = messagereceiver_create(link, NULL, NULL);
        if ((message_receiver == NULL) ||
            (messagereceiver_open(message_receiver, on_message_received, message_receiver) != 0))
        {
            result = -1;
        }
        else
        {
            while (true)
            {
                size_t current_memory_used;
                size_t maximum_memory_used;
                connection_dowork(connection);

                current_memory_used = amqpalloc_get_current_memory_used();
                maximum_memory_used = amqpalloc_get_maximum_memory_used();

                if (current_memory_used != last_memory_used)
                {
                    printf("Current memory usage:%lu (max:%lu)\r\n", (unsigned long)current_memory_used, (unsigned long)maximum_memory_used);
                    last_memory_used = current_memory_used;
                }
            }

            result = 0;
        }

        messagereceiver_destroy(message_receiver);
        link_destroy(link);
        session_destroy(session);
        connection_destroy(connection);
        platform_deinit();

        printf("Max memory usage:%lu\r\n", (unsigned long)amqpalloc_get_maximum_memory_used());
        printf("Current memory usage:%lu\r\n", (unsigned long)amqpalloc_get_current_memory_used());

#ifdef _CRTDBG_MAP_ALLOC
        _CrtDumpMemoryLeaks();
#endif
    }

    return result;
}

The client is not able to create the amqp connection:
azure-uamqp-c-b656452878e391dbac09d514f0c628821c950ccc/azure-c-shared-utility/adapters/socketio_berkeley.c Func:socketio_open Line:274 Failure: getaddrinfo failure -2.
azure-uamqp-c-b656452878e391dbac09d514f0c628821c950ccc/azure-c-shared-utility/adapters/socketio_berkeley.c Func:socketio_open Line:274 Failure: getaddrinfo failure -2.

Many SASLclientIO unit tests are currently disabled.
Also some cleanup is needed in the spec of the module.

Currently the VS2008 build does not build with -Duse_openssl:bool=ON
This should be done in order to avoid errors on older compilers.

If the put-token callback is accessing any memory that has already been freed, it can cause an A/V (obvious, obviously).
What do you think about the possible solutions for this scenario?
A. The upper layer has to then either wait for all operations to complete before destroying (not that cool), or
B. destroy cbs first and then free the memory accessed by the put-token callback (also not sure if that's the best way), OR
C. add an api to set/unset the callback on cbs instance, OR
D. cbs_destroy not to invoke callbacks for pending operations, OR
E. Any other ideas?

What recommendations are there for using this library from multiple threads?
e.g. one connection - one thread? one thread per process? one message_sender per thread?
Looking at the code it does not look thread safe.

Because bool is a built in type in C++, it is not necessary to include cstdbool.

I modified the message_sender and message receiver samples to try to set/send and receive/get the group_id (SessionId in service bus explorer).

If I send a message to my queue with service bus explorer while setting SessionId field, I can run message_receiver_sample, receive messages and get the SessionId by using properties_get_group_id.

If I try to use properties_set_group_id in message_sender_sample, service bus explorer shows messages in the queue but I can't receive them with either service bus explorer or message_receiver_sample. The messages just seem to get stuck in the queue forever until I delete the queue.

See attached samples (I removed my access key but left in my host/queue name to show what I did).

message_sender_receiver.zip

alf

Hey Guys,

I think found another small issue in connection.c. In the function connection_handle_deadlines, the time until the configured idle timeout expires is calculated from the current milliseconds and connection->last_frame_received_time. The problem is that connection->last_frame_received_time is not initialized in connection_create2 which results in "No frame received for the idle timeout" when an idle timeout was specified. To fix this it should be enough to use calloc instead of malloc in the function connection_create2, right?

Best regards,
Mario

Hey guys,

this is not an issue, it is just a question :)

I saw that you were working on a small AMQP as well as a small MQTT (umqtt) implementation that can be used in embedded systems. I just wanted to ask you if you also developed such a library for XMPP?

Best regards,
Mario

amqpvalue_get_sasl_mechanisms (Line 10167) attempts to get the array or symbol and will report an error even though the operation succeeds when the type is a symbol.

Hey guys,

we currently try to connect to an existing device of our Azure IoTHub. Unfortunately we always get an error during the SASL authentication: in sasl.outcome the code field is set to 2 indicating a system error.

The sasl authentication looks like this:
- Username: [email protected]
- Password: SharedAccessSignature=SharedAccessSignature sr=IOTHUBCMIC.azure-devices.net%2Fdevices%2Ftest-001&sig={sig}&se=1495780171

whereas the password is the SAS token generated by the DeviceExplorer. I got the feeling that there is some fault in our connection string, but I got no idea which one :) can somebody help me out?

Best regards,
Mario

saslclientio treats the sasl-server-mechanisms in the sasl_mechanisms performative as an array (mandatory).
That is not correct.

The field sasl-server-mechanisms is multiple symbol, so it can be simply one symbol.