A small memory footprint AMQP implimentation
Dependents: iothub_client_sample_amqp remote_monitoring simplesample_amqp
Diff: link.c
- Revision:
- 41:0e723f9cbd89
- Parent:
- 37:c923ba7f6cf9
- Child:
- 43:4c1e4e94cdd3
--- a/link.c Mon Mar 05 17:41:28 2018 -0800 +++ b/link.c Tue Mar 20 10:29:38 2018 -0700 @@ -51,7 +51,8 @@ sequence_no initial_delivery_count; uint64_t max_message_size; uint64_t peer_max_message_size; - uint32_t link_credit; + uint32_t current_link_credit; + uint32_t max_link_credit; uint32_t available; fields attach_properties; bool is_underlying_session_begun; @@ -110,20 +111,31 @@ if (flow == NULL) { + LogError("NULL flow performative"); result = __FAILURE__; } else { - if ((flow_set_link_credit(flow, link->link_credit) != 0) || - (flow_set_handle(flow, link->handle) != 0) || - (flow_set_delivery_count(flow, link->delivery_count) != 0)) + if (flow_set_link_credit(flow, link->current_link_credit) != 0) + { + LogError("Cannot set link credit on flow performative"); + result = __FAILURE__; + } + else if (flow_set_handle(flow, link->handle) != 0) { + LogError("Cannot set handle on flow performative"); + result = __FAILURE__; + } + else if (flow_set_delivery_count(flow, link->delivery_count) != 0) + { + LogError("Cannot set delivery count on flow performative"); result = __FAILURE__; } else { if (session_send_flow(link->link_endpoint, flow) != 0) { + LogError("Sending flow frame failed in session send"); result = __FAILURE__; } else @@ -145,20 +157,31 @@ DISPOSITION_HANDLE disposition = disposition_create(link_instance->role, delivery_number); if (disposition == NULL) { + LogError("NULL disposition performative"); result = __FAILURE__; } else { - if ((disposition_set_last(disposition, delivery_number) != 0) || - (disposition_set_settled(disposition, true) != 0) || - ((delivery_state != NULL) && (disposition_set_state(disposition, delivery_state) != 0))) + if (disposition_set_last(disposition, delivery_number) != 0) + { + LogError("Failed setting last on disposition performative"); + result = __FAILURE__; + } + else if (disposition_set_settled(disposition, true) != 0) { + LogError("Failed setting settled on disposition performative"); + result = __FAILURE__; + } + else if ((delivery_state != NULL) && (disposition_set_state(disposition, delivery_state) != 0)) + { + LogError("Failed setting state on disposition performative"); result = __FAILURE__; } else { if (session_send_disposition(link_instance->link_endpoint, disposition) != 0) { + LogError("Sending disposition failed in session send"); result = __FAILURE__; } else @@ -181,6 +204,7 @@ detach_performative = detach_create(0); if (detach_performative == NULL) { + LogError("NULL detach performative"); result = __FAILURE__; } else @@ -188,17 +212,20 @@ if ((error_handle != NULL) && (detach_set_error(detach_performative, error_handle) != 0)) { + LogError("Failed setting error on detach frame"); result = __FAILURE__; } else if (close && (detach_set_closed(detach_performative, true) != 0)) { + LogError("Failed setting closed field on detach frame"); result = __FAILURE__; } else { if (session_send_detach(link_instance->link_endpoint, detach_performative) != 0) { + LogError("Sending detach frame failed in session send"); result = __FAILURE__; } else @@ -226,6 +253,7 @@ if (attach == NULL) { + LogError("NULL attach performative"); result = __FAILURE__; } else @@ -241,22 +269,28 @@ attach_set_target(attach, link->target); if (link->attach_properties != NULL) { - attach_set_properties(attach, link->attach_properties); + (void)attach_set_properties(attach, link->attach_properties); } if (role == role_sender) { if (attach_set_initial_delivery_count(attach, link->delivery_count) != 0) { + LogError("Cannot set attach initial delivery count"); result = __FAILURE__; } } if (result == 0) { - if ((attach_set_max_message_size(attach, link->max_message_size) != 0) || - (session_send_attach(link->link_endpoint, attach) != 0)) + if (attach_set_max_message_size(attach, link->max_message_size) != 0) { + LogError("Cannot set max message size"); + result = __FAILURE__; + } + else if (session_send_attach(link->link_endpoint, attach) != 0) + { + LogError("Sending attach failed in session send"); result = __FAILURE__; } else @@ -280,12 +314,16 @@ { ATTACH_HANDLE attach_handle; - if (amqpvalue_get_attach(performative, &attach_handle) == 0) + if (amqpvalue_get_attach(performative, &attach_handle) != 0) + { + LogError("Cannot get attach performative"); + } + else { if ((link_instance->role == role_receiver) && (attach_get_initial_delivery_count(attach_handle, &link_instance->delivery_count) != 0)) { - /* error */ + LogError("Cannot get initial delivery count"); remove_all_pending_deliveries(link_instance, true); set_link_state(link_instance, LINK_STATE_DETACHED); } @@ -301,12 +339,12 @@ { if (link_instance->role == role_receiver) { - link_instance->link_credit = DEFAULT_LINK_CREDIT; + link_instance->current_link_credit = link_instance->max_link_credit; send_flow(link_instance); } else { - link_instance->link_credit = 0; + link_instance->current_link_credit = 0; } if (link_instance->link_state == LINK_STATE_DETACHED) @@ -326,24 +364,33 @@ else if (is_flow_type_by_descriptor(descriptor)) { FLOW_HANDLE flow_handle; - if (amqpvalue_get_flow(performative, &flow_handle) == 0) + if (amqpvalue_get_flow(performative, &flow_handle) != 0) + { + LogError("Cannot get flow performative"); + } + else { if (link_instance->role == role_sender) { delivery_number rcv_delivery_count; uint32_t rcv_link_credit; - if ((flow_get_link_credit(flow_handle, &rcv_link_credit) != 0) || - (flow_get_delivery_count(flow_handle, &rcv_delivery_count) != 0)) + if (flow_get_link_credit(flow_handle, &rcv_link_credit) != 0) { - /* error */ + LogError("Cannot get link credit"); + remove_all_pending_deliveries(link_instance, true); + set_link_state(link_instance, LINK_STATE_DETACHED); + } + else if (flow_get_delivery_count(flow_handle, &rcv_delivery_count) != 0) + { + LogError("Cannot get delivery count"); remove_all_pending_deliveries(link_instance, true); set_link_state(link_instance, LINK_STATE_DETACHED); } else { - link_instance->link_credit = rcv_delivery_count + rcv_link_credit - link_instance->delivery_count; - if (link_instance->link_credit > 0) + link_instance->current_link_credit = rcv_delivery_count + rcv_link_credit - link_instance->delivery_count; + if (link_instance->current_link_credit > 0) { link_instance->on_link_flow_on(link_instance->callback_context); } @@ -358,17 +405,21 @@ if (link_instance->on_transfer_received != NULL) { TRANSFER_HANDLE transfer_handle; - if (amqpvalue_get_transfer(performative, &transfer_handle) == 0) + if (amqpvalue_get_transfer(performative, &transfer_handle) != 0) + { + LogError("Cannot get transfer performative"); + } + else { AMQP_VALUE delivery_state; bool more; bool is_error; - link_instance->link_credit--; + link_instance->current_link_credit--; link_instance->delivery_count++; - if (link_instance->link_credit == 0) + if (link_instance->current_link_credit == 0) { - link_instance->link_credit = DEFAULT_LINK_CREDIT; + link_instance->current_link_credit = link_instance->max_link_credit; send_flow(link_instance); } @@ -437,6 +488,7 @@ { LogError("Cannot send disposition frame"); } + amqpvalue_destroy(delivery_state); } } @@ -451,7 +503,7 @@ DISPOSITION_HANDLE disposition; if (amqpvalue_get_disposition(performative, &disposition) != 0) { - /* error */ + LogError("Cannot get disposition performative"); } else { @@ -460,7 +512,7 @@ if (disposition_get_first(disposition, &first) != 0) { - /* error */ + LogError("Cannot get first field"); } else { @@ -473,7 +525,6 @@ if (disposition_get_settled(disposition, &settled) != 0) { - /* Error */ settled = false; } @@ -486,7 +537,7 @@ ASYNC_OPERATION_HANDLE pending_delivery_operation = (ASYNC_OPERATION_HANDLE)singlylinkedlist_item_get_value(pending_delivery); if (pending_delivery_operation == NULL) { - /* error */ + LogError("Cannot obtain pending delivery"); break; } else @@ -496,23 +547,17 @@ if ((delivery_instance->delivery_id >= first) && (delivery_instance->delivery_id <= last)) { AMQP_VALUE delivery_state; - if (disposition_get_state(disposition, &delivery_state) != 0) - { - /* error */ - } - else + if (disposition_get_state(disposition, &delivery_state) == 0) { delivery_instance->on_delivery_settled(delivery_instance->callback_context, delivery_instance->delivery_id, LINK_DELIVERY_SETTLE_REASON_DISPOSITION_RECEIVED, delivery_state); async_operation_destroy(pending_delivery_operation); if (singlylinkedlist_remove(link_instance->pending_deliveries, pending_delivery) != 0) { - /* error */ + LogError("Cannot remove pending delivery"); break; } - else - { - pending_delivery = next_pending_delivery; - } + + pending_delivery = next_pending_delivery; } } else @@ -532,7 +577,11 @@ DETACH_HANDLE detach; /* Set link state appropriately based on whether we received detach condition */ - if (amqpvalue_get_detach(performative, &detach) == 0) + if (amqpvalue_get_detach(performative, &detach) != 0) + { + LogError("Cannot get detach performative"); + } + else { bool closed = false; ERROR_HANDLE error; @@ -541,7 +590,10 @@ if (link_instance->link_state == LINK_STATE_ATTACHED) { /* Respond with ack */ - (void)send_detach(link_instance, closed, NULL); + if (send_detach(link_instance, closed, NULL) != 0) + { + LogError("Failed sending detach frame"); + } } /* Received a closing detach after we sent a non-closing detach. */ @@ -551,8 +603,15 @@ { /* In this case, we MUST signal that we closed by reattaching and then sending a closing detach.*/ - (void)send_attach(link_instance, link_instance->name, 0, link_instance->role); - (void)send_detach(link_instance, true, NULL); + if (send_attach(link_instance, link_instance->name, 0, link_instance->role) != 0) + { + LogError("Failed sending attach frame"); + } + + if (send_detach(link_instance, true, NULL) != 0) + { + LogError("Failed sending detach frame"); + } } if (detach_get_error(detach, &error) == 0) @@ -629,7 +688,11 @@ LINK_HANDLE link_create(SESSION_HANDLE session, const char* name, role role, AMQP_VALUE source, AMQP_VALUE target) { LINK_INSTANCE* result = (LINK_INSTANCE*)malloc(sizeof(LINK_INSTANCE)); - if (result != NULL) + if (result == NULL) + { + LogError("Cannot create link"); + } + else { result->link_state = LINK_STATE_DETACHED; result->previous_link_state = LINK_STATE_DETACHED; @@ -643,6 +706,7 @@ result->delivery_count = 0; result->initial_delivery_count = 0; result->max_message_size = 0; + result->max_link_credit = DEFAULT_LINK_CREDIT; result->peer_max_message_size = 0; result->is_underlying_session_begun = false; result->is_closed = false; @@ -654,6 +718,7 @@ result->tick_counter = tickcounter_create(); if (result->tick_counter == NULL) { + LogError("Cannot create tick counter for link"); free(result); result = NULL; } @@ -662,6 +727,7 @@ result->pending_deliveries = singlylinkedlist_create(); if (result->pending_deliveries == NULL) { + LogError("Cannot create pending deliveries list"); tickcounter_destroy(result->tick_counter); free(result); result = NULL; @@ -672,6 +738,7 @@ result->name = (char*)malloc(name_length + 1); if (result->name == NULL) { + LogError("Cannot allocate memory for link name"); tickcounter_destroy(result->tick_counter); singlylinkedlist_destroy(result->pending_deliveries); free(result); @@ -687,6 +754,7 @@ result->link_endpoint = session_create_link_endpoint(session, name); if (result->link_endpoint == NULL) { + LogError("Cannot create link endpoint"); tickcounter_destroy(result->tick_counter); singlylinkedlist_destroy(result->pending_deliveries); free(result->name); @@ -704,7 +772,11 @@ LINK_HANDLE link_create_from_endpoint(SESSION_HANDLE session, LINK_ENDPOINT_HANDLE link_endpoint, const char* name, role role, AMQP_VALUE source, AMQP_VALUE target) { LINK_INSTANCE* result = (LINK_INSTANCE*)malloc(sizeof(LINK_INSTANCE)); - if (result != NULL) + if (result == NULL) + { + LogError("Cannot create link"); + } + else { result->link_state = LINK_STATE_DETACHED; result->previous_link_state = LINK_STATE_DETACHED; @@ -715,6 +787,7 @@ result->delivery_count = 0; result->initial_delivery_count = 0; result->max_message_size = 0; + result->max_link_credit = DEFAULT_LINK_CREDIT; result->peer_max_message_size = 0; result->is_underlying_session_begun = false; result->is_closed = false; @@ -724,6 +797,7 @@ result->received_delivery_id = 0; result->source = amqpvalue_clone(target); result->target = amqpvalue_clone(source); + if (role == role_sender) { result->role = role_receiver; @@ -736,6 +810,7 @@ result->tick_counter = tickcounter_create(); if (result->tick_counter == NULL) { + LogError("Cannot create tick counter for link"); free(result); result = NULL; } @@ -744,6 +819,7 @@ result->pending_deliveries = singlylinkedlist_create(); if (result->pending_deliveries == NULL) { + LogError("Cannot create pending deliveries list"); tickcounter_destroy(result->tick_counter); free(result); result = NULL; @@ -754,6 +830,7 @@ result->name = (char*)malloc(name_length + 1); if (result->name == NULL) { + LogError("Cannot allocate memory for link name"); tickcounter_destroy(result->tick_counter); singlylinkedlist_destroy(result->pending_deliveries); free(result); @@ -775,7 +852,11 @@ void link_destroy(LINK_HANDLE link) { - if (link != NULL) + if (link == NULL) + { + LogError("NULL link"); + } + else { remove_all_pending_deliveries((LINK_INSTANCE*)link, false); tickcounter_destroy(link->tick_counter); @@ -811,6 +892,7 @@ if (link == NULL) { + LogError("NULL link"); result = __FAILURE__; } else @@ -829,6 +911,8 @@ if ((link == NULL) || (snd_settle_mode == NULL)) { + LogError("Bad arguments: link = %p, snd_settle_mode = %p", + link, snd_settle_mode); result = __FAILURE__; } else @@ -847,6 +931,7 @@ if (link == NULL) { + LogError("NULL link"); result = __FAILURE__; } else @@ -865,6 +950,8 @@ if ((link == NULL) || (rcv_settle_mode == NULL)) { + LogError("Bad arguments: link = %p, rcv_settle_mode = %p", + link, rcv_settle_mode); result = __FAILURE__; } else @@ -882,6 +969,7 @@ if (link == NULL) { + LogError("NULL link"); result = __FAILURE__; } else @@ -900,6 +988,8 @@ if ((link == NULL) || (initial_delivery_count == NULL)) { + LogError("Bad arguments: link = %p, initial_delivery_count = %p", + link, initial_delivery_count); result = __FAILURE__; } else @@ -917,6 +1007,7 @@ if (link == NULL) { + LogError("NULL link"); result = __FAILURE__; } else @@ -935,6 +1026,8 @@ if ((link == NULL) || (max_message_size == NULL)) { + LogError("Bad arguments: link = %p, max_message_size = %p", + link, max_message_size); result = __FAILURE__; } else @@ -978,6 +1071,7 @@ if (link == NULL) { + LogError("NULL link"); result = __FAILURE__; } else @@ -985,6 +1079,7 @@ link->attach_properties = amqpvalue_clone(attach_properties); if (link->attach_properties == NULL) { + LogError("Failed cloning attach properties"); result = __FAILURE__; } else @@ -996,13 +1091,35 @@ return result; } +int link_set_max_link_credit(LINK_HANDLE link, uint32_t max_link_credit) +{ + int result; + + if (link == NULL) + { + result = __FAILURE__; + } + else + { + link->max_link_credit = max_link_credit; + result = 0; + } + + return result; +} + int link_attach(LINK_HANDLE link, ON_TRANSFER_RECEIVED on_transfer_received, ON_LINK_STATE_CHANGED on_link_state_changed, ON_LINK_FLOW_ON on_link_flow_on, void* callback_context) { int result; - if ((link == NULL) || - (link->is_closed)) + if (link == NULL) { + LogError("NULL link"); + result = __FAILURE__; + } + else if (link->is_closed) + { + LogError("Already attached"); result = __FAILURE__; } else @@ -1016,6 +1133,7 @@ if (session_begin(link->session) != 0) { + LogError("Begin session failed"); result = __FAILURE__; } else @@ -1024,6 +1142,7 @@ if (session_start_link_endpoint(link->link_endpoint, link_frame_received, on_session_state_changed, on_session_flow_on, link) != 0) { + LogError("Binding link endpoint to session failed"); result = __FAILURE__; } else @@ -1047,11 +1166,15 @@ { int result; - if ((link == NULL) || - (link->is_closed)) + if (link == NULL) { + LogError("NULL link"); result = __FAILURE__; } + else if (link->is_closed) + { + result = 0; + } else { switch (link->link_state) @@ -1061,6 +1184,7 @@ /* Sending detach when remote is not yet attached */ if (send_detach(link, close, NULL) != 0) { + LogError("Sending detach frame failed"); result = __FAILURE__; } else @@ -1074,6 +1198,7 @@ /* Send detach and wait for remote to respond */ if (send_detach(link, close, NULL) != 0) { + LogError("Sending detach frame failed"); result = __FAILURE__; } else @@ -1142,17 +1267,25 @@ *link_transfer_error = LINK_TRANSFER_ERROR; } + LogError("Invalid arguments: link = %p, link_transfer_error = %p", + link, link_transfer_error); result = NULL; } else { - if ((link->role != role_sender) || - (link->link_state != LINK_STATE_ATTACHED)) + if (link->role != role_sender) { + LogError("Link is not a sender link"); *link_transfer_error = LINK_TRANSFER_ERROR; result = NULL; } - else if (link->link_credit == 0) + else if (link->link_state != LINK_STATE_ATTACHED) + { + LogError("Link is not attached"); + *link_transfer_error = LINK_TRANSFER_ERROR; + result = NULL; + } + else if (link->current_link_credit == 0) { *link_transfer_error = LINK_TRANSFER_BUSY; result = NULL; @@ -1160,11 +1293,17 @@ else { result = CREATE_ASYNC_OPERATION(DELIVERY_INSTANCE, link_transfer_cancel_handler); - if (result != NULL) + if (result == NULL) + { + LogError("Error creating async operation"); + *link_transfer_error = LINK_TRANSFER_ERROR; + } + else { TRANSFER_HANDLE transfer = transfer_create(0); if (transfer == NULL) { + LogError("Error creating transfer"); *link_transfer_error = LINK_TRANSFER_ERROR; async_operation_destroy(result); result = NULL; @@ -1190,10 +1329,23 @@ settled = true; } - if ((transfer_set_delivery_tag(transfer, delivery_tag) != 0) || - (transfer_set_message_format(transfer, message_format) != 0) || - (transfer_set_settled(transfer, settled) != 0)) + if (transfer_set_delivery_tag(transfer, delivery_tag) != 0) + { + LogError("Failed setting delivery tag"); + *link_transfer_error = LINK_TRANSFER_ERROR; + async_operation_destroy(result); + result = NULL; + } + else if (transfer_set_message_format(transfer, message_format) != 0) { + LogError("Failed setting message format"); + *link_transfer_error = LINK_TRANSFER_ERROR; + async_operation_destroy(result); + result = NULL; + } + else if (transfer_set_settled(transfer, settled) != 0) + { + LogError("Failed setting settled flag"); *link_transfer_error = LINK_TRANSFER_ERROR; async_operation_destroy(result); result = NULL; @@ -1201,9 +1353,9 @@ else { AMQP_VALUE transfer_value = amqpvalue_create_transfer(transfer); - if (transfer_value == NULL) { + LogError("Failed creating transfer performative AMQP value"); *link_transfer_error = LINK_TRANSFER_ERROR; async_operation_destroy(result); result = NULL; @@ -1213,6 +1365,7 @@ DELIVERY_INSTANCE* pending_delivery = GET_ASYNC_OPERATION_CONTEXT(DELIVERY_INSTANCE, result); if (pending_delivery == NULL) { + LogError("Failed getting pending delivery"); *link_transfer_error = LINK_TRANSFER_ERROR; async_operation_destroy(result); result = NULL; @@ -1221,6 +1374,7 @@ { if (tickcounter_get_current_ms(link->tick_counter, &pending_delivery->start_tick) != 0) { + LogError("Failed getting current tick"); *link_transfer_error = LINK_TRANSFER_ERROR; async_operation_destroy(result); result = NULL; @@ -1236,6 +1390,7 @@ if (delivery_instance_list_item == NULL) { + LogError("Failed adding delivery to list"); *link_transfer_error = LINK_TRANSFER_ERROR; async_operation_destroy(result); result = NULL; @@ -1247,7 +1402,12 @@ { default: case SESSION_SEND_TRANSFER_ERROR: - singlylinkedlist_remove(link->pending_deliveries, delivery_instance_list_item); + LogError("Failed session send transfer"); + if (singlylinkedlist_remove(link->pending_deliveries, delivery_instance_list_item) != 0) + { + LogError("Error removing pending delivery from the list"); + } + *link_transfer_error = LINK_TRANSFER_ERROR; async_operation_destroy(result); result = NULL; @@ -1255,7 +1415,12 @@ case SESSION_SEND_TRANSFER_BUSY: /* Ensure we remove from list again since sender will attempt to transfer again on flow on */ - singlylinkedlist_remove(link->pending_deliveries, delivery_instance_list_item); + LogError("Failed session send transfer"); + if (singlylinkedlist_remove(link->pending_deliveries, delivery_instance_list_item) != 0) + { + LogError("Error removing pending delivery from the list"); + } + *link_transfer_error = LINK_TRANSFER_BUSY; async_operation_destroy(result); result = NULL; @@ -1263,7 +1428,7 @@ case SESSION_SEND_TRANSFER_OK: link->delivery_count = delivery_count; - link->link_credit--; + link->current_link_credit--; break; } } @@ -1289,6 +1454,7 @@ if (link == NULL) { + LogError("NULL link"); result = __FAILURE__; } else @@ -1306,6 +1472,7 @@ if (link == NULL) { + LogError("NULL link"); result = __FAILURE__; } else @@ -1320,6 +1487,7 @@ int link_send_disposition(LINK_HANDLE link, delivery_number message_id, AMQP_VALUE delivery_state) { int result; + if (delivery_state == NULL) { result = 0; @@ -1327,24 +1495,29 @@ else { result = send_disposition(link, message_id, delivery_state); - if ( result != 0) + if (result != 0) { LogError("Cannot send disposition frame"); result = __FAILURE__; } } + return result; } void link_dowork(LINK_HANDLE link) { - if (link != NULL) + if (link == NULL) + { + LogError("NULL link"); + } + else { tickcounter_ms_t current_tick; if (tickcounter_get_current_ms(link->tick_counter, ¤t_tick) != 0) { - /* error */ + LogError("Cannot get tick counter value"); } else { @@ -1364,7 +1537,11 @@ delivery_instance->on_delivery_settled(delivery_instance->callback_context, delivery_instance->delivery_id, LINK_DELIVERY_SETTLE_REASON_TIMEOUT, NULL); } - (void)singlylinkedlist_remove(link->pending_deliveries, item); + if (singlylinkedlist_remove(link->pending_deliveries, item) != 0) + { + LogError("Cannot remove item from list"); + } + async_operation_destroy(delivery_instance_async_operation); }