// Copyright (c) Microsoft. All rights reserved. // Licensed under the MIT license. See LICENSE file in the project root for full license information. #include #include "azure_macro_utils/macro_utils.h" #include "azure_c_shared_utility/gballoc.h" #include "azure_c_shared_utility/xlogging.h" #include "azure_uamqp_c/link.h" #include "azure_uamqp_c/amqp_definitions.h" #include "azure_uamqp_c/message.h" #include "azure_uamqp_c/message_receiver.h" #include "azure_uamqp_c/amqpvalue.h" typedef struct MESSAGE_RECEIVER_INSTANCE_TAG { LINK_HANDLE link; ON_MESSAGE_RECEIVED on_message_received; ON_MESSAGE_RECEIVER_STATE_CHANGED on_message_receiver_state_changed; MESSAGE_RECEIVER_STATE message_receiver_state; const void* on_message_receiver_state_changed_context; const void* callback_context; MESSAGE_HANDLE decoded_message; bool decode_error; } MESSAGE_RECEIVER_INSTANCE; static void set_message_receiver_state(MESSAGE_RECEIVER_INSTANCE* message_receiver, MESSAGE_RECEIVER_STATE new_state) { MESSAGE_RECEIVER_STATE previous_state = message_receiver->message_receiver_state; message_receiver->message_receiver_state = new_state; if (message_receiver->on_message_receiver_state_changed != NULL) { message_receiver->on_message_receiver_state_changed(message_receiver->on_message_receiver_state_changed_context, new_state, previous_state); } } static void decode_message_value_callback(void* context, AMQP_VALUE decoded_value) { MESSAGE_RECEIVER_INSTANCE* message_receiver = (MESSAGE_RECEIVER_INSTANCE*)context; MESSAGE_HANDLE decoded_message = message_receiver->decoded_message; AMQP_VALUE descriptor = amqpvalue_get_inplace_descriptor(decoded_value); if (is_application_properties_type_by_descriptor(descriptor)) { if (message_set_application_properties(decoded_message, decoded_value) != 0) { LogError("Error setting application properties on received message"); message_receiver->decode_error = true; } } else if (is_properties_type_by_descriptor(descriptor)) { PROPERTIES_HANDLE properties; if (amqpvalue_get_properties(decoded_value, &properties) != 0) { LogError("Error getting message properties"); message_receiver->decode_error = true; } else { if (message_set_properties(decoded_message, properties) != 0) { LogError("Error setting message properties on received message"); message_receiver->decode_error = true; } properties_destroy(properties); } } else if (is_delivery_annotations_type_by_descriptor(descriptor)) { annotations delivery_annotations = amqpvalue_get_inplace_described_value(decoded_value); if (delivery_annotations == NULL) { LogError("Error getting delivery annotations"); message_receiver->decode_error = true; } else { if (message_set_delivery_annotations(decoded_message, delivery_annotations) != 0) { LogError("Error setting delivery annotations on received message"); message_receiver->decode_error = true; } } } else if (is_message_annotations_type_by_descriptor(descriptor)) { annotations message_annotations = amqpvalue_get_inplace_described_value(decoded_value); if (message_annotations == NULL) { LogError("Error getting message annotations"); message_receiver->decode_error = true; } else { if (message_set_message_annotations(decoded_message, message_annotations) != 0) { LogError("Error setting message annotations on received message"); message_receiver->decode_error = true; } } } else if (is_header_type_by_descriptor(descriptor)) { HEADER_HANDLE header; if (amqpvalue_get_header(decoded_value, &header) != 0) { LogError("Error getting message header"); message_receiver->decode_error = true; } else { if (message_set_header(decoded_message, header) != 0) { LogError("Error setting message header on received message"); message_receiver->decode_error = true; } header_destroy(header); } } else if (is_footer_type_by_descriptor(descriptor)) { annotations footer = amqpvalue_get_inplace_described_value(decoded_value); if (footer == NULL) { LogError("Error getting message footer"); message_receiver->decode_error = true; } else { if (message_set_footer(decoded_message, footer) != 0) { LogError("Error setting message footer on received message"); message_receiver->decode_error = true; } } } else if (is_amqp_value_type_by_descriptor(descriptor)) { MESSAGE_BODY_TYPE body_type; if (message_get_body_type(decoded_message, &body_type) != 0) { LogError("Error getting message body type"); message_receiver->decode_error = true; } else { if (body_type != MESSAGE_BODY_TYPE_NONE) { LogError("Body already set on received message"); message_receiver->decode_error = true; } else { AMQP_VALUE body_amqp_value = amqpvalue_get_inplace_described_value(decoded_value); if (body_amqp_value == NULL) { LogError("Error getting body AMQP value"); message_receiver->decode_error = true; } else { if (message_set_body_amqp_value(decoded_message, body_amqp_value) != 0) { LogError("Error setting body AMQP value on received message"); message_receiver->decode_error = true; } } } } } else if (is_data_type_by_descriptor(descriptor)) { MESSAGE_BODY_TYPE body_type; if (message_get_body_type(decoded_message, &body_type) != 0) { LogError("Error getting message body type"); message_receiver->decode_error = true; } else { if ((body_type != MESSAGE_BODY_TYPE_NONE) && (body_type != MESSAGE_BODY_TYPE_DATA)) { LogError("Message body type already set to something different than AMQP DATA"); message_receiver->decode_error = true; } else { AMQP_VALUE body_data_value = amqpvalue_get_inplace_described_value(decoded_value); if (body_data_value == NULL) { LogError("Error getting body DATA value"); message_receiver->decode_error = true; } else { data data_value; if (amqpvalue_get_data(body_data_value, &data_value) != 0) { LogError("Error getting body DATA AMQP value"); message_receiver->decode_error = true; } else { BINARY_DATA binary_data; binary_data.bytes = (const unsigned char*)data_value.bytes; binary_data.length = data_value.length; if (message_add_body_amqp_data(decoded_message, binary_data) != 0) { LogError("Error adding body DATA to received message"); message_receiver->decode_error = true; } } } } } } else { LogError("Failed decoding descriptor"); message_receiver->decode_error = true; } } static AMQP_VALUE on_transfer_received(void* context, TRANSFER_HANDLE transfer, uint32_t payload_size, const unsigned char* payload_bytes) { AMQP_VALUE result = NULL; MESSAGE_RECEIVER_INSTANCE* message_receiver = (MESSAGE_RECEIVER_INSTANCE*)context; (void)transfer; if (message_receiver->on_message_received != NULL) { MESSAGE_HANDLE message = message_create(); if (message == NULL) { LogError("Cannot create message"); set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_ERROR); } else { AMQPVALUE_DECODER_HANDLE amqpvalue_decoder = amqpvalue_decoder_create(decode_message_value_callback, message_receiver); if (amqpvalue_decoder == NULL) { LogError("Cannot create AMQP value decoder"); set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_ERROR); } else { message_receiver->decoded_message = message; message_receiver->decode_error = false; if (amqpvalue_decode_bytes(amqpvalue_decoder, payload_bytes, payload_size) != 0) { LogError("Cannot decode bytes"); set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_ERROR); } else { if (message_receiver->decode_error) { LogError("Error decoding message"); set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_ERROR); } else { result = message_receiver->on_message_received(message_receiver->callback_context, message); } } amqpvalue_decoder_destroy(amqpvalue_decoder); } message_destroy(message); } } return result; } static void on_link_state_changed(void* context, LINK_STATE new_link_state, LINK_STATE previous_link_state) { MESSAGE_RECEIVER_INSTANCE* message_receiver = (MESSAGE_RECEIVER_INSTANCE*)context; (void)previous_link_state; switch (new_link_state) { default: break; case LINK_STATE_ATTACHED: if (message_receiver->message_receiver_state == MESSAGE_RECEIVER_STATE_OPENING) { set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_OPEN); } break; case LINK_STATE_DETACHED: if ((message_receiver->message_receiver_state == MESSAGE_RECEIVER_STATE_OPEN) || (message_receiver->message_receiver_state == MESSAGE_RECEIVER_STATE_CLOSING)) { /* User initiated transition, we should be good */ set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_IDLE); } else if (message_receiver->message_receiver_state != MESSAGE_RECEIVER_STATE_IDLE) { /* Any other transition must be an error */ set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_ERROR); } break; case LINK_STATE_ERROR: if (message_receiver->message_receiver_state != MESSAGE_RECEIVER_STATE_ERROR) { set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_ERROR); } break; } } MESSAGE_RECEIVER_HANDLE messagereceiver_create(LINK_HANDLE link, ON_MESSAGE_RECEIVER_STATE_CHANGED on_message_receiver_state_changed, void* context) { MESSAGE_RECEIVER_INSTANCE* message_receiver = (MESSAGE_RECEIVER_INSTANCE*)calloc(1, sizeof(MESSAGE_RECEIVER_INSTANCE)); if (message_receiver == NULL) { LogError("Error creating message receiver"); } else { message_receiver->link = link; message_receiver->on_message_receiver_state_changed = on_message_receiver_state_changed; message_receiver->on_message_receiver_state_changed_context = context; message_receiver->message_receiver_state = MESSAGE_RECEIVER_STATE_IDLE; } return message_receiver; } void messagereceiver_destroy(MESSAGE_RECEIVER_HANDLE message_receiver) { if (message_receiver == NULL) { LogError("NULL message_receiver"); } else { (void)messagereceiver_close(message_receiver); free(message_receiver); } } int messagereceiver_open(MESSAGE_RECEIVER_HANDLE message_receiver, ON_MESSAGE_RECEIVED on_message_received, void* callback_context) { int result; if (message_receiver == NULL) { LogError("NULL message_receiver"); result = MU_FAILURE; } else { if (message_receiver->message_receiver_state == MESSAGE_RECEIVER_STATE_IDLE) { set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_OPENING); if (link_attach(message_receiver->link, on_transfer_received, on_link_state_changed, NULL, message_receiver) != 0) { LogError("Link attach failed"); result = MU_FAILURE; set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_ERROR); } else { message_receiver->on_message_received = on_message_received; message_receiver->callback_context = callback_context; result = 0; } } else { result = 0; } } return result; } int messagereceiver_close(MESSAGE_RECEIVER_HANDLE message_receiver) { int result; if (message_receiver == NULL) { LogError("NULL message_receiver"); result = MU_FAILURE; } else { if ((message_receiver->message_receiver_state == MESSAGE_RECEIVER_STATE_OPENING) || (message_receiver->message_receiver_state == MESSAGE_RECEIVER_STATE_OPEN)) { set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_CLOSING); if (link_detach(message_receiver->link, true, NULL, NULL, NULL) != 0) { LogError("link detach failed"); result = MU_FAILURE; set_message_receiver_state(message_receiver, MESSAGE_RECEIVER_STATE_ERROR); } else { result = 0; } } else { result = 0; } } return result; } int messagereceiver_get_link_name(MESSAGE_RECEIVER_HANDLE message_receiver, const char** link_name) { int result; if (message_receiver == NULL) { LogError("NULL message_receiver"); result = MU_FAILURE; } else { if (link_get_name(message_receiver->link, link_name) != 0) { LogError("Getting link name failed"); result = MU_FAILURE; } else { result = 0; } } return result; } int messagereceiver_get_received_message_id(MESSAGE_RECEIVER_HANDLE message_receiver, delivery_number* message_id) { int result; if (message_receiver == NULL) { LogError("NULL message_receiver"); result = MU_FAILURE; } else { if (link_get_received_message_id(message_receiver->link, message_id) != 0) { LogError("Failed getting received message Id"); result = MU_FAILURE; } else { result = 0; } } return result; } int messagereceiver_send_message_disposition(MESSAGE_RECEIVER_HANDLE message_receiver, const char* link_name, delivery_number message_number, AMQP_VALUE delivery_state) { int result; if (message_receiver == NULL) { LogError("NULL message_receiver"); result = MU_FAILURE; } else { if (message_receiver->message_receiver_state != MESSAGE_RECEIVER_STATE_OPEN) { LogError("Message received not open"); result = MU_FAILURE; } else { const char* my_name; if (link_get_name(message_receiver->link, &my_name) != 0) { LogError("Failed getting link name"); result = MU_FAILURE; } else { if (strcmp(link_name, my_name) != 0) { LogError("Link name does not match"); result = MU_FAILURE; } else { if (link_send_disposition(message_receiver->link, message_number, delivery_state) != 0) { LogError("Seding disposition failed"); result = MU_FAILURE; } else { result = 0; } } } } } return result; } void messagereceiver_set_trace(MESSAGE_RECEIVER_HANDLE message_receiver, bool trace_on) { if (message_receiver == NULL) { LogError("NULL message_receiver"); } else { /* No tracing is yet implemented for message receiver */ (void)trace_on; } }