File content as of revision 0:6ae2f7bca550:

// 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 <stdio.h>
#include <string.h>
#include "azure_uamqp_c/amqp_management.h"
#include "azure_uamqp_c/link.h"
#include "azure_uamqp_c/amqpalloc.h"
#include "azure_uamqp_c/message_sender.h"
#include "azure_uamqp_c/message_receiver.h"
#include "azure_uamqp_c/messaging.h"
#include "azure_uamqp_c/amqpvalue_to_string.h"
#include "azure_uamqp_c/consolelogger.h"

typedef enum OPERATION_STATE_TAG
{
	OPERATION_STATE_NOT_SENT,
	OPERATION_STATE_AWAIT_REPLY
} OPERATION_STATE;

typedef struct OPERATION_MESSAGE_INSTANCE_TAG
{
	MESSAGE_HANDLE message;
	OPERATION_STATE operation_state;
	ON_OPERATION_COMPLETE on_operation_complete;
	void* callback_context;
	unsigned long message_id;
} OPERATION_MESSAGE_INSTANCE;

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;
	ON_AMQP_MANAGEMENT_STATE_CHANGED on_amqp_management_state_changed;
	void* callback_context;
	AMQP_MANAGEMENT_STATE amqp_management_state;
	AMQP_MANAGEMENT_STATE previous_amqp_management_state;
	unsigned char sender_connected : 1;
	unsigned char receiver_connected : 1;
} AMQP_MANAGEMENT_INSTANCE;

static void amqpmanagement_set_state(AMQP_MANAGEMENT_INSTANCE* amqp_management_instance, AMQP_MANAGEMENT_STATE amqp_management_state)
{
	amqp_management_instance->previous_amqp_management_state = amqp_management_instance->amqp_management_state;
	amqp_management_instance->amqp_management_state = amqp_management_state;

	if (amqp_management_instance->on_amqp_management_state_changed != NULL)
	{
		amqp_management_instance->on_amqp_management_state_changed(amqp_management_instance->callback_context, amqp_management_instance->amqp_management_state, amqp_management_instance->previous_amqp_management_state);
	}
}

static void remove_operation_message_by_index(AMQP_MANAGEMENT_INSTANCE* amqp_management_instance, size_t index)
{
	message_destroy(amqp_management_instance->operation_messages[index]->message);
	amqpalloc_free(amqp_management_instance->operation_messages[index]);

	if (amqp_management_instance->operation_message_count - index > 1)
	{
		memmove(&amqp_management_instance->operation_messages[index], &amqp_management_instance->operation_messages[index + 1], sizeof(OPERATION_MESSAGE_INSTANCE*));
	}

	if (amqp_management_instance->operation_message_count == 1)
	{
		amqpalloc_free(amqp_management_instance->operation_messages);
		amqp_management_instance->operation_messages = NULL;
	}
	else
	{
		OPERATION_MESSAGE_INSTANCE** new_operation_messages = (OPERATION_MESSAGE_INSTANCE**)amqpalloc_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--;
}

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)
		{
			/* error */
		}
		else
		{
			AMQP_VALUE key;
			AMQP_VALUE value;
			AMQP_VALUE map;
			AMQP_VALUE correlation_id_value;

			if (properties_get_correlation_id(response_properties, &correlation_id_value) != 0)
			{
				/* error */
			}
			else
			{
				map = amqpvalue_get_inplace_described_value(application_properties);
				if (map == NULL)
				{
					/* error */
				}
				else
				{
					key = amqpvalue_create_string("status-code");
					if (key == NULL)
					{
						/* error */
					}
					else
					{
						value = amqpvalue_get_map_value(map, key);
						if (value == NULL)
						{
							/* error */
						}
						else
						{
							int32_t status_code;
							if (amqpvalue_get_int(value, &status_code) != 0)
							{
								/* error */
							}
							else
							{
								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);
										OPERATION_RESULT operation_result;

										if (expected_message_id == NULL)
										{
											break;
										}
										else
										{
											if (amqpvalue_are_equal(correlation_id_value, expected_message_id))
											{
												/* 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))
												{
													operation_result = OPERATION_RESULT_OPERATION_FAILED;
												}
												else
												{
													operation_result = OPERATION_RESULT_OK;
												}

												amqp_management_instance->operation_messages[i]->on_operation_complete(amqp_management_instance->operation_messages[i]->callback_context, operation_result, 0, NULL);

												remove_operation_message_by_index(amqp_management_instance, i);

												amqpvalue_destroy(expected_message_id);

												break;
											}

											amqpvalue_destroy(expected_message_id);
										}
									}
								}
							}

							amqpvalue_destroy(value);
						}

						amqpvalue_destroy(key);
					}
				}
			}

			properties_destroy(response_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 = __LINE__;
		}
		else
		{
			result = 0;
		}
	}
	else
	{
		result = 0;
	}

	return result;
}

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;
	switch (new_state)
	{
	default:
		break;

	case MESSAGE_SENDER_STATE_OPEN:
		amqp_management_instance->sender_connected = 1;
		(void)send_operation_messages(amqp_management_instance);
		break;

	case MESSAGE_SENDER_STATE_CLOSING:
	case MESSAGE_SENDER_STATE_ERROR:
		amqp_management_instance->sender_connected = 0;
		amqpmanagement_set_state(amqp_management_instance, AMQP_MANAGEMENT_STATE_ERROR);
		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;
	switch (new_state)
	{
	default:
		break;

	case MESSAGE_RECEIVER_STATE_OPEN:
		amqp_management_instance->receiver_connected = 1;
		(void)send_operation_messages(amqp_management_instance);
		break;

	case MESSAGE_RECEIVER_STATE_CLOSING:
	case MESSAGE_RECEIVER_STATE_ERROR:
		amqp_management_instance->receiver_connected = 0;
		amqpmanagement_set_state(amqp_management_instance, AMQP_MANAGEMENT_STATE_ERROR);
		break;
	}
}

static int set_message_id(MESSAGE_HANDLE message, unsigned long next_message_id)
{
	int result = 0;

	PROPERTIES_HANDLE properties;
	if (message_get_properties(message, &properties) != 0)
	{
		result = __LINE__;
	}
	else
	{
		AMQP_VALUE message_id = amqpvalue_create_message_id_ulong(next_message_id);
		if (message_id == NULL)
		{
			result = __LINE__;
		}
		else
		{
			if (properties_set_message_id(properties, message_id) != 0)
			{
				result = __LINE__;
			}

			amqpvalue_destroy(message_id);
		}

		if (message_set_properties(message, properties) != 0)
		{
			result = __LINE__;
		}

		properties_destroy(properties);
	}

	return result;
}

static int add_string_key_value_pair_to_map(AMQP_VALUE map, const char* key, const char* value)
{
	int result;

	AMQP_VALUE key_value = amqpvalue_create_string(key);
	if (key == NULL)
	{
		result = __LINE__;
	}
	else
	{
		AMQP_VALUE value_value = amqpvalue_create_string(value);
		if (value_value == NULL)
		{
			result = __LINE__;
		}
		else
		{
			if (amqpvalue_set_map_value(map, key_value, value_value) != 0)
			{
				result = __LINE__;
			}
			else
			{
				result = 0;
			}

			amqpvalue_destroy(key_value);
		}

		amqpvalue_destroy(value_value);
	}

	return result;
}

AMQP_MANAGEMENT_HANDLE amqpmanagement_create(SESSION_HANDLE session, const char* management_node, ON_AMQP_MANAGEMENT_STATE_CHANGED on_amqp_management_state_changed, void* callback_context)
{
	AMQP_MANAGEMENT_INSTANCE* result;

	if (session == NULL)
	{
		result = NULL;
	}
	else
	{
		result = (AMQP_MANAGEMENT_INSTANCE*)amqpalloc_malloc(sizeof(AMQP_MANAGEMENT_INSTANCE));
		if (result != NULL)
		{
			result->session = session;
			result->sender_connected = 0;
			result->receiver_connected = 0;
			result->operation_message_count = 0;
			result->operation_messages = NULL;
			result->on_amqp_management_state_changed = on_amqp_management_state_changed;
			result->callback_context = callback_context;

			AMQP_VALUE source = messaging_create_source(management_node);
			if (source == NULL)
			{
				amqpalloc_free(result);
				result = NULL;
			}
			else
			{
				AMQP_VALUE target = messaging_create_target(management_node);
				if (target == NULL)
				{
					amqpalloc_free(result);
					result = NULL;
				}
				else
				{
					static const char* sender_suffix = "-sender";

					char* sender_link_name = (char*)amqpalloc_malloc(strlen(management_node) + strlen(sender_suffix) + 1);
					if (sender_link_name == NULL)
					{
						result = NULL;
					}
					else
					{
						static const char* receiver_suffix = "-receiver";

						(void)strcpy(sender_link_name, management_node);
						(void)strcat(sender_link_name, sender_suffix);

						char* receiver_link_name = (char*)amqpalloc_malloc(strlen(management_node) + strlen(receiver_suffix) + 1);
						if (receiver_link_name == NULL)
						{
							result = NULL;
						}
						else
						{
							(void)strcpy(receiver_link_name, management_node);
							(void)strcat(receiver_link_name, receiver_suffix);

							result->sender_link = link_create(session, "cbs-sender", role_sender, source, target);
							if (result->sender_link == NULL)
							{
								amqpalloc_free(result);
								result = NULL;
							}
							else
							{
								result->receiver_link = link_create(session, "cbs-receiver", role_receiver, source, target);
								if (result->receiver_link == NULL)
								{
									link_destroy(result->sender_link);
									amqpalloc_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))
									{
										link_destroy(result->sender_link);
										link_destroy(result->receiver_link);
										amqpalloc_free(result);
										result = NULL;
									}
									else
									{
										result->message_sender = messagesender_create(result->sender_link, on_message_sender_state_changed, result, NULL);
										if (result->message_sender == NULL)
										{
											link_destroy(result->sender_link);
											link_destroy(result->receiver_link);
											amqpalloc_free(result);
											result = NULL;
										}
										else
										{
											result->message_receiver = messagereceiver_create(result->receiver_link, on_message_receiver_state_changed, result);
											if (result->message_receiver == NULL)
											{
												messagesender_destroy(result->message_sender);
												link_destroy(result->sender_link);
												link_destroy(result->receiver_link);
												amqpalloc_free(result);
												result = NULL;
											}
											else
											{
												result->next_message_id = 0;
											}
										}
									}
								}
							}

							amqpalloc_free(receiver_link_name);
						}

						amqpalloc_free(sender_link_name);
					}

					amqpvalue_destroy(target);
				}

				amqpvalue_destroy(source);
			}
		}
	}

	return result;
}

void amqpmanagement_destroy(AMQP_MANAGEMENT_HANDLE amqp_management)
{
	if (amqp_management != NULL)
	{
		(void)amqpmanagement_close(amqp_management);

		if (amqp_management->operation_message_count > 0)
		{
			size_t i;
			for (i = 0; i < amqp_management->operation_message_count; i++)
			{
				message_destroy(amqp_management->operation_messages[i]->message);
				amqpalloc_free(amqp_management->operation_messages[i]);
			}

			amqpalloc_free(amqp_management->operation_messages);
		}

		link_destroy(amqp_management->sender_link);
		link_destroy(amqp_management->receiver_link);
		messagesender_destroy(amqp_management->message_sender);
		messagereceiver_destroy(amqp_management->message_receiver);
		amqpalloc_free(amqp_management);
	}
}

int amqpmanagement_open(AMQP_MANAGEMENT_HANDLE amqp_management)
{
	int result;

	if (amqp_management == NULL)
	{
		result = __LINE__;
	}
	else
	{
		if (messagereceiver_open(amqp_management->message_receiver, on_message_received, amqp_management) != 0)
		{
			result = __LINE__;
		}
		else
		{
			if (messagesender_open(amqp_management->message_sender) != 0)
			{
				messagereceiver_close(amqp_management->message_receiver);
				result = __LINE__;
			}
			else
			{
				result = 0;
			}
		}
	}

	return result;
}

int amqpmanagement_close(AMQP_MANAGEMENT_HANDLE amqp_management)
{
	int result;

	if (amqp_management == NULL)
	{
		result = __LINE__;
	}
	else
	{
		if ((messagesender_close(amqp_management->message_sender) != 0) ||
			(messagereceiver_close(amqp_management->message_receiver) != 0))
		{
			result = __LINE__;
		}
		else
		{
			result = 0;
		}
	}

	return result;
}

int amqpmanagement_start_operation(AMQP_MANAGEMENT_HANDLE amqp_management, const char* operation, const char* type, const char* locales, MESSAGE_HANDLE message, ON_OPERATION_COMPLETE on_operation_complete, void* context)
{
	int result;

	if ((amqp_management == NULL) ||
		(operation == NULL))
	{
		result = __LINE__;
	}
	else
	{
		AMQP_VALUE application_properties;
		if (message_get_application_properties(message, &application_properties) != 0)
		{
			result = __LINE__;
		}
		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)))
			{
				result = __LINE__;
			}
			else
			{
				if ((message_set_application_properties(message, application_properties) != 0) ||
					(set_message_id(message, amqp_management->next_message_id) != 0))
				{
					result = __LINE__;
				}
				else
				{
					OPERATION_MESSAGE_INSTANCE* pending_operation_message = amqpalloc_malloc(sizeof(OPERATION_MESSAGE_INSTANCE));
					if (pending_operation_message == NULL)
					{
						result = __LINE__;
					}
					else
					{
						pending_operation_message->message = message_clone(message);
						pending_operation_message->callback_context = context;
						pending_operation_message->on_operation_complete = on_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 = amqpalloc_realloc(amqp_management->operation_messages, (amqp_management->operation_message_count + 1) * sizeof(OPERATION_MESSAGE_INSTANCE*));
						if (new_operation_messages == NULL)
						{
							message_destroy(message);
							amqpalloc_free(pending_operation_message);
							result = __LINE__;
						}
						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)
							{
								if (on_operation_complete != NULL)
								{
									on_operation_complete(context, OPERATION_RESULT_CBS_ERROR, 0, NULL);
								}

								result = __LINE__;
							}
							else
							{
								result = 0;
							}
						}
					}
				}
			}

			amqpvalue_destroy(application_properties);
		}
	}

	return result;
}