Azure IoT
A small memory footprint AMQP implimentation
Dependents: iothub_client_sample_amqp remote_monitoring simplesample_amqp
Diff: amqp_management.c
- Revision:
- 23:1111ee8bcba4
- Parent:
- 22:524bded3f7a8
- Child:
- 25:1101516ee67d
--- a/amqp_management.c Thu Apr 06 14:11:27 2017 -0700
+++ b/amqp_management.c Fri Apr 21 14:50:32 2017 -0700
@@ -7,6 +7,7 @@
#include <string.h>
#include "azure_c_shared_utility/optimize_size.h"
#include "azure_c_shared_utility/gballoc.h"
+#include "azure_c_shared_utility/singlylinkedlist.h"
#include "azure_uamqp_c/amqp_management.h"
#include "azure_uamqp_c/link.h"
#include "azure_uamqp_c/message_sender.h"
@@ -14,19 +15,16 @@
#include "azure_uamqp_c/messaging.h"
#include "azure_uamqp_c/amqpvalue_to_string.h"
-typedef enum OPERATION_STATE_TAG
-{
- OPERATION_STATE_NOT_SENT,
- OPERATION_STATE_AWAIT_REPLY
-} OPERATION_STATE;
+static const char sender_suffix[] = "-sender";
+static const char receiver_suffix[] = "-receiver";
+
+#define COUNT_CHARS(str) (sizeof(str) / sizeof((str)[0]) - 1)
typedef struct OPERATION_MESSAGE_INSTANCE_TAG
{
- MESSAGE_HANDLE message;
- OPERATION_STATE operation_state;
ON_AMQP_MANAGEMENT_EXECUTE_OPERATION_COMPLETE on_execute_operation_complete;
void* callback_context;
- unsigned long message_id;
+ uint64_t message_id;
} OPERATION_MESSAGE_INSTANCE;
typedef enum AMQP_MANAGEMENT_STATE_TAG
@@ -39,14 +37,12 @@
typedef struct AMQP_MANAGEMENT_INSTANCE_TAG
{
- SESSION_HANDLE session;
LINK_HANDLE sender_link;
LINK_HANDLE receiver_link;
MESSAGE_SENDER_HANDLE message_sender;
MESSAGE_RECEIVER_HANDLE message_receiver;
- OPERATION_MESSAGE_INSTANCE** operation_messages;
- size_t operation_message_count;
- unsigned long next_message_id;
+ SINGLYLINKEDLIST_HANDLE pending_operations;
+ uint64_t next_message_id;
ON_AMQP_MANAGEMENT_OPEN_COMPLETE on_amqp_management_open_complete;
void* on_amqp_management_open_complete_context;
ON_AMQP_MANAGEMENT_ERROR on_amqp_management_error;
@@ -56,214 +52,274 @@
int receiver_connected : 1;
} AMQP_MANAGEMENT_INSTANCE;
-static void remove_operation_message_by_index(AMQP_MANAGEMENT_INSTANCE* amqp_management_instance, size_t index)
+static AMQP_VALUE on_message_received(const void* context, MESSAGE_HANDLE message)
{
- message_destroy(amqp_management_instance->operation_messages[index]->message);
- free(amqp_management_instance->operation_messages[index]);
+ AMQP_VALUE result;
- if (amqp_management_instance->operation_message_count - index > 1)
+ if (context == NULL)
{
- memmove(&amqp_management_instance->operation_messages[index], &amqp_management_instance->operation_messages[index + 1], sizeof(OPERATION_MESSAGE_INSTANCE*) * (amqp_management_instance->operation_message_count - index - 1));
- }
-
- if (amqp_management_instance->operation_message_count == 1)
- {
- free(amqp_management_instance->operation_messages);
- amqp_management_instance->operation_messages = NULL;
+ /* Codes_SRS_AMQP_MANAGEMENT_01_108: [ When `on_message_received` is called with a NULL context, it shall do nothing. ]*/
+ LogError("NULL context in on_message_received");
+ result = NULL;
}
else
{
- OPERATION_MESSAGE_INSTANCE** new_operation_messages = (OPERATION_MESSAGE_INSTANCE**)realloc(amqp_management_instance->operation_messages, sizeof(OPERATION_MESSAGE_INSTANCE*) * (amqp_management_instance->operation_message_count - 1));
- if (new_operation_messages != NULL)
- {
- amqp_management_instance->operation_messages = new_operation_messages;
- }
- }
-
- amqp_management_instance->operation_message_count--;
-}
+ AMQP_MANAGEMENT_HANDLE amqp_management = (AMQP_MANAGEMENT_HANDLE)context;
+ AMQP_VALUE application_properties;
-static AMQP_VALUE on_message_received(const void* context, MESSAGE_HANDLE message)
-{
- AMQP_MANAGEMENT_INSTANCE* amqp_management_instance = (AMQP_MANAGEMENT_INSTANCE*)context;
-
- AMQP_VALUE application_properties;
- if (message_get_application_properties(message, &application_properties) != 0)
- {
- /* error */
- }
- else
- {
- PROPERTIES_HANDLE response_properties;
-
- if (message_get_properties(message, &response_properties) != 0)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_109: [ `on_message_received` shall obtain the application properties from the message by calling `message_get_application_properties`. ]*/
+ if (message_get_application_properties(message, &application_properties) != 0)
{
- /* error */
+ /* Codes_SRS_AMQP_MANAGEMENT_01_113: [ If obtaining the application properties or message properties fails, an error shall be indicated by calling `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ LogError("Could not retrieve application properties");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_136: [ When `on_message_received` fails due to errors in parsing the response message `on_message_received` shall call `messaging_delivery_rejected` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_rejected("amqp:internal-error", "Could not get application properties on AMQP management response.");
}
else
{
- AMQP_VALUE key;
- AMQP_VALUE value;
- AMQP_VALUE desc_key;
- AMQP_VALUE desc_value;
- AMQP_VALUE map;
- AMQP_VALUE correlation_id_value;
+ PROPERTIES_HANDLE response_properties;
- if (properties_get_correlation_id(response_properties, &correlation_id_value) != 0)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_110: [ `on_message_received` shall obtain the message properties from the message by calling `message_get_properties`. ]*/
+ if (message_get_properties(message, &response_properties) != 0)
{
- /* error */
+ /* Codes_SRS_AMQP_MANAGEMENT_01_113: [ If obtaining the application properties or message properties fails, an error shall be indicated by calling `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ LogError("Could not retrieve message properties");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_136: [ When `on_message_received` fails due to errors in parsing the response message `on_message_received` shall call `messaging_delivery_rejected` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_rejected("amqp:internal-error", "Could not get message properties on AMQP management response.");
}
else
{
- map = amqpvalue_get_inplace_described_value(application_properties);
- if (map == NULL)
+ AMQP_VALUE key;
+ AMQP_VALUE value;
+ AMQP_VALUE desc_key;
+ AMQP_VALUE desc_value;
+ AMQP_VALUE map;
+ AMQP_VALUE correlation_id_value;
+ uint64_t correlation_id;
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_111: [ `on_message_received` shall obtain the correlation Id from the message properties by using `properties_get_correlation_id`. ]*/
+ if (properties_get_correlation_id(response_properties, &correlation_id_value) != 0)
{
- /* error */
+ /* Codes_SRS_AMQP_MANAGEMENT_01_114: [ If obtaining the correlation Id fails, an error shall be indicated by calling `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ] */
+ LogError("Could not retrieve correlation Id");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_136: [ When `on_message_received` fails due to errors in parsing the response message `on_message_received` shall call `messaging_delivery_rejected` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_rejected("amqp:internal-error", "Could not get correlation Id from AMQP management response.");
}
else
{
- key = amqpvalue_create_string("status-code");
- if (key == NULL)
+ if (amqpvalue_get_ulong(correlation_id_value, &correlation_id) != 0)
{
- /* error */
+ /* Codes_SRS_AMQP_MANAGEMENT_01_132: [ If any functions manipulating AMQP values, application properties, etc., fail, an error shall be indicated to the consumer by calling the `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ LogError("Could not retrieve correlation Id ulong value");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_136: [ When `on_message_received` fails due to errors in parsing the response message `on_message_received` shall call `messaging_delivery_rejected` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_rejected("amqp:internal-error", "Could not get correlation Id from AMQP management response.");
}
else
{
- value = amqpvalue_get_map_value(map, key);
- if (value == NULL)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_119: [ `on_message_received` shall obtain the application properties map by calling `amqpvalue_get_inplace_described_value`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_070: [ Response messages have the following application-properties: ]*/
+ map = amqpvalue_get_inplace_described_value(application_properties);
+ if (map == NULL)
{
- /* error */
+ /* Codes_SRS_AMQP_MANAGEMENT_01_132: [ If any functions manipulating AMQP values, application properties, etc., fail, an error shall be indicated to the consumer by calling the `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ LogError("Could not retrieve application property map");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_136: [ When `on_message_received` fails due to errors in parsing the response message `on_message_received` shall call `messaging_delivery_rejected` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_rejected("amqp:internal-error", "Could not get application property map from the application properties in the AMQP management response.");
}
else
{
- int32_t status_code;
- if (amqpvalue_get_int(value, &status_code) != 0)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_120: [ An AMQP value used to lookup the status code shall be created by calling `amqpvalue_create_string` with `status-code` as argument. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_071: [ statusCode integer Yes HTTP response code [RFC2616] ]*/
+ key = amqpvalue_create_string("status-code");
+ if (key == NULL)
{
- /* error */
+ /* Codes_SRS_AMQP_MANAGEMENT_01_132: [ If any functions manipulating AMQP values, application properties, etc., fail, an error shall be indicated to the consumer by calling the `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ LogError("Could not create status-code amqp value");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_135: [ When an error occurs in creating AMQP values (for status code, etc.) `on_message_received` shall call `messaging_delivery_released` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_released();
}
else
{
- desc_key = amqpvalue_create_string("status-description");
- if (desc_key == NULL)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_121: [ The status code shall be looked up in the application properties by using `amqpvalue_get_map_value`. ]*/
+ value = amqpvalue_get_map_value(map, key);
+ if (value == NULL)
{
- /* error */
+ /* Codes_SRS_AMQP_MANAGEMENT_01_122: [ If status code is not found an error shall be indicated to the consumer by calling the `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ LogError("Could not retrieve status code from application properties");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_136: [ When `on_message_received` fails due to errors in parsing the response message `on_message_received` shall call `messaging_delivery_rejected` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_rejected("amqp:internal-error", "Could not retrieve status code from the application properties in the AMQP management response.");
}
else
{
- const char* status_description = NULL;
-
- desc_value = amqpvalue_get_map_value(map, desc_key);
- if (desc_value != NULL)
+ int32_t status_code;
+ /* Codes_SRS_AMQP_MANAGEMENT_01_133: [ The status code value shall be extracted from the value found in the map by using `amqpvalue_get_int`. ]*/
+ if (amqpvalue_get_int(value, &status_code) != 0)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_132: [ If any functions manipulating AMQP values, application properties, etc., fail, an error shall be indicated to the consumer by calling the `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ LogError("Could not retrieve status code int value");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_136: [ When `on_message_received` fails due to errors in parsing the response message `on_message_received` shall call `messaging_delivery_rejected` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_rejected("amqp:internal-error", "Could not retrieve status code value from the application properties in the AMQP management response.");
+ }
+ else
{
- amqpvalue_get_string(desc_value, &status_description);
- }
+ /* Codes_SRS_AMQP_MANAGEMENT_01_123: [ An AMQP value used to lookup the status description shall be created by calling `amqpvalue_create_string` with `status-description` as argument. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_072: [ statusDescription string No Description of the status. ]*/
+ desc_key = amqpvalue_create_string("status-description");
+ if (desc_key == NULL)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_132: [ If any functions manipulating AMQP values, application properties, etc., fail, an error shall be indicated to the consumer by calling the `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ LogError("Could not create status-description amqp value");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_135: [ When an error occurs in creating AMQP values (for status code, etc.) `on_message_received` shall call `messaging_delivery_released` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_released();
+ }
+ else
+ {
+ const char* status_description = NULL;
+ LIST_ITEM_HANDLE list_item_handle;
+ bool found = false;
+ bool is_error = false;
- size_t i = 0;
- while (i < amqp_management_instance->operation_message_count)
- {
- if (amqp_management_instance->operation_messages[i]->operation_state == OPERATION_STATE_AWAIT_REPLY)
- {
- AMQP_VALUE expected_message_id = amqpvalue_create_ulong(amqp_management_instance->operation_messages[i]->message_id);
- AMQP_MANAGEMENT_EXECUTE_OPERATION_RESULT execute_operation_result;
-
- if (expected_message_id == NULL)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_124: [ The status description shall be looked up in the application properties by using `amqpvalue_get_map_value`. ]*/
+ desc_value = amqpvalue_get_map_value(map, desc_key);
+ if (desc_value != NULL)
{
- break;
+ /* Codes_SRS_AMQP_MANAGEMENT_01_134: [ The status description value shall be extracted from the value found in the map by using `amqpvalue_get_string`. ]*/
+ if (amqpvalue_get_string(desc_value, &status_description) != 0)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_125: [ If status description is not found, NULL shall be passed to the user callback as `status_description` argument. ]*/
+ status_description = NULL;
+ }
}
else
{
- if (amqpvalue_are_equal(correlation_id_value, expected_message_id))
+ /* Codes_SRS_AMQP_MANAGEMENT_01_125: [ If status description is not found, NULL shall be passed to the user callback as `status_description` argument. ]*/
+ status_description = NULL;
+ }
+
+ list_item_handle = singlylinkedlist_get_head_item(amqp_management->pending_operations);
+ while (list_item_handle != NULL)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_116: [ Each pending operation item value shall be obtained by calling `singlylinkedlist_item_get_value`. ]*/
+ OPERATION_MESSAGE_INSTANCE* operation_message = (OPERATION_MESSAGE_INSTANCE*)singlylinkedlist_item_get_value(list_item_handle);
+ if (operation_message == NULL)
{
- /* 202 is not mentioned in the draft in any way, this is a workaround for an EH bug for now */
- if ((status_code != 200) && (status_code != 202))
- {
- execute_operation_result = AMQP_MANAGEMENT_EXECUTE_OPERATION_FAILED_BAD_STATUS;
- }
- else
- {
- execute_operation_result = AMQP_MANAGEMENT_EXECUTE_OPERATION_OK;
- }
-
- amqp_management_instance->operation_messages[i]->on_execute_operation_complete(amqp_management_instance->operation_messages[i]->callback_context, execute_operation_result, status_code, status_description);
-
- remove_operation_message_by_index(amqp_management_instance, i);
-
- amqpvalue_destroy(expected_message_id);
-
+ /* Codes_SRS_AMQP_MANAGEMENT_01_117: [ If iterating through the pending operations list fails, an error shall be indicated by calling `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ LogError("Could not create status-description amqp value");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_135: [ When an error occurs in creating AMQP values (for status code, etc.) `on_message_received` shall call `messaging_delivery_released` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_released();
break;
}
else
{
- i++;
+ AMQP_MANAGEMENT_EXECUTE_OPERATION_RESULT execute_operation_result;
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_112: [ `on_message_received` shall check if the correlation Id matches the stored message Id of any pending operation. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_068: [ The correlation-id of the response message MUST be the correlation-id from the request message (if present) ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_069: [ else the message-id from the request message. ]*/
+ if (correlation_id == operation_message->message_id)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_074: [ Successful operations MUST result in a statusCode in the 2xx range as defined in Section 10.2 of [RFC2616]. ]*/
+ if ((status_code < 200) || (status_code > 299))
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_128: [ If the status indicates that the operation failed, the result callback argument shall be `AMQP_MANAGEMENT_EXECUTE_OPERATION_FAILED_BAD_STATUS`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_075: [ Unsuccessful operations MUST NOT result in a statusCode in the 2xx range as defined in Section 10.2 of [RFC2616]. ]*/
+ execute_operation_result = AMQP_MANAGEMENT_EXECUTE_OPERATION_FAILED_BAD_STATUS;
+ }
+ else
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_127: [ If the operation succeeded the result callback argument shall be `AMQP_MANAGEMENT_EXECUTE_OPERATION_OK`. ]*/
+ execute_operation_result = AMQP_MANAGEMENT_EXECUTE_OPERATION_OK;
+ }
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_126: [ If a corresponding correlation Id is found in the pending operations list, the callback associated with the pending operation shall be called. ]*/
+ operation_message->on_execute_operation_complete(operation_message->callback_context, execute_operation_result, status_code, status_description);
+
+ free(operation_message);
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_129: [ After calling the callback, the pending operation shall be removed from the pending operations list by calling `singlylinkedlist_remove`. ]*/
+ if (singlylinkedlist_remove(amqp_management->pending_operations, list_item_handle) != 0)
+ {
+ LogError("Cannot remove pending operation");
+ is_error = true;
+ break;
+ }
+ else
+ {
+ found = true;
+ }
+
+ break;
+ }
}
- amqpvalue_destroy(expected_message_id);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_115: [ Iterating through the pending operations shall be done by using `singlylinkedlist_get_head_item` and `singlylinkedlist_get_next_item` until the enm of the pending operations singly linked list is reached. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_117: [ If iterating through the pending operations list fails, an error shall be indicated by calling `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ list_item_handle = singlylinkedlist_get_next_item(list_item_handle);
+ }
+
+ if (is_error)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_117: [ If iterating through the pending operations list fails, an error shall be indicated by calling `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_135: [ When an error occurs in creating AMQP values (for status code, etc.) `on_message_received` shall call `messaging_delivery_released` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_released();
}
- }
- else
- {
- i++;
+ else
+ {
+ if (!found)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_118: [ If no pending operation is found matching the correlation Id, an error shall be indicated by calling `on_amqp_management_error` and passing the `on_amqp_management_error_context` to it. ]*/
+ LogError("Could not match AMQP management response to request");
+ amqp_management->on_amqp_management_error(amqp_management->on_amqp_management_error_context);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_135: [ When an error occurs in creating AMQP values (for status code, etc.) `on_message_received` shall call `messaging_delivery_released` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_rejected("amqp:internal-error", "Could not match AMQP management response to request");
+ }
+ else
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_130: [ The `on_message_received` shall call `messaging_delivery_accepted` and return the created delivery AMQP value. ]*/
+ result = messaging_delivery_accepted();
+ }
+ }
+
+ if (desc_value != NULL)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_131: [ All temporary values like AMQP values used as keys shall be freed before exiting the callback. ]*/
+ amqpvalue_destroy(desc_value);
+ }
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_131: [ All temporary values like AMQP values used as keys shall be freed before exiting the callback. ]*/
+ amqpvalue_destroy(desc_key);
}
}
- if (desc_value != NULL)
- {
- amqpvalue_destroy(desc_value);
- }
- amqpvalue_destroy(desc_key);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_131: [ All temporary values like AMQP values used as keys shall be freed before exiting the callback. ]*/
+ amqpvalue_destroy(value);
}
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_131: [ All temporary values like AMQP values used as keys shall be freed before exiting the callback. ]*/
+ amqpvalue_destroy(key);
}
- amqpvalue_destroy(value);
}
- amqpvalue_destroy(key);
}
}
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_131: [ All temporary values like AMQP values used as keys shall be freed before exiting the callback. ]*/
+ properties_destroy(response_properties);
}
- properties_destroy(response_properties);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_131: [ All temporary values like AMQP values used as keys shall be freed before exiting the callback. ]*/
+ application_properties_destroy(application_properties);
}
-
- application_properties_destroy(application_properties);
- }
-
- return messaging_delivery_accepted();
-}
-
-static int send_operation_messages(AMQP_MANAGEMENT_INSTANCE* amqp_management_instance)
-{
- int result;
-
- if ((amqp_management_instance->sender_connected != 0) &&
- (amqp_management_instance->receiver_connected != 0))
- {
- size_t i;
-
- for (i = 0; i < amqp_management_instance->operation_message_count; i++)
- {
- if (amqp_management_instance->operation_messages[i]->operation_state == OPERATION_STATE_NOT_SENT)
- {
- if (messagesender_send(amqp_management_instance->message_sender, amqp_management_instance->operation_messages[i]->message, NULL, NULL) != 0)
- {
- /* error */
- break;
- }
-
- amqp_management_instance->operation_messages[i]->operation_state = OPERATION_STATE_AWAIT_REPLY;
- }
- }
-
- if (i < amqp_management_instance->operation_message_count)
- {
- result = __FAILURE__;
- }
- else
- {
- result = 0;
- }
- }
- else
- {
- result = 0;
}
return result;
@@ -271,136 +327,177 @@
static void on_message_sender_state_changed(void* context, MESSAGE_SENDER_STATE new_state, MESSAGE_SENDER_STATE previous_state)
{
- AMQP_MANAGEMENT_INSTANCE* amqp_management_instance = (AMQP_MANAGEMENT_INSTANCE*)context;
- (void)previous_state;
- switch (new_state)
+ if (context == NULL)
{
- default:
- break;
-
- case MESSAGE_SENDER_STATE_OPEN:
- switch (amqp_management_instance->amqp_management_state)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_137: [ When `on_message_sender_state_changed` is called with NULL `context`, it shall do nothing. ]*/
+ LogError("on_message_sender_state_changed called with NULL context");
+ }
+ else
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_138: [ When `on_message_sender_state_changed` is called and the `new_state` is different than `previous_state`, the following actions shall be taken: ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_148: [ When no state change is detected, `on_message_sender_state_changed` shall do nothing. ]*/
+ if (new_state != previous_state)
{
- default:
- case AMQP_MANAGEMENT_STATE_ERROR:
- /* do nothing */
- break;
+ AMQP_MANAGEMENT_INSTANCE* amqp_management_instance = (AMQP_MANAGEMENT_INSTANCE*)context;
+ switch (amqp_management_instance->amqp_management_state)
+ {
+ default:
+ break;
- case AMQP_MANAGEMENT_STATE_OPENING:
- amqp_management_instance->sender_connected = -1;
- if (amqp_management_instance->receiver_connected != 0)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_139: [ For the current state of AMQP management being `OPENING`: ]*/
+ case AMQP_MANAGEMENT_STATE_OPENING:
{
- (void)send_operation_messages(amqp_management_instance);
- amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_OPEN;
- amqp_management_instance->on_amqp_management_open_complete(amqp_management_instance->on_amqp_management_open_complete_context, AMQP_MANAGEMENT_OPEN_OK);
- }
- break;
+ switch (new_state)
+ {
+ case MESSAGE_SENDER_STATE_OPENING:
+ /* Codes_SRS_AMQP_MANAGEMENT_01_165: [ - If `new_state` is `MESSAGE_SENDER_STATE_OPEING` the transition shall be ignored. ]*/
+ break;
+
+ default:
+ /* Codes_SRS_AMQP_MANAGEMENT_01_140: [ - If `new_state` is `MESSAGE_SENDER_STATE_IDLE`, `MESSAGE_SENDER_STATE_CLOSING` or `MESSAGE_SENDER_STATE_ERROR`, the `on_amqp_management_open_complete` callback shall be called with `AMQP_MANAGEMENT_OPEN_ERROR`, while also passing the context passed in `amqp_management_open_async`. ]*/
+ case MESSAGE_SENDER_STATE_IDLE:
+ case MESSAGE_SENDER_STATE_CLOSING:
+ case MESSAGE_SENDER_STATE_ERROR:
+ amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE;
+ amqp_management_instance->on_amqp_management_open_complete(amqp_management_instance->on_amqp_management_open_complete_context, AMQP_MANAGEMENT_OPEN_ERROR);
+ break;
- case AMQP_MANAGEMENT_STATE_OPEN:
- amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_ERROR;
- amqp_management_instance->on_amqp_management_error(amqp_management_instance->on_amqp_management_error_context);
- break;
- }
-
- break;
+ case MESSAGE_SENDER_STATE_OPEN:
+ amqp_management_instance->sender_connected = -1;
+ /* Codes_SRS_AMQP_MANAGEMENT_01_142: [ - If `new_state` is `MESSAGE_SENDER_STATE_OPEN` and the message receiver did not yet indicate its state as `MESSAGE_RECEIVER_STATE_OPEN`, the `on_amqp_management_open_complete` callback shall not be called.]*/
+ if (amqp_management_instance->receiver_connected != 0)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_141: [ - If `new_state` is `MESSAGE_SENDER_STATE_OPEN` and the message receiver already indicated its state as `MESSAGE_RECEIVER_STATE_OPEN`, the `on_amqp_management_open_complete` callback shall be called with `AMQP_MANAGEMENT_OPEN_OK`, while also passing the context passed in `amqp_management_open_async`. ]*/
+ amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_OPEN;
+ amqp_management_instance->on_amqp_management_open_complete(amqp_management_instance->on_amqp_management_open_complete_context, AMQP_MANAGEMENT_OPEN_OK);
+ }
+ break;
+ }
+ break;
+ }
+ /* Codes_SRS_AMQP_MANAGEMENT_01_144: [ For the current state of AMQP management being `OPEN`: ]*/
+ case AMQP_MANAGEMENT_STATE_OPEN:
+ {
+ switch (new_state)
+ {
+ default:
+ /* Codes_SRS_AMQP_MANAGEMENT_01_143: [ - If `new_state` is `MESSAGE_SENDER_STATE_IDLE`, `MESSAGE_SENDER_STATE_OPENING`, `MESSAGE_SENDER_STATE_CLOSING` or `MESSAGE_SENDER_STATE_ERROR` the `on_amqp_management_error` callback shall be invoked while passing the `on_amqp_management_error_context` as argument. ]*/
+ case MESSAGE_SENDER_STATE_IDLE:
+ case MESSAGE_SENDER_STATE_CLOSING:
+ case MESSAGE_SENDER_STATE_ERROR:
+ amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_ERROR;
+ amqp_management_instance->on_amqp_management_error(amqp_management_instance->on_amqp_management_error_context);
+ break;
- case MESSAGE_SENDER_STATE_CLOSING:
- case MESSAGE_SENDER_STATE_IDLE:
- case MESSAGE_SENDER_STATE_ERROR:
- amqp_management_instance->sender_connected = 0;
- switch (amqp_management_instance->amqp_management_state)
- {
- default:
- case AMQP_MANAGEMENT_STATE_ERROR:
- /* do nothing */
- break;
-
- case AMQP_MANAGEMENT_STATE_OPENING:
- amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE;
- amqp_management_instance->on_amqp_management_open_complete(amqp_management_instance->on_amqp_management_open_complete_context, AMQP_MANAGEMENT_OPEN_ERROR);
- break;
-
- case AMQP_MANAGEMENT_STATE_OPEN:
- amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_ERROR;
- amqp_management_instance->on_amqp_management_error(amqp_management_instance->on_amqp_management_error_context);
- break;
+ case MESSAGE_SENDER_STATE_OPEN:
+ /* Codes_SRS_AMQP_MANAGEMENT_01_145: [ - If `new_state` is `MESSAGE_SENDER_STATE_OPEN`, `on_message_sender_state_changed` shall do nothing. ]*/
+ break;
+ }
+ break;
+ }
+ /* Codes_SRS_AMQP_MANAGEMENT_01_146: [ For the current state of AMQP management being `ERROR`: ]*/
+ case AMQP_MANAGEMENT_STATE_ERROR:
+ /* Codes_SRS_AMQP_MANAGEMENT_01_147: [ - All state transitions shall be ignored. ]*/
+ break;
+ }
}
- break;
}
}
static void on_message_receiver_state_changed(const void* context, MESSAGE_RECEIVER_STATE new_state, MESSAGE_RECEIVER_STATE previous_state)
{
- AMQP_MANAGEMENT_INSTANCE* amqp_management_instance = (AMQP_MANAGEMENT_INSTANCE*)context;
- (void)previous_state;
- switch (new_state)
+ if (context == NULL)
{
- default:
- break;
-
- case MESSAGE_RECEIVER_STATE_OPEN:
- switch (amqp_management_instance->amqp_management_state)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_149: [ When `on_message_receiver_state_changed` is called with NULL `context`, it shall do nothing. ]*/
+ LogError("on_message_receiver_state_changed called with NULL context");
+ }
+ else
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_150: [ When `on_message_receiver_state_changed` is called and the `new_state` is different than `previous_state`, the following actions shall be taken: ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_160: [ When no state change is detected, `on_message_receiver_state_changed` shall do nothing. ]*/
+ if (new_state != previous_state)
{
- default:
- case AMQP_MANAGEMENT_STATE_ERROR:
- /* do nothing */
- break;
-
- case AMQP_MANAGEMENT_STATE_OPENING:
- amqp_management_instance->receiver_connected = -1;
+ AMQP_MANAGEMENT_INSTANCE* amqp_management_instance = (AMQP_MANAGEMENT_INSTANCE*)context;
+ switch (amqp_management_instance->amqp_management_state)
+ {
+ default:
+ break;
- if (amqp_management_instance->sender_connected != 0)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_151: [ For the current state of AMQP management being `OPENING`: ]*/
+ case AMQP_MANAGEMENT_STATE_OPENING:
{
- (void)send_operation_messages(amqp_management_instance);
- amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_OPEN;
- amqp_management_instance->on_amqp_management_open_complete(amqp_management_instance->on_amqp_management_open_complete_context, AMQP_MANAGEMENT_OPEN_OK);
- }
- break;
+ switch (new_state)
+ {
+ case MESSAGE_RECEIVER_STATE_OPENING:
+ /* Codes_SRS_AMQP_MANAGEMENT_01_164: [ - If `new_state` is `MESSAGE_RECEIVER_STATE_OPEING` the transition shall be ignored. ]*/
+ break;
+
+ default:
+ /* Codes_SRS_AMQP_MANAGEMENT_01_152: [ - If `new_state` is `MESSAGE_RECEIVER_STATE_IDLE`, `MESSAGE_RECEIVER_STATE_CLOSING` or `MESSAGE_RECEIVER_STATE_ERROR`, the `on_amqp_management_open_complete` callback shall be called with `AMQP_MANAGEMENT_OPEN_ERROR`, while also passing the context passed in `amqp_management_open_async`. ]*/
+ case MESSAGE_RECEIVER_STATE_IDLE:
+ case MESSAGE_RECEIVER_STATE_CLOSING:
+ case MESSAGE_RECEIVER_STATE_ERROR:
+ amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE;
+ amqp_management_instance->on_amqp_management_open_complete(amqp_management_instance->on_amqp_management_open_complete_context, AMQP_MANAGEMENT_OPEN_ERROR);
+ break;
- case AMQP_MANAGEMENT_STATE_OPEN:
- amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_ERROR;
- amqp_management_instance->on_amqp_management_error(amqp_management_instance->on_amqp_management_error_context);
- break;
- }
-
- break;
+ case MESSAGE_RECEIVER_STATE_OPEN:
+ amqp_management_instance->receiver_connected = -1;
+ /* Codes_SRS_AMQP_MANAGEMENT_01_154: [ - If `new_state` is `MESSAGE_RECEIVER_STATE_OPEN` and the message sender did not yet indicate its state as `MESSAGE_RECEIVER_STATE_OPEN`, the `on_amqp_management_open_complete` callback shall not be called. ]*/
+ if (amqp_management_instance->sender_connected != 0)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_153: [ - If `new_state` is `MESSAGE_RECEIVER_STATE_OPEN` and the message sender already indicated its state as `MESSAGE_RECEIVER_STATE_OPEN`, the `on_amqp_management_open_complete` callback shall be called with `AMQP_MANAGEMENT_OPEN_OK`, while also passing the context passed in `amqp_management_open_async`. ]*/
+ amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_OPEN;
+ amqp_management_instance->on_amqp_management_open_complete(amqp_management_instance->on_amqp_management_open_complete_context, AMQP_MANAGEMENT_OPEN_OK);
+ }
+ break;
+ }
+ break;
+ }
+ /* Codes_SRS_AMQP_MANAGEMENT_01_155: [ For the current state of AMQP management being `OPEN`: ]*/
+ case AMQP_MANAGEMENT_STATE_OPEN:
+ {
+ switch (new_state)
+ {
+ default:
+ /* Codes_SRS_AMQP_MANAGEMENT_01_156: [ - If `new_state` is `MESSAGE_RECEIVER_STATE_IDLE`, `MESSAGE_RECEIVER_STATE_OPENING`, `MESSAGE_RECEIVER_STATE_CLOSING` or `MESSAGE_RECEIVER_STATE_ERROR` the `on_amqp_management_error` callback shall be invoked while passing the `on_amqp_management_error_context` as argument. ]*/
+ case MESSAGE_RECEIVER_STATE_IDLE:
+ case MESSAGE_RECEIVER_STATE_CLOSING:
+ case MESSAGE_RECEIVER_STATE_ERROR:
+ amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_ERROR;
+ amqp_management_instance->on_amqp_management_error(amqp_management_instance->on_amqp_management_error_context);
+ break;
- case MESSAGE_RECEIVER_STATE_CLOSING:
- case MESSAGE_RECEIVER_STATE_IDLE:
- case MESSAGE_RECEIVER_STATE_ERROR:
- amqp_management_instance->sender_connected = 0;
- switch (amqp_management_instance->amqp_management_state)
- {
- default:
- case AMQP_MANAGEMENT_STATE_ERROR:
- /* do nothing */
- break;
-
- case AMQP_MANAGEMENT_STATE_OPENING:
- amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE;
- amqp_management_instance->on_amqp_management_open_complete(amqp_management_instance->on_amqp_management_open_complete_context, AMQP_MANAGEMENT_OPEN_ERROR);
- break;
-
- case AMQP_MANAGEMENT_STATE_OPEN:
- amqp_management_instance->amqp_management_state = AMQP_MANAGEMENT_STATE_ERROR;
- amqp_management_instance->on_amqp_management_error(amqp_management_instance->on_amqp_management_error_context);
- break;
+ case MESSAGE_RECEIVER_STATE_OPEN:
+ /* Codes_SRS_AMQP_MANAGEMENT_01_157: [ - If `new_state` is `MESSAGE_RECEIVER_STATE_OPEN`, `on_message_receiver_state_changed` shall do nothing. ]*/
+ break;
+ }
+ break;
+ }
+ /* Codes_SRS_AMQP_MANAGEMENT_01_158: [ For the current state of AMQP management being `ERROR`: ]*/
+ case AMQP_MANAGEMENT_STATE_ERROR:
+ /* Codes_SRS_AMQP_MANAGEMENT_01_159: [ - All state transitions shall be ignored. ]*/
+ break;
+ }
}
- break;
}
}
-static int set_message_id(MESSAGE_HANDLE message, unsigned long next_message_id)
+static int set_message_id(MESSAGE_HANDLE message, uint64_t next_message_id)
{
- int result = 0;
+ int result;
+ PROPERTIES_HANDLE properties;
- PROPERTIES_HANDLE properties;
+ /* Codes_SRS_AMQP_MANAGEMENT_01_094: [ In order to set the message Id on the message, the properties shall be obtained by calling `message_get_properties`. ]*/
if (message_get_properties(message, &properties) != 0)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_098: [ If any API fails while setting the message Id, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not retrieve message properties");
result = __FAILURE__;
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_099: [ If the properties were not set on the message, a new properties instance shall be created by calling `properties_create`. ]*/
if (properties == NULL)
{
properties = properties_create();
@@ -408,30 +505,45 @@
if (properties == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_098: [ If any API fails while setting the message Id, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not create message properties");
result = __FAILURE__;
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_095: [ A message Id with the next ulong value to be used shall be created by calling `amqpvalue_create_message_id_ulong`. ]*/
AMQP_VALUE message_id = amqpvalue_create_message_id_ulong(next_message_id);
if (message_id == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_098: [ If any API fails while setting the message Id, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not create message id value");
result = __FAILURE__;
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_096: [ The message Id value shall be set on the properties by calling `properties_set_message_id`. ]*/
if (properties_set_message_id(properties, message_id) != 0)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_098: [ If any API fails while setting the message Id, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not set message Id on the properties");
result = __FAILURE__;
}
+ /* Codes_SRS_AMQP_MANAGEMENT_01_097: [ The properties thus modified to contain the message Id shall be set on the message by calling `message_set_properties`. ]*/
+ else if (message_set_properties(message, properties) != 0)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_098: [ If any API fails while setting the message Id, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not set message properties");
+ result = __FAILURE__;
+ }
+ else
+ {
+ result = 0;
+ }
amqpvalue_destroy(message_id);
}
- if (message_set_properties(message, properties) != 0)
- {
- result = __FAILURE__;
- }
-
+ /* Codes_SRS_AMQP_MANAGEMENT_01_100: [ After setting the properties, the properties instance shall be freed by `properties_destroy`. ]*/
properties_destroy(properties);
}
}
@@ -443,22 +555,31 @@
{
int result;
+ /* Codes_SRS_AMQP_MANAGEMENT_01_084: [ For each of the arguments `operation`, `type` and `locales` an AMQP value of type string shall be created by calling `amqpvalue_create_string` in order to be used as key in the application properties map. ]*/
AMQP_VALUE key_value = amqpvalue_create_string(key);
- if (key == NULL)
+ if (key_value == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_090: [ If any APIs used to create and set the application properties on the message fails, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not create key value for %s", key);
result = __FAILURE__;
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_085: [ For each of the arguments `operation`, `type` and `locales` an AMQP value of type string containing the argument value shall be created by calling `amqpvalue_create_string` in order to be used as value in the application properties map. ]*/
AMQP_VALUE value_value = amqpvalue_create_string(value);
if (value_value == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_090: [ If any APIs used to create and set the application properties on the message fails, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not create value for key %s", key);
result = __FAILURE__;
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_086: [ The key/value pairs for `operation`, `type` and `locales` shall be added to the application properties map by calling `amqpvalue_set_map_value`. ]*/
if (amqpvalue_set_map_value(map, key_value, value_value) != 0)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_090: [ If any APIs used to create and set the application properties on the message fails, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not set the value in the map for key %s", key);
result = __FAILURE__;
}
else
@@ -466,10 +587,10 @@
result = 0;
}
- amqpvalue_destroy(key_value);
+ amqpvalue_destroy(value_value);
}
- amqpvalue_destroy(value_value);
+ amqpvalue_destroy(key_value);
}
return result;
@@ -479,96 +600,140 @@
{
AMQP_MANAGEMENT_INSTANCE* result;
- if (session == NULL)
+ if ((session == NULL) ||
+ (management_node == NULL))
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_002: [ If `session` or `management_node` is NULL then `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Bad arguments: session = %p, management_node = %p", session, management_node);
+ result = NULL;
+ }
+ else if (strlen(management_node) == 0)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_030: [ If `management_node` is an empty string, then `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Empty string management node");
result = NULL;
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_001: [ `amqp_management_create` shall create a new CBS instance and on success return a non-NULL handle to it. ]*/
result = (AMQP_MANAGEMENT_INSTANCE*)malloc(sizeof(AMQP_MANAGEMENT_INSTANCE));
- if (result != NULL)
+ if (result == NULL)
{
- result->session = session;
+ /* Codes_SRS_AMQP_MANAGEMENT_01_005: [ If allocating memory for the new handle fails, `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Cannot allocate memory for AMQP management handle");
+ }
+ else
+ {
result->sender_connected = 0;
result->receiver_connected = 0;
- result->operation_message_count = 0;
- result->operation_messages = NULL;
result->on_amqp_management_open_complete = NULL;
result->on_amqp_management_open_complete_context = NULL;
result->on_amqp_management_error = NULL;
result->on_amqp_management_error_context = NULL;
result->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE;
- AMQP_VALUE source = messaging_create_source(management_node);
- if (source == NULL)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_003: [ `amqp_management_create` shall create a singly linked list for pending operations by calling `singlylinkedlist_create`. ]*/
+ result->pending_operations = singlylinkedlist_create();
+ if (result->pending_operations == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_004: [ If `singlylinkedlist_create` fails, `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Cannot create pending operations list");
free(result);
result = NULL;
}
else
{
- AMQP_VALUE target = messaging_create_target(management_node);
- if (target == NULL)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_010: [ The `source` argument shall be a value created by calling `messaging_create_source` with `management_node` as argument. ]*/
+ AMQP_VALUE source = messaging_create_source(management_node);
+ if (source == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_012: [ If `messaging_create_source` fails then `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Failed creating source AMQP value");
+ singlylinkedlist_destroy(result->pending_operations);
free(result);
result = NULL;
}
else
{
- static const char* sender_suffix = "-sender";
-
- char* sender_link_name = (char*)malloc(strlen(management_node) + strlen(sender_suffix) + 1);
- if (sender_link_name == NULL)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_011: [ The `target` argument shall be a value created by calling `messaging_create_target` with `management_node` as argument. ]*/
+ AMQP_VALUE target = messaging_create_target(management_node);
+ if (target == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_013: [ If `messaging_create_target` fails then `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Failed creating target AMQP value");
+ singlylinkedlist_destroy(result->pending_operations);
+ free(result);
result = NULL;
}
else
{
- static const char* receiver_suffix = "-receiver";
+ size_t management_node_length = strlen(management_node);
- (void)strcpy(sender_link_name, management_node);
- (void)strcat(sender_link_name, sender_suffix);
-
- char* receiver_link_name = (char*)malloc(strlen(management_node) + strlen(receiver_suffix) + 1);
- if (receiver_link_name == NULL)
+ char* sender_link_name = (char*)malloc(management_node_length + COUNT_CHARS(sender_suffix) + 1);
+ if (sender_link_name == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_033: [ If any other error occurs `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Failed allocating memory for sender link name");
+ free(result);
result = NULL;
}
else
{
- (void)strcpy(receiver_link_name, management_node);
- (void)strcat(receiver_link_name, receiver_suffix);
+ (void)memcpy(sender_link_name, management_node, management_node_length);
+ (void)memcpy(sender_link_name + management_node_length, sender_suffix, COUNT_CHARS(sender_suffix) + 1);
- result->sender_link = link_create(session, "cbs-sender", role_sender, source, target);
- if (result->sender_link == NULL)
+ char* receiver_link_name = (char*)malloc(management_node_length + COUNT_CHARS(receiver_suffix) + 1);
+ if (receiver_link_name == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_033: [ If any other error occurs `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Failed allocating memory for receiver link name");
free(result);
result = NULL;
}
else
{
- result->receiver_link = link_create(session, "cbs-receiver", role_receiver, source, target);
- if (result->receiver_link == NULL)
+ (void)memcpy(receiver_link_name, management_node, management_node_length);
+ (void)memcpy(receiver_link_name + management_node_length, receiver_suffix, COUNT_CHARS(receiver_suffix) + 1);
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_006: [ `amqp_management_create` shall create a sender link by calling `link_create`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_007: [ The `session` argument shall be set to `session`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_008: [ The `name` argument shall be constructed by concatenating the `management_node` value with `-sender`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_009: [ The `role` argument shall be `role_sender`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_019: [ The `source` argument shall be the value created by calling `messaging_create_source`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_020: [ The `target` argument shall be the value created by calling `messaging_create_target`. ]*/
+ result->sender_link = link_create(session, sender_link_name, role_sender, source, target);
+ if (result->sender_link == NULL)
{
- link_destroy(result->sender_link);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_014: [ If `link_create` fails when creating the sender link then `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Failed creating sender link");
free(result);
result = NULL;
}
else
{
- if ((link_set_max_message_size(result->sender_link, 65535) != 0) ||
- (link_set_max_message_size(result->receiver_link, 65535) != 0))
+ /* Codes_SRS_AMQP_MANAGEMENT_01_015: [ `amqp_management_create` shall create a receiver link by calling `link_create`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_016: [ The `session` argument shall be set to `session`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_017: [ The `name` argument shall be constructed by concatenating the `management_node` value with `-receiver`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_018: [ The `role` argument shall be `role_receiver`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_019: [ The `source` argument shall be the value created by calling `messaging_create_source`. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_020: [ The `target` argument shall be the value created by calling `messaging_create_target`. ]*/
+ result->receiver_link = link_create(session, receiver_link_name, role_receiver, source, target);
+ if (result->receiver_link == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_021: [ If `link_create` fails when creating the receiver link then `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Failed creating receiver link");
link_destroy(result->sender_link);
- link_destroy(result->receiver_link);
free(result);
result = NULL;
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_022: [ `amqp_management_create` shall create a message sender by calling `messagesender_create` and passing to it the sender link handle. ]*/
result->message_sender = messagesender_create(result->sender_link, on_message_sender_state_changed, result);
if (result->message_sender == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_031: [ If `messagesender_create` fails then `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Failed creating message sender");
link_destroy(result->sender_link);
link_destroy(result->receiver_link);
free(result);
@@ -576,9 +741,12 @@
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_023: [ `amqp_management_create` shall create a message receiver by calling `messagereceiver_create` and passing to it the receiver link handle. ]*/
result->message_receiver = messagereceiver_create(result->receiver_link, on_message_receiver_state_changed, result);
if (result->message_receiver == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_032: [ If `messagereceiver_create` fails then `amqp_management_create` shall fail and return NULL. ]*/
+ LogError("Failed creating message receiver");
messagesender_destroy(result->message_sender);
link_destroy(result->sender_link);
link_destroy(result->receiver_link);
@@ -587,23 +755,24 @@
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_106: [ The message Id set on the message properties shall start at 0. ]*/
result->next_message_id = 0;
}
}
}
}
+
+ free(receiver_link_name);
}
- free(receiver_link_name);
+ free(sender_link_name);
}
- free(sender_link_name);
+ amqpvalue_destroy(target);
}
- amqpvalue_destroy(target);
+ amqpvalue_destroy(source);
}
-
- amqpvalue_destroy(source);
}
}
}
@@ -613,26 +782,28 @@
void amqp_management_destroy(AMQP_MANAGEMENT_HANDLE amqp_management)
{
- if (amqp_management != NULL)
+ if (amqp_management == NULL)
{
- (void)amqp_management_close(amqp_management);
-
- if (amqp_management->operation_message_count > 0)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_025: [ If `amqp_management` is NULL, `amqp_management_destroy` shall do nothing. ]*/
+ LogError("NULL amqp_management");
+ }
+ else
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_024: [ `amqp_management_destroy` shall free all the resources allocated by `amqp_management_create`. ]*/
+ if (amqp_management->amqp_management_state != AMQP_MANAGEMENT_STATE_IDLE)
{
- size_t i;
- for (i = 0; i < amqp_management->operation_message_count; i++)
- {
- message_destroy(amqp_management->operation_messages[i]->message);
- free(amqp_management->operation_messages[i]);
- }
-
- free(amqp_management->operation_messages);
+ (void)amqp_management_close(amqp_management);
}
+ /* Codes_SRS_AMQP_MANAGEMENT_01_028: [ `amqp_management_destroy` shall free the message sender by calling `messagesender_destroy`. ]*/
+ messagesender_destroy(amqp_management->message_sender);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_029: [ `amqp_management_destroy` shall free the message receiver by calling `messagereceiver_destroy`. ]*/
+ messagereceiver_destroy(amqp_management->message_receiver);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_027: [ `amqp_management_destroy` shall free the sender and receiver links by calling `link_destroy`. ]*/
link_destroy(amqp_management->sender_link);
link_destroy(amqp_management->receiver_link);
- messagesender_destroy(amqp_management->message_sender);
- messagereceiver_destroy(amqp_management->message_receiver);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_026: [ `amqp_management_destroy` shall free the singly linked list by calling `singlylinkedlist_destroy`. ]*/
+ singlylinkedlist_destroy(amqp_management->pending_operations);
free(amqp_management);
}
}
@@ -641,33 +812,55 @@
{
int result;
- if (amqp_management == NULL)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_044: [ `on_amqp_management_open_complete_context` and `on_amqp_management_error_context` shall be allowed to be NULL. ]*/
+ if ((amqp_management == NULL) ||
+ (on_amqp_management_open_complete == NULL) ||
+ (on_amqp_management_error == NULL))
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_038: [ If `amqp_management`, `on_amqp_management_open_complete` or `on_amqp_management_error` is NULL, `amqp_management_open_async` shall fail and return a non-zero value. ]*/
+ LogError("Bad arguments: amqp_management = %p, on_amqp_management_open_complete = %p, on_amqp_management_error = %p",
+ amqp_management,
+ on_amqp_management_open_complete,
+ on_amqp_management_error);
+ result = __FAILURE__;
+ }
+ else if (amqp_management->amqp_management_state != AMQP_MANAGEMENT_STATE_IDLE)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_043: [ If the AMQP management instance is already OPEN or OPENING, `amqp_management_open_async` shall fail and return a non-zero value. ]*/
+ LogError("AMQP management instance already OPEN");
result = __FAILURE__;
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_036: [ `amqp_management_open_async` shall start opening the AMQP management instance and save the callbacks so that they can be called when opening is complete. ]*/
amqp_management->on_amqp_management_open_complete = on_amqp_management_open_complete;
amqp_management->on_amqp_management_open_complete_context = on_amqp_management_open_complete_context;
amqp_management->on_amqp_management_error = on_amqp_management_error;
amqp_management->on_amqp_management_error_context = on_amqp_management_error_context;
amqp_management->amqp_management_state = AMQP_MANAGEMENT_STATE_OPENING;
+ /* Codes_SRS_AMQP_MANAGEMENT_01_040: [ `amqp_management_open_async` shall open the message receiver by calling `messagereceiver_open`. ]*/
if (messagereceiver_open(amqp_management->message_receiver, on_message_received, amqp_management) != 0)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_042: [ If `messagereceiver_open` fails, `amqp_management_open_async` shall fail and return a non-zero value. ]*/
+ LogError("Failed opening message receiver");
amqp_management->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE;
result = __FAILURE__;
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_039: [ `amqp_management_open_async` shall open the message sender by calling `messagesender_open`. ]*/
if (messagesender_open(amqp_management->message_sender) != 0)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_041: [ If `messagesender_open` fails, `amqp_management_open_async` shall fail and return a non-zero value. ]*/
+ LogError("Failed opening message sender");
amqp_management->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE;
- messagereceiver_close(amqp_management->message_receiver);
+ (void)messagereceiver_close(amqp_management->message_receiver);
result = __FAILURE__;
}
else
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_037: [ On success it shall return 0. ]*/
result = 0;
}
}
@@ -682,18 +875,67 @@
if (amqp_management == NULL)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_047: [ If `amqp_management` is NULL, `amqp_management_close` shall fail and return a non-zero value. ]*/
+ LogError("NULL amqp_management");
+ result = __FAILURE__;
+ }
+ else if (amqp_management->amqp_management_state == AMQP_MANAGEMENT_STATE_IDLE)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_049: [ `amqp_management_close` on an AMQP management instance that is not OPEN, shall fail and return a non-zero value. ]*/
+ LogError("AMQP management instance not open");
result = __FAILURE__;
}
else
{
- if ((messagesender_close(amqp_management->message_sender) != 0) ||
- (messagereceiver_close(amqp_management->message_receiver) != 0))
+ /* Codes_SRS_AMQP_MANAGEMENT_01_045: [ `amqp_management_close` shall close the AMQP management instance. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_050: [ `amqp_management_close` shall close the message sender by calling `messagesender_close`. ]*/
+ if (messagesender_close(amqp_management->message_sender) != 0)
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_052: [ If `messagesender_close` fails, `amqp_management_close` shall fail and return a non-zero value. ]*/
+ LogError("messagesender_close failed");
+ result = __FAILURE__;
+ }
+ /* Codes_SRS_AMQP_MANAGEMENT_01_051: [ `amqp_management_close` shall close the message receiver by calling `messagereceiver_close`. ]*/
+ else if (messagereceiver_close(amqp_management->message_receiver) != 0)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_053: [ If `messagereceiver_close` fails, `amqp_management_close` shall fail and return a non-zero value. ]*/
+ LogError("messagereceiver_close failed");
result = __FAILURE__;
}
else
{
+ LIST_ITEM_HANDLE list_item_handle = singlylinkedlist_get_head_item(amqp_management->pending_operations);
+ while (list_item_handle != NULL)
+ {
+ OPERATION_MESSAGE_INSTANCE* operation_message = (OPERATION_MESSAGE_INSTANCE*)singlylinkedlist_item_get_value(list_item_handle);
+ if (operation_message == NULL)
+ {
+ LogError("Cannot obtain pending operation");
+ }
+ else
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_054: [ All pending operations shall be indicated complete with the code `AMQP_MANAGEMENT_EXECUTE_OPERATION_INSTANCE_CLOSED`. ]*/
+ operation_message->on_execute_operation_complete(operation_message->callback_context, AMQP_MANAGEMENT_EXECUTE_OPERATION_INSTANCE_CLOSED, 0, NULL);
+ free(operation_message);
+ }
+
+ if (singlylinkedlist_remove(amqp_management->pending_operations, list_item_handle) != 0)
+ {
+ LogError("Cannot remove item");
+ }
+
+ list_item_handle = singlylinkedlist_get_head_item(amqp_management->pending_operations);
+ }
+
+ if (amqp_management->amqp_management_state == AMQP_MANAGEMENT_STATE_OPENING)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_048: [ `amqp_management_close` on an AMQP management instance that is OPENING shall trigger the `on_amqp_management_open_complete` callback with `AMQP_MANAGEMENT_OPEN_CANCELLED`, while also passing the context passed in `amqp_management_open_async`. ]*/
+ amqp_management->on_amqp_management_open_complete(amqp_management->on_amqp_management_open_complete_context, AMQP_MANAGEMENT_OPEN_CANCELLED);
+ }
+
amqp_management->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE;
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_046: [ On success it shall return 0. ]*/
result = 0;
}
}
@@ -706,90 +948,170 @@
int result;
if ((amqp_management == NULL) ||
- (operation == NULL))
+ (operation == NULL) ||
+ (type == NULL) ||
+ (on_execute_operation_complete == NULL))
{
+ /* Codes_SRS_AMQP_MANAGEMENT_01_057: [ If `amqp_management`, `operation`, `type` or `on_execute_operation_complete` is NULL, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Bad arguments: amqp_management = %p, operation = %p, type = %p",
+ amqp_management, operation, type);
+ result = __FAILURE__;
+ }
+ /* Codes_SRS_AMQP_MANAGEMENT_01_081: [ If `amqp_management_execute_operation_async` is called when not OPEN, it shall fail and return a non-zero value. ]*/
+ else if ((amqp_management->amqp_management_state == AMQP_MANAGEMENT_STATE_IDLE) ||
+ /* Codes_SRS_AMQP_MANAGEMENT_01_104: [ If `amqp_management_execute_operation_async` is called when the AMQP management is in error, it shall fail and return a non-zero value. ]*/
+ (amqp_management->amqp_management_state == AMQP_MANAGEMENT_STATE_ERROR))
+ {
+ LogError("amqp_management_execute_operation_async called while not open or in error");
result = __FAILURE__;
}
else
{
AMQP_VALUE application_properties;
- if (message_get_application_properties(message, &application_properties) != 0)
+ MESSAGE_HANDLE cloned_message;
+
+ if (message == NULL)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_102: [ If `message` is NULL, a new message shall be created by calling `message_create`. ]*/
+ cloned_message = message_create();
+ }
+ else
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_103: [ Otherwise the existing message shall be cloned by using `message_clone` before being modified accordingly and used for the pending operation. ]*/
+ cloned_message = message_clone(message);
+ if (cloned_message == NULL)
+ {
+ LogError("Could not clone message");
+ }
+ }
+
+ if (cloned_message == NULL)
{
result = __FAILURE__;
}
else
{
- if ((add_string_key_value_pair_to_map(application_properties, "operation", operation) != 0) ||
- (add_string_key_value_pair_to_map(application_properties, "type", type) != 0) ||
- ((locales != NULL) && (add_string_key_value_pair_to_map(application_properties, "locales", locales) != 0)))
+ /* Codes_SRS_AMQP_MANAGEMENT_01_055: [ `amqp_management_execute_operation_async` shall start an AMQP management operation. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_082: [ `amqp_management_execute_operation_async` shall obtain the application properties from the message by calling `message_get_application_properties`. ]*/
+ if (message_get_application_properties(cloned_message, &application_properties) != 0)
{
+ LogError("Could not get application properties");
result = __FAILURE__;
}
else
{
- if ((message_set_application_properties(message, application_properties) != 0) ||
- (set_message_id(message, amqp_management->next_message_id) != 0))
+ if (application_properties == NULL)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_083: [ If no application properties were set on the message, a new application properties instance shall be created by calling `amqpvalue_create_map`; ]*/
+ application_properties = amqpvalue_create_map();
+ if (application_properties == NULL)
+ {
+ LogError("Could not create application properties");
+ }
+ }
+
+ if (application_properties == NULL)
{
result = __FAILURE__;
}
else
{
- OPERATION_MESSAGE_INSTANCE* pending_operation_message = malloc(sizeof(OPERATION_MESSAGE_INSTANCE));
- if (pending_operation_message == NULL)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_084: [ For each of the arguments `operation`, `type` and `locales` an AMQP value of type string shall be created by calling `amqpvalue_create_string` in order to be used as key in the application properties map. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_085: [ For each of the arguments `operation`, `type` and `locales` an AMQP value of type string containing the argument value shall be created by calling `amqpvalue_create_string` in order to be used as value in the application properties map. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_058: [ Request messages have the following application-properties: ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_059: [ operation string Yes The management operation to be performed. ] */
+ if ((add_string_key_value_pair_to_map(application_properties, "operation", operation) != 0) ||
+ /* Codes_SRS_AMQP_MANAGEMENT_01_061: [ type string Yes The Manageable Entity Type of the Manageable Entity to be managed. ]*/
+ (add_string_key_value_pair_to_map(application_properties, "type", type) != 0) ||
+ /* Codes_SRS_AMQP_MANAGEMENT_01_093: [ If `locales` is NULL, no key/value pair shall be added for it in the application properties map. ]*/
+ /* Codes_SRS_AMQP_MANAGEMENT_01_063: [ locales string No A list of locales that the sending peer permits for incoming informational text in response messages. ]*/
+ ((locales != NULL) && (add_string_key_value_pair_to_map(application_properties, "locales", locales) != 0)))
{
result = __FAILURE__;
}
else
{
- pending_operation_message->message = message_clone(message);
- pending_operation_message->callback_context = on_execute_operation_complete_context;
- pending_operation_message->on_execute_operation_complete = on_execute_operation_complete;
- pending_operation_message->operation_state = OPERATION_STATE_NOT_SENT;
- pending_operation_message->message_id = amqp_management->next_message_id;
-
- amqp_management->next_message_id++;
-
- OPERATION_MESSAGE_INSTANCE** new_operation_messages = realloc(amqp_management->operation_messages, (amqp_management->operation_message_count + 1) * sizeof(OPERATION_MESSAGE_INSTANCE*));
- if (new_operation_messages == NULL)
+ /* Codes_SRS_AMQP_MANAGEMENT_01_087: [ The application properties obtained after adding the key/value pairs shall be set on the message by calling `message_set_application_properties`. ]*/
+ if (message_set_application_properties(cloned_message, application_properties) != 0)
{
- message_destroy(message);
- free(pending_operation_message);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_090: [ If any APIs used to create and set the application properties on the message fails, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not set application properties");
+ result = __FAILURE__;
+ }
+ else if (set_message_id(cloned_message, amqp_management->next_message_id) != 0)
+ {
result = __FAILURE__;
}
else
{
- amqp_management->operation_messages = new_operation_messages;
- amqp_management->operation_messages[amqp_management->operation_message_count] = pending_operation_message;
- amqp_management->operation_message_count++;
-
- if (send_operation_messages(amqp_management) != 0)
+ OPERATION_MESSAGE_INSTANCE* pending_operation_message = (OPERATION_MESSAGE_INSTANCE*)malloc(sizeof(OPERATION_MESSAGE_INSTANCE));
+ if (pending_operation_message == NULL)
{
- if (on_execute_operation_complete != NULL)
- {
- on_execute_operation_complete(on_execute_operation_complete_context, AMQP_MANAGEMENT_EXECUTE_OPERATION_ERROR, 0, NULL);
- }
-
result = __FAILURE__;
}
else
{
- result = 0;
+ LIST_ITEM_HANDLE added_item;
+ pending_operation_message->callback_context = on_execute_operation_complete_context;
+ pending_operation_message->on_execute_operation_complete = on_execute_operation_complete;
+ pending_operation_message->message_id = amqp_management->next_message_id;
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_091: [ Once the request message has been sent, an entry shall be stored in the pending operations list by calling `singlylinkedlist_add`. ]*/
+ added_item = singlylinkedlist_add(amqp_management->pending_operations, pending_operation_message);
+ if (added_item == NULL)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_092: [ If `singlylinkedlist_add` fails then `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not add the operation to the pending operations list.");
+ free(pending_operation_message);
+ result = __FAILURE__;
+ }
+ else
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_088: [ `amqp_management_execute_operation_async` shall send the message by calling `messagesender_send`. ]*/
+ if (messagesender_send(amqp_management->message_sender, cloned_message, NULL, NULL) != 0)
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_089: [ If `messagesender_send` fails, `amqp_management_execute_operation_async` shall fail and return a non-zero value. ]*/
+ LogError("Could not send request message");
+ (void)singlylinkedlist_remove(amqp_management->pending_operations, added_item);
+ free(pending_operation_message);
+ result = __FAILURE__;
+ }
+ else
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_107: [ The message Id set on the message properties shall be incremented with each operation. ]*/
+ amqp_management->next_message_id++;
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_056: [ On success it shall return 0. ]*/
+ result = 0;
+ }
+ }
}
}
}
+
+ /* Codes_SRS_AMQP_MANAGEMENT_01_101: [ After setting the application properties, the application properties instance shall be freed by `amqpvalue_destroy`. ]*/
+ amqpvalue_destroy(application_properties);
}
}
- amqpvalue_destroy(application_properties);
+ message_destroy(cloned_message);
}
}
return result;
}
-void amqp_management_set_trace(AMQP_MANAGEMENT_HANDLE amqp_management, bool traceOn)
+void amqp_management_set_trace(AMQP_MANAGEMENT_HANDLE amqp_management, bool trace_on)
{
- if (amqp_management != NULL)
+ if (amqp_management == NULL)
{
- messagesender_set_trace(amqp_management->message_sender, traceOn);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_163: [ If `amqp_management` is NULL, `amqp_management_set_trace` shal do nothing. ]*/
+ LogError("NULL amqp_management");
+ }
+ else
+ {
+ /* Codes_SRS_AMQP_MANAGEMENT_01_161: [ `amqp_management_set_trace` shall call `messagesender_set_trace` to enable/disable tracing on the message sender. ]*/
+ messagesender_set_trace(amqp_management->message_sender, trace_on);
+ /* Codes_SRS_AMQP_MANAGEMENT_01_162: [ `amqp_management_set_trace` shall call `messagereceiver_set_trace` to enable/disable tracing on the message receiver. ]*/
+ messagereceiver_set_trace(amqp_management->message_receiver, trace_on);
}
}
