A small memory footprint AMQP implimentation
Dependents: iothub_client_sample_amqp remote_monitoring simplesample_amqp
Diff: amqp_management.c
- Revision:
- 22:524bded3f7a8
- Parent:
- 21:f9c433d8e6ca
- Child:
- 23:1111ee8bcba4
diff -r f9c433d8e6ca -r 524bded3f7a8 amqp_management.c --- a/amqp_management.c Fri Mar 24 16:35:45 2017 -0700 +++ b/amqp_management.c Thu Apr 06 14:11:27 2017 -0700 @@ -24,11 +24,19 @@ { MESSAGE_HANDLE message; OPERATION_STATE operation_state; - ON_OPERATION_COMPLETE on_operation_complete; + ON_AMQP_MANAGEMENT_EXECUTE_OPERATION_COMPLETE on_execute_operation_complete; void* callback_context; unsigned long message_id; } OPERATION_MESSAGE_INSTANCE; +typedef enum AMQP_MANAGEMENT_STATE_TAG +{ + AMQP_MANAGEMENT_STATE_IDLE, + AMQP_MANAGEMENT_STATE_OPENING, + AMQP_MANAGEMENT_STATE_OPEN, + AMQP_MANAGEMENT_STATE_ERROR +} AMQP_MANAGEMENT_STATE; + typedef struct AMQP_MANAGEMENT_INSTANCE_TAG { SESSION_HANDLE session; @@ -39,25 +47,15 @@ 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; + 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; + void* on_amqp_management_error_context; AMQP_MANAGEMENT_STATE amqp_management_state; - AMQP_MANAGEMENT_STATE previous_amqp_management_state; int sender_connected : 1; int 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); @@ -166,7 +164,7 @@ 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; + AMQP_MANAGEMENT_EXECUTE_OPERATION_RESULT execute_operation_result; if (expected_message_id == NULL) { @@ -179,14 +177,14 @@ /* 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; + execute_operation_result = AMQP_MANAGEMENT_EXECUTE_OPERATION_FAILED_BAD_STATUS; } else { - operation_result = OPERATION_RESULT_OK; + execute_operation_result = AMQP_MANAGEMENT_EXECUTE_OPERATION_OK; } - amqp_management_instance->operation_messages[i]->on_operation_complete(amqp_management_instance->operation_messages[i]->callback_context, operation_result, status_code, status_description); + 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); @@ -281,14 +279,52 @@ break; case MESSAGE_SENDER_STATE_OPEN: - amqp_management_instance->sender_connected = 1; - (void)send_operation_messages(amqp_management_instance); + switch (amqp_management_instance->amqp_management_state) + { + default: + case AMQP_MANAGEMENT_STATE_ERROR: + /* do nothing */ + break; + + case AMQP_MANAGEMENT_STATE_OPENING: + amqp_management_instance->sender_connected = -1; + if (amqp_management_instance->receiver_connected != 0) + { + (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; + + 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_CLOSING: + case MESSAGE_SENDER_STATE_IDLE: case MESSAGE_SENDER_STATE_ERROR: amqp_management_instance->sender_connected = 0; - amqpmanagement_set_state(amqp_management_instance, AMQP_MANAGEMENT_STATE_ERROR); + 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; + } break; } } @@ -303,14 +339,53 @@ break; case MESSAGE_RECEIVER_STATE_OPEN: - amqp_management_instance->receiver_connected = 1; - (void)send_operation_messages(amqp_management_instance); + switch (amqp_management_instance->amqp_management_state) + { + default: + case AMQP_MANAGEMENT_STATE_ERROR: + /* do nothing */ + break; + + case AMQP_MANAGEMENT_STATE_OPENING: + amqp_management_instance->receiver_connected = -1; + + if (amqp_management_instance->sender_connected != 0) + { + (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; + + 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_CLOSING: + case MESSAGE_RECEIVER_STATE_IDLE: case MESSAGE_RECEIVER_STATE_ERROR: - amqp_management_instance->receiver_connected = 0; - amqpmanagement_set_state(amqp_management_instance, AMQP_MANAGEMENT_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; + } break; } } @@ -400,7 +475,7 @@ 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_HANDLE amqp_management_create(SESSION_HANDLE session, const char* management_node) { AMQP_MANAGEMENT_INSTANCE* result; @@ -418,8 +493,10 @@ 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; + 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); @@ -534,11 +611,11 @@ return result; } -void amqpmanagement_destroy(AMQP_MANAGEMENT_HANDLE amqp_management) +void amqp_management_destroy(AMQP_MANAGEMENT_HANDLE amqp_management) { if (amqp_management != NULL) { - (void)amqpmanagement_close(amqp_management); + (void)amqp_management_close(amqp_management); if (amqp_management->operation_message_count > 0) { @@ -560,7 +637,7 @@ } } -int amqpmanagement_open(AMQP_MANAGEMENT_HANDLE amqp_management) +int amqp_management_open_async(AMQP_MANAGEMENT_HANDLE amqp_management, 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, void* on_amqp_management_error_context) { int result; @@ -570,14 +647,22 @@ } else { + 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; + if (messagereceiver_open(amqp_management->message_receiver, on_message_received, amqp_management) != 0) { + amqp_management->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE; result = __FAILURE__; } else { if (messagesender_open(amqp_management->message_sender) != 0) { + amqp_management->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE; messagereceiver_close(amqp_management->message_receiver); result = __FAILURE__; } @@ -591,7 +676,7 @@ return result; } -int amqpmanagement_close(AMQP_MANAGEMENT_HANDLE amqp_management) +int amqp_management_close(AMQP_MANAGEMENT_HANDLE amqp_management) { int result; @@ -608,6 +693,7 @@ } else { + amqp_management->amqp_management_state = AMQP_MANAGEMENT_STATE_IDLE; result = 0; } } @@ -615,7 +701,7 @@ 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 amqp_management_execute_operation_async(AMQP_MANAGEMENT_HANDLE amqp_management, const char* operation, const char* type, const char* locales, MESSAGE_HANDLE message, ON_AMQP_MANAGEMENT_EXECUTE_OPERATION_COMPLETE on_execute_operation_complete, void* on_execute_operation_complete_context) { int result; @@ -656,8 +742,8 @@ 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->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; @@ -678,9 +764,9 @@ if (send_operation_messages(amqp_management) != 0) { - if (on_operation_complete != NULL) + if (on_execute_operation_complete != NULL) { - on_operation_complete(context, OPERATION_RESULT_ERROR, 0, NULL); + on_execute_operation_complete(on_execute_operation_complete_context, AMQP_MANAGEMENT_EXECUTE_OPERATION_ERROR, 0, NULL); } result = __FAILURE__; @@ -700,7 +786,7 @@ return result; } -void amqpmanagement_set_trace(AMQP_MANAGEMENT_HANDLE amqp_management, bool traceOn) +void amqp_management_set_trace(AMQP_MANAGEMENT_HANDLE amqp_management, bool traceOn) { if (amqp_management != NULL) {