A small memory footprint AMQP implimentation
Dependents: iothub_client_sample_amqp remote_monitoring simplesample_amqp
Diff: message_sender.c
- Revision:
- 0:6ae2f7bca550
- Child:
- 1:eab586236bfe
diff -r 000000000000 -r 6ae2f7bca550 message_sender.c --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/message_sender.c Fri Apr 08 12:01:10 2016 -0700 @@ -0,0 +1,682 @@ +// Copyright (c) Microsoft. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +#include <stdlib.h> +#ifdef _CRTDBG_MAP_ALLOC +#include <crtdbg.h> +#endif +#include <string.h> +#include "azure_uamqp_c/message_sender.h" +#include "azure_uamqp_c/amqpalloc.h" +#include "azure_c_shared_utility/xlogging.h" +#include "azure_uamqp_c/amqpvalue_to_string.h" + +typedef enum MESSAGE_SEND_STATE_TAG +{ + MESSAGE_SEND_STATE_NOT_SENT, + MESSAGE_SEND_STATE_PENDING +} MESSAGE_SEND_STATE; + +typedef enum SEND_ONE_MESSAGE_RESULT_TAG +{ + SEND_ONE_MESSAGE_OK, + SEND_ONE_MESSAGE_ERROR, + SEND_ONE_MESSAGE_BUSY +} SEND_ONE_MESSAGE_RESULT; + +typedef struct MESSAGE_WITH_CALLBACK_TAG +{ + MESSAGE_HANDLE message; + ON_MESSAGE_SEND_COMPLETE on_message_send_complete; + void* context; + MESSAGE_SENDER_HANDLE message_sender; + MESSAGE_SEND_STATE message_send_state; +} MESSAGE_WITH_CALLBACK; + +typedef struct MESSAGE_SENDER_INSTANCE_TAG +{ + LINK_HANDLE link; + size_t message_count; + MESSAGE_WITH_CALLBACK** messages; + MESSAGE_SENDER_STATE message_sender_state; + ON_MESSAGE_SENDER_STATE_CHANGED on_message_sender_state_changed; + void* on_message_sender_state_changed_context; + LOGGER_LOG logger_log; +} MESSAGE_SENDER_INSTANCE; + +static void remove_pending_message_by_index(MESSAGE_SENDER_INSTANCE* message_sender_instance, size_t index) +{ + MESSAGE_WITH_CALLBACK** new_messages; + + if (message_sender_instance->messages[index]->message != NULL) + { + message_destroy(message_sender_instance->messages[index]->message); + message_sender_instance->messages[index]->message = NULL; + } + + amqpalloc_free(message_sender_instance->messages[index]); + + if (message_sender_instance->message_count - index > 1) + { + (void)memmove(&message_sender_instance->messages[index], &message_sender_instance->messages[index + 1], sizeof(MESSAGE_WITH_CALLBACK*) * (message_sender_instance->message_count - index - 1)); + } + + message_sender_instance->message_count--; + + if (message_sender_instance->message_count > 0) + { + new_messages = (MESSAGE_WITH_CALLBACK**)amqpalloc_realloc(message_sender_instance->messages, sizeof(MESSAGE_WITH_CALLBACK*) * (message_sender_instance->message_count)); + if (new_messages != NULL) + { + message_sender_instance->messages = new_messages; + } + } + else + { + amqpalloc_free(message_sender_instance->messages); + message_sender_instance->messages = NULL; + } +} + +static void remove_pending_message(MESSAGE_SENDER_INSTANCE* message_sender_instance, MESSAGE_WITH_CALLBACK* message_with_callback) +{ + size_t i; + + for (i = 0; i < message_sender_instance->message_count; i++) + { + if (message_sender_instance->messages[i] == message_with_callback) + { + remove_pending_message_by_index(message_sender_instance, i); + break; + } + } +} + +static void on_delivery_settled(void* context, delivery_number delivery_no) +{ + MESSAGE_WITH_CALLBACK* message_with_callback = (MESSAGE_WITH_CALLBACK*)context; + MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_with_callback->message_sender; + + if (message_with_callback->on_message_send_complete != NULL) + { + message_with_callback->on_message_send_complete(message_with_callback->context, MESSAGE_SEND_OK); + } + + remove_pending_message(message_sender_instance, message_with_callback); +} + +static int encode_bytes(void* context, const unsigned char* bytes, size_t length) +{ + PAYLOAD* payload = (PAYLOAD*)context; + (void)memcpy((unsigned char*)payload->bytes + payload->length, bytes, length); + payload->length += length; + return 0; +} + +static void log_message_chunk(MESSAGE_SENDER_INSTANCE* message_sender_instance, const char* name, AMQP_VALUE value) +{ + if (message_sender_instance->logger_log != NULL) + { + char* value_as_string = NULL; + LOG(message_sender_instance->logger_log, 0, "%s", name); + LOG(message_sender_instance->logger_log, 0, "%s", (value_as_string = amqpvalue_to_string(value))); + if (value_as_string != NULL) + { + amqpalloc_free(value_as_string); + } + } +} + +static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message_sender_instance, MESSAGE_WITH_CALLBACK* message_with_callback, MESSAGE_HANDLE message) +{ + SEND_ONE_MESSAGE_RESULT result; + + size_t encoded_size; + size_t total_encoded_size = 0; + MESSAGE_BODY_TYPE message_body_type; + message_format message_format; + + if ((message_get_body_type(message, &message_body_type) != 0) || + (message_get_message_format(message, &message_format) != 0)) + { + result = SEND_ONE_MESSAGE_ERROR; + } + else + { + // header + HEADER_HANDLE header; + AMQP_VALUE header_amqp_value; + PROPERTIES_HANDLE properties; + AMQP_VALUE properties_amqp_value; + AMQP_VALUE application_properties; + AMQP_VALUE application_properties_value; + AMQP_VALUE body_amqp_value = NULL; + size_t body_data_count; + + message_get_header(message, &header); + header_amqp_value = amqpvalue_create_header(header); + if (header != NULL) + { + amqpvalue_get_encoded_size(header_amqp_value, &encoded_size); + total_encoded_size += encoded_size; + } + + // properties + message_get_properties(message, &properties); + properties_amqp_value = amqpvalue_create_properties(properties); + if (properties != NULL) + { + amqpvalue_get_encoded_size(properties_amqp_value, &encoded_size); + total_encoded_size += encoded_size; + } + + // application properties + message_get_application_properties(message, &application_properties); + application_properties_value = amqpvalue_create_application_properties(application_properties); + if (application_properties != NULL) + { + amqpvalue_get_encoded_size(application_properties_value, &encoded_size); + total_encoded_size += encoded_size; + } + + result = SEND_ONE_MESSAGE_OK; + + // body - amqp data + switch (message_body_type) + { + default: + result = SEND_ONE_MESSAGE_ERROR; + break; + + case MESSAGE_BODY_TYPE_VALUE: + { + AMQP_VALUE message_body_amqp_value; + if (message_get_inplace_body_amqp_value(message, &message_body_amqp_value) != 0) + { + result = SEND_ONE_MESSAGE_ERROR; + } + else + { + body_amqp_value = amqpvalue_create_amqp_value(message_body_amqp_value); + if ((body_amqp_value == NULL) || + (amqpvalue_get_encoded_size(body_amqp_value, &encoded_size) != 0)) + { + result = SEND_ONE_MESSAGE_ERROR; + } + else + { + total_encoded_size += encoded_size; + } + } + + break; + } + + case MESSAGE_BODY_TYPE_DATA: + { + BINARY_DATA binary_data; + size_t i; + + if (message_get_body_amqp_data_count(message, &body_data_count) != 0) + { + result = SEND_ONE_MESSAGE_ERROR; + } + else + { + for (i = 0; i < body_data_count; i++) + { + if (message_get_body_amqp_data(message, i, &binary_data) != 0) + { + result = SEND_ONE_MESSAGE_ERROR; + } + else + { + amqp_binary binary_value = { binary_data.bytes, binary_data.length }; + AMQP_VALUE body_amqp_data = amqpvalue_create_data(binary_value); + if (body_amqp_data == NULL) + { + result = SEND_ONE_MESSAGE_ERROR; + } + else + { + if (amqpvalue_get_encoded_size(body_amqp_data, &encoded_size) != 0) + { + result = SEND_ONE_MESSAGE_ERROR; + } + else + { + total_encoded_size += encoded_size; + } + + amqpvalue_destroy(body_amqp_data); + } + } + } + } + break; + } + } + + if (result == 0) + { + void* data_bytes = amqpalloc_malloc(total_encoded_size); + PAYLOAD payload = { data_bytes, 0 }; + result = SEND_ONE_MESSAGE_OK; + + if (header != NULL) + { + if (amqpvalue_encode(header_amqp_value, encode_bytes, &payload) != 0) + { + result = SEND_ONE_MESSAGE_ERROR; + } + + log_message_chunk(message_sender_instance, "Header:", header_amqp_value); + } + + if ((result == SEND_ONE_MESSAGE_OK) && (properties != NULL)) + { + if (amqpvalue_encode(properties_amqp_value, encode_bytes, &payload) != 0) + { + result = SEND_ONE_MESSAGE_ERROR; + } + + log_message_chunk(message_sender_instance, "Properties:", properties_amqp_value); + } + + if ((result == SEND_ONE_MESSAGE_OK) && (application_properties != NULL)) + { + if (amqpvalue_encode(application_properties_value, encode_bytes, &payload) != 0) + { + result = SEND_ONE_MESSAGE_ERROR; + } + + log_message_chunk(message_sender_instance, "Application properties:", application_properties_value); + } + + if (result == SEND_ONE_MESSAGE_OK) + { + switch (message_body_type) + { + case MESSAGE_BODY_TYPE_VALUE: + { + if (amqpvalue_encode(body_amqp_value, encode_bytes, &payload) != 0) + { + result = SEND_ONE_MESSAGE_ERROR; + } + + log_message_chunk(message_sender_instance, "Body - amqp value:", body_amqp_value); + break; + } + case MESSAGE_BODY_TYPE_DATA: + { + BINARY_DATA binary_data; + size_t i; + + for (i = 0; i < body_data_count; i++) + { + if (message_get_body_amqp_data(message, i, &binary_data) != 0) + { + result = SEND_ONE_MESSAGE_ERROR; + } + else + { + amqp_binary binary_value = { binary_data.bytes, binary_data.length }; + AMQP_VALUE body_amqp_data = amqpvalue_create_data(binary_value); + if (body_amqp_data == NULL) + { + result = SEND_ONE_MESSAGE_ERROR; + } + else + { + if (amqpvalue_encode(body_amqp_data, encode_bytes, &payload) != 0) + { + result = SEND_ONE_MESSAGE_ERROR; + break; + } + + amqpvalue_destroy(body_amqp_data); + } + } + } + break; + } + } + } + + if (result == SEND_ONE_MESSAGE_OK) + { + message_with_callback->message_send_state = MESSAGE_SEND_STATE_PENDING; + switch (link_transfer(message_sender_instance->link, message_format, &payload, 1, on_delivery_settled, message_with_callback)) + { + default: + case LINK_TRANSFER_ERROR: + if (message_with_callback->on_message_send_complete != NULL) + { + message_with_callback->on_message_send_complete(message_with_callback->context, MESSAGE_SEND_ERROR); + } + + result = SEND_ONE_MESSAGE_ERROR; + break; + + case LINK_TRANSFER_BUSY: + message_with_callback->message_send_state = MESSAGE_SEND_STATE_NOT_SENT; + result = SEND_ONE_MESSAGE_BUSY; + break; + + case LINK_TRANSFER_OK: + result = SEND_ONE_MESSAGE_OK; + break; + } + } + + amqpalloc_free(data_bytes); + + if (body_amqp_value != NULL) + { + amqpvalue_destroy(body_amqp_value); + } + + amqpvalue_destroy(application_properties); + amqpvalue_destroy(application_properties_value); + amqpvalue_destroy(properties_amqp_value); + properties_destroy(properties); + } + } + + return result; +} + +static void send_all_pending_messages(MESSAGE_SENDER_INSTANCE* message_sender_instance) +{ + size_t i; + + for (i = 0; i < message_sender_instance->message_count; i++) + { + if (message_sender_instance->messages[i]->message_send_state == MESSAGE_SEND_STATE_NOT_SENT) + { + switch (send_one_message(message_sender_instance, message_sender_instance->messages[i], message_sender_instance->messages[i]->message)) + { + default: + case SEND_ONE_MESSAGE_ERROR: + { + ON_MESSAGE_SEND_COMPLETE on_message_send_complete = message_sender_instance->messages[i]->on_message_send_complete; + void* context = message_sender_instance->messages[i]->context; + remove_pending_message_by_index(message_sender_instance, i); + + on_message_send_complete(context, MESSAGE_SEND_ERROR); + i = message_sender_instance->message_count; + break; + } + case SEND_ONE_MESSAGE_BUSY: + i = message_sender_instance->message_count + 1; + break; + + case SEND_ONE_MESSAGE_OK: + break; + } + + i--; + } + } +} + +static void set_message_sender_state(MESSAGE_SENDER_INSTANCE* message_sender_instance, MESSAGE_SENDER_STATE new_state) +{ + MESSAGE_SENDER_STATE previous_state = message_sender_instance->message_sender_state; + message_sender_instance->message_sender_state = new_state; + if (message_sender_instance->on_message_sender_state_changed != NULL) + { + message_sender_instance->on_message_sender_state_changed(message_sender_instance->on_message_sender_state_changed_context, new_state, previous_state); + } +} + +static void on_link_state_changed(void* context, LINK_STATE new_link_state, LINK_STATE previous_link_state) +{ + MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)context; + + switch (new_link_state) + { + case LINK_STATE_ATTACHED: + if (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_OPENING) + { + set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_OPEN); + } + break; + case LINK_STATE_DETACHED: + if ((message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_OPEN) || + (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_CLOSING)) + { + /* User initiated transition, we should be good */ + set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_IDLE); + } + else if (message_sender_instance->message_sender_state != MESSAGE_SENDER_STATE_IDLE) + { + /* Any other transition must be an error */ + set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_ERROR); + } + break; + } +} + +static void on_link_flow_on(void* context) +{ + MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)context; + send_all_pending_messages(message_sender_instance); +} + +MESSAGE_SENDER_HANDLE messagesender_create(LINK_HANDLE link, ON_MESSAGE_SENDER_STATE_CHANGED on_message_sender_state_changed, void* context, LOGGER_LOG logger_log) +{ + MESSAGE_SENDER_INSTANCE* result = amqpalloc_malloc(sizeof(MESSAGE_SENDER_INSTANCE)); + if (result != NULL) + { + result->messages = NULL; + result->message_count = 0; + result->link = link; + result->on_message_sender_state_changed = on_message_sender_state_changed; + result->on_message_sender_state_changed_context = context; + result->message_sender_state = MESSAGE_SENDER_STATE_IDLE; + result->logger_log = logger_log; + } + + return result; +} + +void messagesender_destroy(MESSAGE_SENDER_HANDLE message_sender) +{ + if (message_sender != NULL) + { + MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_sender; + size_t i; + + messagesender_close(message_sender_instance); + + for (i = 0; i < message_sender_instance->message_count; i++) + { + if (message_sender_instance->messages[i]->on_message_send_complete != NULL) + { + message_sender_instance->messages[i]->on_message_send_complete(message_sender_instance->messages[i]->context, MESSAGE_SEND_ERROR); + } + + message_destroy(message_sender_instance->messages[i]->message); + amqpalloc_free(message_sender_instance->messages[i]); + } + + if (message_sender_instance->messages != NULL) + { + amqpalloc_free(message_sender_instance->messages); + } + + amqpalloc_free(message_sender); + } +} + +int messagesender_open(MESSAGE_SENDER_HANDLE message_sender) +{ + int result; + + if (message_sender == NULL) + { + result = __LINE__; + } + else + { + MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_sender; + + if (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_IDLE) + { + set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_OPENING); + if (link_attach(message_sender_instance->link, NULL, on_link_state_changed, on_link_flow_on, message_sender_instance) != 0) + { + result = __LINE__; + set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_ERROR); + } + else + { + result = 0; + } + } + else + { + result = 0; + } + } + + return result; +} + +int messagesender_close(MESSAGE_SENDER_HANDLE message_sender) +{ + int result; + + if (message_sender == NULL) + { + result = __LINE__; + } + else + { + MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_sender; + + if ((message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_OPENING) || + (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_OPEN)) + { + set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_CLOSING); + if (link_detach(message_sender_instance->link) != 0) + { + result = __LINE__; + set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_ERROR); + } + else + { + set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_IDLE); + result = 0; + } + } + else + { + result = 0; + } + } + + return result; +} + +int messagesender_send(MESSAGE_SENDER_HANDLE message_sender, MESSAGE_HANDLE message, ON_MESSAGE_SEND_COMPLETE on_message_send_complete, void* callback_context) +{ + int result; + + if ((message_sender == NULL) || + (message == NULL)) + { + result = __LINE__; + } + else + { + MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_sender; + if (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_ERROR) + { + result = __LINE__; + } + else + { + MESSAGE_WITH_CALLBACK* message_with_callback = (MESSAGE_WITH_CALLBACK*)amqpalloc_malloc(sizeof(MESSAGE_WITH_CALLBACK)); + if (message_with_callback == NULL) + { + result = __LINE__; + } + else + { + MESSAGE_WITH_CALLBACK** new_messages = (MESSAGE_WITH_CALLBACK**)amqpalloc_realloc(message_sender_instance->messages, sizeof(MESSAGE_WITH_CALLBACK*) * (message_sender_instance->message_count + 1)); + if (new_messages == NULL) + { + amqpalloc_free(message_with_callback); + result = __LINE__; + } + else + { + result = 0; + + message_sender_instance->messages = new_messages; + if (message_sender_instance->message_sender_state != MESSAGE_SENDER_STATE_OPEN) + { + message_with_callback->message = message_clone(message); + if (message_with_callback->message == NULL) + { + amqpalloc_free(message_with_callback); + result = __LINE__; + } + + message_with_callback->message_send_state = MESSAGE_SEND_STATE_NOT_SENT; + } + else + { + message_with_callback->message = NULL; + message_with_callback->message_send_state = MESSAGE_SEND_STATE_PENDING; + } + + if (result == 0) + { + message_with_callback->on_message_send_complete = on_message_send_complete; + message_with_callback->context = callback_context; + message_with_callback->message_sender = message_sender_instance; + + message_sender_instance->messages[message_sender_instance->message_count] = message_with_callback; + message_sender_instance->message_count++; + + if (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_OPEN) + { + switch (send_one_message(message_sender_instance, message_with_callback, message)) + { + default: + case SEND_ONE_MESSAGE_ERROR: + remove_pending_message_by_index(message_sender_instance, message_sender_instance->message_count - 1); + result = __LINE__; + break; + + case SEND_ONE_MESSAGE_BUSY: + message_with_callback->message = message_clone(message); + if (message_with_callback->message == NULL) + { + amqpalloc_free(message_with_callback); + result = __LINE__; + } + else + { + message_with_callback->message_send_state = MESSAGE_SEND_STATE_NOT_SENT; + result = 0; + } + + break; + + case SEND_ONE_MESSAGE_OK: + result = 0; + break; + } + } + } + } + } + } + } + + return result; +}