A small memory footprint AMQP implimentation
Dependents: iothub_client_sample_amqp remote_monitoring simplesample_amqp
Diff: connection.c
- Revision:
- 24:2c59c2d43ebf
- Parent:
- 23:1111ee8bcba4
- Child:
- 25:1101516ee67d
diff -r 1111ee8bcba4 -r 2c59c2d43ebf connection.c --- a/connection.c Fri Apr 21 14:50:32 2017 -0700 +++ b/connection.c Mon May 08 10:50:45 2017 -0700 @@ -79,53 +79,56 @@ } CONNECTION_INSTANCE; /* Codes_SRS_CONNECTION_01_258: [on_connection_state_changed shall be invoked whenever the connection state changes.]*/ -static void connection_set_state(CONNECTION_INSTANCE* connection_instance, CONNECTION_STATE connection_state) +static void connection_set_state(CONNECTION_HANDLE connection, CONNECTION_STATE connection_state) { uint64_t i; - CONNECTION_STATE previous_state = connection_instance->connection_state; - connection_instance->connection_state = connection_state; + CONNECTION_STATE previous_state = connection->connection_state; + connection->connection_state = connection_state; /* Codes_SRS_CONNECTION_22_001: [If a connection state changed occurs and a callback is registered the callback shall be called.] */ - if (connection_instance->on_connection_state_changed) + if (connection->on_connection_state_changed) { - connection_instance->on_connection_state_changed(connection_instance->on_connection_state_changed_callback_context, connection_state, previous_state); + connection->on_connection_state_changed(connection->on_connection_state_changed_callback_context, connection_state, previous_state); } /* Codes_SRS_CONNECTION_01_260: [Each endpoint's on_connection_state_changed shall be called.] */ - for (i = 0; i < connection_instance->endpoint_count; i++) + for (i = 0; i < connection->endpoint_count; i++) { /* Codes_SRS_CONNECTION_01_259: [The callback_context passed in connection_create_endpoint.] */ - connection_instance->endpoints[i]->on_connection_state_changed(connection_instance->endpoints[i]->callback_context, connection_state, previous_state); + connection->endpoints[i]->on_connection_state_changed(connection->endpoints[i]->callback_context, connection_state, previous_state); } } -static int send_header(CONNECTION_INSTANCE* connection_instance) +static int send_header(CONNECTION_HANDLE connection) { int result; /* Codes_SRS_CONNECTION_01_093: [_ When the client opens a new socket connection to a server, it MUST send a protocol header with the client's preferred protocol version.] */ /* Codes_SRS_CONNECTION_01_104: [Sending the protocol header shall be done by using xio_send.] */ - if (xio_send(connection_instance->io, amqp_header, sizeof(amqp_header), NULL, NULL) != 0) + if (xio_send(connection->io, amqp_header, sizeof(amqp_header), NULL, NULL) != 0) { /* Codes_SRS_CONNECTION_01_106: [When sending the protocol header fails, the connection shall be immediately closed.] */ - xio_close(connection_instance->io, NULL, NULL); + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } /* Codes_SRS_CONNECTION_01_057: [END In this state it is illegal for either endpoint to write anything more onto the connection. The connection can be safely closed and discarded.] */ - connection_set_state(connection_instance, CONNECTION_STATE_END); + connection_set_state(connection, CONNECTION_STATE_END); /* Codes_SRS_CONNECTION_01_105: [When xio_send fails, connection_dowork shall return a non-zero value.] */ result = __FAILURE__; } else { - if (connection_instance->is_trace_on == 1) + if (connection->is_trace_on == 1) { LOG(AZ_LOG_TRACE, LOG_LINE, "-> Header (AMQP 0.1.0.0)"); } /* Codes_SRS_CONNECTION_01_041: [HDR SENT In this state the connection header has been sent to the peer but no connection header has been received.] */ - connection_set_state(connection_instance, CONNECTION_STATE_HDR_SENT); + connection_set_state(connection, CONNECTION_STATE_HDR_SENT); result = 0; } @@ -186,8 +189,12 @@ UNUSED(performative); #else AMQP_VALUE descriptor = amqpvalue_get_inplace_descriptor(performative); - if (descriptor != NULL) - { + if (descriptor == NULL) + { + LogError("Error getting performative descriptor"); + } + else + { LOG(AZ_LOG_TRACE, 0, "<- "); LOG(AZ_LOG_TRACE, 0, (char*)get_frame_type_as_string(descriptor)); char* performative_as_string = NULL; @@ -206,9 +213,13 @@ UNUSED(performative); #else AMQP_VALUE descriptor = amqpvalue_get_inplace_descriptor(performative); - if (descriptor != NULL) - { - LOG(AZ_LOG_TRACE, 0, "-> "); + if (descriptor == NULL) + { + LogError("Error getting performative descriptor"); + } + else + { + LOG(AZ_LOG_TRACE, 0, "-> "); LOG(AZ_LOG_TRACE, 0, (char*)get_frame_type_as_string(descriptor)); char* performative_as_string = NULL; LOG(AZ_LOG_TRACE, LOG_LINE, (performative_as_string = amqpvalue_to_string(performative))); @@ -222,73 +233,113 @@ static void on_bytes_encoded(void* context, const unsigned char* bytes, size_t length, bool encode_complete) { - CONNECTION_INSTANCE* connection_instance = (CONNECTION_INSTANCE*)context; - if (xio_send(connection_instance->io, bytes, length, encode_complete ? connection_instance->on_send_complete : NULL, connection_instance->on_send_complete_callback_context) != 0) + CONNECTION_HANDLE connection = (CONNECTION_HANDLE)context; + if (xio_send(connection->io, bytes, length, encode_complete ? connection->on_send_complete : NULL, connection->on_send_complete_callback_context) != 0) { - xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + LogError("Cannot send encoded bytes"); + + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); } } -static int send_open_frame(CONNECTION_INSTANCE* connection_instance) +static int send_open_frame(CONNECTION_HANDLE connection) { int result; /* Codes_SRS_CONNECTION_01_151: [The connection max_frame_size setting shall be passed down to the frame_codec when the Open frame is sent.] */ - if (frame_codec_set_max_frame_size(connection_instance->frame_codec, connection_instance->max_frame_size) != 0) + if (frame_codec_set_max_frame_size(connection->frame_codec, connection->max_frame_size) != 0) { + LogError("Cannot set max frame size"); + /* Codes_SRS_CONNECTION_01_207: [If frame_codec_set_max_frame_size fails the connection shall be closed and the state set to END.] */ - xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } else { /* Codes_SRS_CONNECTION_01_134: [The container id field shall be filled with the container id specified in connection_create.] */ - OPEN_HANDLE open_performative = open_create(connection_instance->container_id); + OPEN_HANDLE open_performative = open_create(connection->container_id); if (open_performative == NULL) { - /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ - xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + LogError("Cannot create OPEN performative"); + + /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } else { /* Codes_SRS_CONNECTION_01_137: [The max_frame_size connection setting shall be set in the open frame by using open_set_max_frame_size.] */ - if (open_set_max_frame_size(open_performative, connection_instance->max_frame_size) != 0) + if (open_set_max_frame_size(open_performative, connection->max_frame_size) != 0) { - /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ - xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + LogError("Cannot set max frame size"); + + /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } /* Codes_SRS_CONNECTION_01_139: [The channel_max connection setting shall be set in the open frame by using open_set_channel_max.] */ - else if (open_set_channel_max(open_performative, connection_instance->channel_max) != 0) + else if (open_set_channel_max(open_performative, connection->channel_max) != 0) { - /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ - xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + LogError("Cannot set max channel"); + + /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } /* Codes_SRS_CONNECTION_01_142: [If no idle_timeout value has been specified, no value shall be stamped in the open frame (no call to open_set_idle_time_out shall be made).] */ - else if ((connection_instance->idle_timeout_specified) && + else if ((connection->idle_timeout_specified) && /* Codes_SRS_CONNECTION_01_141: [If idle_timeout has been specified by a call to connection_set_idle_timeout, then that value shall be stamped in the open frame.] */ - (open_set_idle_time_out(open_performative, connection_instance->idle_timeout) != 0)) + (open_set_idle_time_out(open_performative, connection->idle_timeout) != 0)) { /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ - xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } /* Codes_SRS_CONNECTION_01_136: [If no hostname value has been specified, no value shall be stamped in the open frame (no call to open_set_hostname shall be made).] */ - else if ((connection_instance->host_name != NULL) && + else if ((connection->host_name != NULL) && /* Codes_SRS_CONNECTION_01_135: [If hostname has been specified by a call to connection_set_hostname, then that value shall be stamped in the open frame.] */ - (open_set_hostname(open_performative, connection_instance->host_name) != 0)) + (open_set_hostname(open_performative, connection->host_name) != 0)) { - /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ - xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + LogError("Cannot set hostname"); + + /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } else @@ -296,9 +347,15 @@ AMQP_VALUE open_performative_value = amqpvalue_create_open(open_performative); if (open_performative_value == NULL) { - /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ - xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + LogError("Cannot create OPEN AMQP value"); + + /* Codes_SRS_CONNECTION_01_208: [If the open frame cannot be constructed, the connection shall be closed and set to the END state.] */ + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } else @@ -308,24 +365,30 @@ /* Codes_SRS_CONNECTION_01_005: [The open frame describes the capabilities and limits of that peer.] */ /* Codes_SRS_CONNECTION_01_205: [Sending the AMQP OPEN frame shall be done by calling amqp_frame_codec_begin_encode_frame with channel number 0, the actual performative payload and 0 as payload_size.] */ /* Codes_SRS_CONNECTION_01_006: [The open frame can only be sent on channel 0.] */ - connection_instance->on_send_complete = NULL; - connection_instance->on_send_complete_callback_context = NULL; - if (amqp_frame_codec_encode_frame(connection_instance->amqp_frame_codec, 0, open_performative_value, NULL, 0, on_bytes_encoded, connection_instance) != 0) + connection->on_send_complete = NULL; + connection->on_send_complete_callback_context = NULL; + if (amqp_frame_codec_encode_frame(connection->amqp_frame_codec, 0, open_performative_value, NULL, 0, on_bytes_encoded, connection) != 0) { - /* Codes_SRS_CONNECTION_01_206: [If sending the frame fails, the connection shall be closed and state set to END.] */ - xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + LogError("amqp_frame_codec_encode_frame failed"); + + /* Codes_SRS_CONNECTION_01_206: [If sending the frame fails, the connection shall be closed and state set to END.] */ + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } else { - if (connection_instance->is_trace_on == 1) + if (connection->is_trace_on == 1) { log_outgoing_frame(open_performative_value); } /* Codes_SRS_CONNECTION_01_046: [OPEN SENT In this state the connection headers have been exchanged. An open frame has been sent to the peer but no open frame has yet been received.] */ - connection_set_state(connection_instance, CONNECTION_STATE_OPEN_SENT); + connection_set_state(connection, CONNECTION_STATE_OPEN_SENT); result = 0; } @@ -340,7 +403,7 @@ return result; } -static int send_close_frame(CONNECTION_INSTANCE* connection_instance, ERROR_HANDLE error_handle) +static int send_close_frame(CONNECTION_HANDLE connection, ERROR_HANDLE error_handle) { int result; CLOSE_HANDLE close_performative; @@ -349,7 +412,8 @@ close_performative = close_create(); if (close_performative == NULL) { - result = __FAILURE__; + LogError("Cannot create close performative"); + result = __FAILURE__; } else { @@ -357,28 +421,31 @@ /* Codes_SRS_CONNECTION_01_238: [If set, this field indicates that the connection is being closed due to an error condition.] */ (close_set_error(close_performative, error_handle) != 0)) { - result = __FAILURE__; + LogError("Cannot set error on CLOSE"); + result = __FAILURE__; } else { AMQP_VALUE close_performative_value = amqpvalue_create_close(close_performative); if (close_performative_value == NULL) { - result = __FAILURE__; + LogError("Cannot create AMQP CLOSE performative value"); + result = __FAILURE__; } else { /* Codes_SRS_CONNECTION_01_215: [Sending the AMQP CLOSE frame shall be done by calling amqp_frame_codec_begin_encode_frame with channel number 0, the actual performative payload and 0 as payload_size.] */ /* Codes_SRS_CONNECTION_01_013: [However, implementations SHOULD send it on channel 0] */ - connection_instance->on_send_complete = NULL; - connection_instance->on_send_complete_callback_context = NULL; - if (amqp_frame_codec_encode_frame(connection_instance->amqp_frame_codec, 0, close_performative_value, NULL, 0, on_bytes_encoded, connection_instance) != 0) + connection->on_send_complete = NULL; + connection->on_send_complete_callback_context = NULL; + if (amqp_frame_codec_encode_frame(connection->amqp_frame_codec, 0, close_performative_value, NULL, 0, on_bytes_encoded, connection) != 0) { - result = __FAILURE__; + LogError("amqp_frame_codec_encode_frame failed"); + result = __FAILURE__; } else { - if (connection_instance->is_trace_on == 1) + if (connection->is_trace_on == 1) { log_outgoing_frame(close_performative_value); } @@ -396,38 +463,59 @@ return result; } -static void close_connection_with_error(CONNECTION_INSTANCE* connection_instance, const char* condition_value, const char* description) +static void close_connection_with_error(CONNECTION_HANDLE connection, const char* condition_value, const char* description) { ERROR_HANDLE error_handle = error_create(condition_value); if (error_handle == NULL) { /* Codes_SRS_CONNECTION_01_214: [If the close frame cannot be constructed or sent, the connection shall be closed and set to the END state.] */ - (void)xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); } else { /* Codes_SRS_CONNECTION_01_219: [The error description shall be set to an implementation defined string.] */ - if ((error_set_description(error_handle, description) != 0) || - (send_close_frame(connection_instance, error_handle) != 0)) - { - /* Codes_SRS_CONNECTION_01_214: [If the close frame cannot be constructed or sent, the connection shall be closed and set to the END state.] */ - (void)xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + if (error_set_description(error_handle, description) != 0) + { + LogError("Cannot set error description on CLOSE frame"); + + /* Codes_SRS_CONNECTION_01_214: [If the close frame cannot be constructed or sent, the connection shall be closed and set to the END state.] */ + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); + } + else if (send_close_frame(connection, error_handle) != 0) + { + LogError("Cannot send CLOSE frame"); + + /* Codes_SRS_CONNECTION_01_214: [If the close frame cannot be constructed or sent, the connection shall be closed and set to the END state.] */ + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); } else { /* Codes_SRS_CONNECTION_01_213: [When passing the bytes to frame_codec fails, a CLOSE frame shall be sent and the state shall be set to DISCARDING.] */ /* Codes_SRS_CONNECTION_01_055: [DISCARDING The DISCARDING state is a variant of the CLOSE SENT state where the close is triggered by an error.] */ /* Codes_SRS_CONNECTION_01_010: [After writing this frame the peer SHOULD continue to read from the connection until it receives the partner's close frame ] */ - connection_set_state(connection_instance, CONNECTION_STATE_DISCARDING); + connection_set_state(connection, CONNECTION_STATE_DISCARDING); } error_destroy(error_handle); } } -static ENDPOINT_INSTANCE* find_session_endpoint_by_outgoing_channel(CONNECTION_INSTANCE* connection, uint16_t outgoing_channel) +static ENDPOINT_INSTANCE* find_session_endpoint_by_outgoing_channel(CONNECTION_HANDLE connection, uint16_t outgoing_channel) { uint32_t i; ENDPOINT_INSTANCE* result; @@ -442,7 +530,8 @@ if (i == connection->endpoint_count) { - result = NULL; + LogError("Cannot find session endpoint for channel %u", (unsigned int)outgoing_channel); + result = NULL; } else { @@ -452,7 +541,7 @@ return result; } -static ENDPOINT_INSTANCE* find_session_endpoint_by_incoming_channel(CONNECTION_INSTANCE* connection, uint16_t incoming_channel) +static ENDPOINT_INSTANCE* find_session_endpoint_by_incoming_channel(CONNECTION_HANDLE connection, uint16_t incoming_channel) { uint32_t i; ENDPOINT_INSTANCE* result; @@ -467,7 +556,8 @@ if (i == connection->endpoint_count) { - result = NULL; + LogError("Cannot find session endpoint for channel %u", (unsigned int)incoming_channel); + result = NULL; } else { @@ -477,14 +567,15 @@ return result; } -static int connection_byte_received(CONNECTION_INSTANCE* connection_instance, unsigned char b) +static int connection_byte_received(CONNECTION_HANDLE connection, unsigned char b) { int result; - switch (connection_instance->connection_state) + switch (connection->connection_state) { default: - result = __FAILURE__; + LogError("Unknown connection state: %d", (int)connection->connection_state); + result = __FAILURE__; break; /* Codes_SRS_CONNECTION_01_039: [START In this state a connection exists, but nothing has been sent or received. This is the state an implementation would be in immediately after performing a socket connect or socket accept.] */ @@ -492,28 +583,33 @@ /* Codes_SRS_CONNECTION_01_041: [HDR SENT In this state the connection header has been sent to the peer but no connection header has been received.] */ case CONNECTION_STATE_HDR_SENT: - if (b != amqp_header[connection_instance->header_bytes_received]) + if (b != amqp_header[connection->header_bytes_received]) { /* Codes_SRS_CONNECTION_01_089: [If the incoming and outgoing protocol headers do not match, both peers MUST close their outgoing stream] */ - xio_close(connection_instance->io, NULL, NULL); - connection_set_state(connection_instance, CONNECTION_STATE_END); + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } else { - connection_instance->header_bytes_received++; - if (connection_instance->header_bytes_received == sizeof(amqp_header)) + connection->header_bytes_received++; + if (connection->header_bytes_received == sizeof(amqp_header)) { - if (connection_instance->is_trace_on == 1) + if (connection->is_trace_on == 1) { LOG(AZ_LOG_TRACE, LOG_LINE, "<- Header (AMQP 0.1.0.0)"); } - connection_set_state(connection_instance, CONNECTION_STATE_HDR_EXCH); + connection_set_state(connection, CONNECTION_STATE_HDR_EXCH); - if (send_open_frame(connection_instance) != 0) + if (send_open_frame(connection) != 0) { - connection_set_state(connection_instance, CONNECTION_STATE_END); + LogError("Cannot send open frame"); + connection_set_state(connection, CONNECTION_STATE_END); } } @@ -537,11 +633,12 @@ /* Codes_SRS_CONNECTION_01_048: [OPENED In this state the connection header and the open frame have been both sent and received.] */ case CONNECTION_STATE_OPENED: /* Codes_SRS_CONNECTION_01_212: [After the initial handshake has been done all bytes received from the io instance shall be passed to the frame_codec for decoding by calling frame_codec_receive_bytes.] */ - if (frame_codec_receive_bytes(connection_instance->frame_codec, &b, 1) != 0) + if (frame_codec_receive_bytes(connection->frame_codec, &b, 1) != 0) { - /* Codes_SRS_CONNECTION_01_218: [The error amqp:internal-error shall be set in the error.condition field of the CLOSE frame.] */ + LogError("Cannot process received bytes"); + /* Codes_SRS_CONNECTION_01_218: [The error amqp:internal-error shall be set in the error.condition field of the CLOSE frame.] */ /* Codes_SRS_CONNECTION_01_219: [The error description shall be set to an implementation defined string.] */ - close_connection_with_error(connection_instance, "amqp:internal-error", "connection_byte_received::frame_codec_receive_bytes failed"); + close_connection_with_error(connection, "amqp:internal-error", "connection_byte_received::frame_codec_receive_bytes failed"); result = __FAILURE__; } else @@ -561,29 +658,34 @@ for (i = 0; i < size; i++) { - if (connection_byte_received((CONNECTION_INSTANCE*)context, buffer[i]) != 0) + if (connection_byte_received((CONNECTION_HANDLE)context, buffer[i]) != 0) { - break; + LogError("Cannot process received bytes"); + break; } } } static void connection_on_io_open_complete(void* context, IO_OPEN_RESULT io_open_result) { - CONNECTION_INSTANCE* connection_instance = (CONNECTION_INSTANCE*)context; + CONNECTION_HANDLE connection = (CONNECTION_HANDLE)context; if (io_open_result == IO_OPEN_OK) { /* Codes_SRS_CONNECTION_01_084: [The connection_instance state machine implementing the protocol requirements shall be run as part of connection_dowork.] */ - switch (connection_instance->connection_state) + switch (connection->connection_state) { default: - break; + LogError("Unknown connection state: %d", (int)connection->connection_state); + break; case CONNECTION_STATE_START: /* Codes_SRS_CONNECTION_01_086: [Prior to sending any frames on a connection_instance, each peer MUST start by sending a protocol header that indicates the protocol version used on the connection_instance.] */ /* Codes_SRS_CONNECTION_01_091: [The AMQP peer which acted in the role of the TCP client (i.e. the peer that actively opened the connection_instance) MUST immediately send its outgoing protocol header on establishment of the TCP connection_instance.] */ - (void)send_header(connection_instance); + if (send_header(connection) != 0) + { + LogError("Cannot send header"); + } break; case CONNECTION_STATE_HDR_SENT: @@ -595,9 +697,10 @@ /* Codes_SRS_CONNECTION_01_002: [Each AMQP connection_instance begins with an exchange of capabilities and limitations, including the maximum frame size.] */ /* Codes_SRS_CONNECTION_01_004: [After establishing or accepting a TCP connection_instance and sending the protocol header, each peer MUST send an open frame before sending any other frames.] */ /* Codes_SRS_CONNECTION_01_005: [The open frame describes the capabilities and limits of that peer.] */ - if (send_open_frame(connection_instance) != 0) + if (send_open_frame(connection) != 0) { - connection_set_state(connection_instance, CONNECTION_STATE_END); + LogError("Cannot send OPEN frame"); + connection_set_state(connection, CONNECTION_STATE_END); } break; @@ -607,70 +710,77 @@ } else { - connection_set_state(connection_instance, CONNECTION_STATE_END); + connection_set_state(connection, CONNECTION_STATE_END); } } static void connection_on_io_error(void* context) { - CONNECTION_INSTANCE* connection_instance = (CONNECTION_INSTANCE*)context; + CONNECTION_HANDLE connection = (CONNECTION_HANDLE)context; /* Codes_SRS_CONNECTION_22_005: [If the io notifies the connection instance of an IO_STATE_ERROR state and an io error callback is registered, the connection shall call the registered callback.] */ - if (connection_instance->on_io_error) + if (connection->on_io_error) { - connection_instance->on_io_error(connection_instance->on_io_error_callback_context); + connection->on_io_error(connection->on_io_error_callback_context); } - if (connection_instance->connection_state != CONNECTION_STATE_END) + if (connection->connection_state != CONNECTION_STATE_END) { /* Codes_SRS_CONNECTION_01_202: [If the io notifies the connection instance of an IO_STATE_ERROR state the connection shall be closed and the state set to END.] */ - connection_set_state(connection_instance, CONNECTION_STATE_ERROR); - (void)xio_close(connection_instance->io, NULL, NULL); - } + connection_set_state(connection, CONNECTION_STATE_ERROR); + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + } } static void on_empty_amqp_frame_received(void* context, uint16_t channel) { - (void)channel; - /* It does not matter on which channel we received the frame */ - CONNECTION_INSTANCE* connection_instance = (CONNECTION_INSTANCE*)context; - if (connection_instance->is_trace_on == 1) + CONNECTION_HANDLE connection = (CONNECTION_HANDLE)context; + /* It does not matter on which channel we received the frame */ + (void)channel; + + if (connection->is_trace_on == 1) { LOG(AZ_LOG_TRACE, LOG_LINE, "<- Empty frame"); } - if (tickcounter_get_current_ms(connection_instance->tick_counter, &connection_instance->last_frame_received_time) != 0) + if (tickcounter_get_current_ms(connection->tick_counter, &connection->last_frame_received_time) != 0) { - /* error */ - } + LogError("Cannot get tickcounter value"); + } } static void on_amqp_frame_received(void* context, uint16_t channel, AMQP_VALUE performative, const unsigned char* payload_bytes, uint32_t payload_size) { - (void)channel; - CONNECTION_INSTANCE* connection_instance = (CONNECTION_INSTANCE*)context; + CONNECTION_HANDLE connection = (CONNECTION_HANDLE)context; + + (void)channel; - if (tickcounter_get_current_ms(connection_instance->tick_counter, &connection_instance->last_frame_received_time) != 0) + if (tickcounter_get_current_ms(connection->tick_counter, &connection->last_frame_received_time) != 0) { - close_connection_with_error(connection_instance, "amqp:internal-error", "cannot get current tick count"); + LogError("Cannot get tickcounter value"); + close_connection_with_error(connection, "amqp:internal-error", "cannot get current tick count"); } else { - if (connection_instance->is_underlying_io_open) + if (connection->is_underlying_io_open) { - switch (connection_instance->connection_state) + switch (connection->connection_state) { default: if (performative == NULL) { - /* Codes_SRS_CONNECTION_01_223: [If the on_endpoint_frame_received is called with a NULL performative then the connection shall be closed with the error condition amqp:internal-error and an implementation defined error description.] */ - close_connection_with_error(connection_instance, "amqp:internal-error", "connection_endpoint_frame_received::NULL performative"); - } + /* Codes_SRS_CONNECTION_01_223: [If the on_endpoint_frame_received is called with a NULL performative then the connection shall be closed with the error condition amqp:internal-error and an implementation defined error description.] */ + close_connection_with_error(connection, "amqp:internal-error", "connection_endpoint_frame_received::NULL performative"); + LogError("connection_endpoint_frame_received::NULL performative"); + } else { AMQP_VALUE descriptor = amqpvalue_get_inplace_descriptor(performative); uint64_t performative_ulong; - if (connection_instance->is_trace_on == 1) + if (connection->is_trace_on == 1) { log_incoming_frame(performative); } @@ -679,52 +789,56 @@ { if (channel != 0) { - /* Codes_SRS_CONNECTION_01_006: [The open frame can only be sent on channel 0.] */ + /* Codes_SRS_CONNECTION_01_006: [The open frame can only be sent on channel 0.] */ /* Codes_SRS_CONNECTION_01_222: [If an Open frame is received in a manner violating the ISO specification, the connection shall be closed with condition amqp:not-allowed and description being an implementation defined string.] */ - close_connection_with_error(connection_instance, "amqp:not-allowed", "OPEN frame received on a channel that is not 0"); - } + close_connection_with_error(connection, "amqp:not-allowed", "OPEN frame received on a channel that is not 0"); + LogError("OPEN frame received on a channel that is not 0"); + } - if (connection_instance->connection_state == CONNECTION_STATE_OPENED) + if (connection->connection_state == CONNECTION_STATE_OPENED) { /* Codes_SRS_CONNECTION_01_239: [If an Open frame is received in the Opened state the connection shall be closed with condition amqp:illegal-state and description being an implementation defined string.] */ - close_connection_with_error(connection_instance, "amqp:illegal-state", "OPEN frame received in the OPENED state"); - } - else if ((connection_instance->connection_state == CONNECTION_STATE_OPEN_SENT) || - (connection_instance->connection_state == CONNECTION_STATE_HDR_EXCH)) + close_connection_with_error(connection, "amqp:illegal-state", "OPEN frame received in the OPENED state"); + LogError("OPEN frame received in the OPENED state"); + } + else if ((connection->connection_state == CONNECTION_STATE_OPEN_SENT) || + (connection->connection_state == CONNECTION_STATE_HDR_EXCH)) { OPEN_HANDLE open_handle; if (amqpvalue_get_open(performative, &open_handle) != 0) { /* Codes_SRS_CONNECTION_01_143: [If any of the values in the received open frame are invalid then the connection shall be closed.] */ /* Codes_SRS_CONNECTION_01_220: [The error amqp:invalid-field shall be set in the error.condition field of the CLOSE frame.] */ - close_connection_with_error(connection_instance, "amqp:invalid-field", "connection_endpoint_frame_received::failed parsing OPEN frame"); - } + close_connection_with_error(connection, "amqp:invalid-field", "connection_endpoint_frame_received::failed parsing OPEN frame"); + LogError("connection_endpoint_frame_received::failed parsing OPEN frame"); + } else { - (void)open_get_idle_time_out(open_handle, &connection_instance->remote_idle_timeout); - if ((open_get_max_frame_size(open_handle, &connection_instance->remote_max_frame_size) != 0) || + (void)open_get_idle_time_out(open_handle, &connection->remote_idle_timeout); + if ((open_get_max_frame_size(open_handle, &connection->remote_max_frame_size) != 0) || /* Codes_SRS_CONNECTION_01_167: [Both peers MUST accept frames of up to 512 (MIN-MAX-FRAME-SIZE) octets.] */ - (connection_instance->remote_max_frame_size < 512)) + (connection->remote_max_frame_size < 512)) { /* Codes_SRS_CONNECTION_01_143: [If any of the values in the received open frame are invalid then the connection shall be closed.] */ /* Codes_SRS_CONNECTION_01_220: [The error amqp:invalid-field shall be set in the error.condition field of the CLOSE frame.] */ - close_connection_with_error(connection_instance, "amqp:invalid-field", "connection_endpoint_frame_received::failed parsing OPEN frame"); - } + close_connection_with_error(connection, "amqp:invalid-field", "connection_endpoint_frame_received::failed parsing OPEN frame"); + LogError("connection_endpoint_frame_received::failed parsing OPEN frame"); + } else { - if (connection_instance->connection_state == CONNECTION_STATE_OPEN_SENT) + if (connection->connection_state == CONNECTION_STATE_OPEN_SENT) { - connection_set_state(connection_instance, CONNECTION_STATE_OPENED); + connection_set_state(connection, CONNECTION_STATE_OPENED); } else { - if (send_open_frame(connection_instance) != 0) + if (send_open_frame(connection) != 0) { - connection_set_state(connection_instance, CONNECTION_STATE_END); + connection_set_state(connection, CONNECTION_STATE_END); } else { - connection_set_state(connection_instance, CONNECTION_STATE_OPENED); + connection_set_state(connection, CONNECTION_STATE_OPENED); } } } @@ -743,44 +857,56 @@ /* Codes_SRS_CONNECTION_01_242: [The connection module shall accept CLOSE frames even if they have extra payload bytes besides the Close performative.] */ /* Codes_SRS_CONNECTION_01_225: [HDR_RCVD HDR OPEN] */ - if ((connection_instance->connection_state == CONNECTION_STATE_HDR_RCVD) || + if ((connection->connection_state == CONNECTION_STATE_HDR_RCVD) || /* Codes_SRS_CONNECTION_01_227: [HDR_EXCH OPEN OPEN] */ - (connection_instance->connection_state == CONNECTION_STATE_HDR_EXCH) || + (connection->connection_state == CONNECTION_STATE_HDR_EXCH) || /* Codes_SRS_CONNECTION_01_228: [OPEN_RCVD OPEN *] */ - (connection_instance->connection_state == CONNECTION_STATE_OPEN_RCVD) || + (connection->connection_state == CONNECTION_STATE_OPEN_RCVD) || /* Codes_SRS_CONNECTION_01_235: [CLOSE_SENT - * TCP Close for Write] */ - (connection_instance->connection_state == CONNECTION_STATE_CLOSE_SENT) || + (connection->connection_state == CONNECTION_STATE_CLOSE_SENT) || /* Codes_SRS_CONNECTION_01_236: [DISCARDING - * TCP Close for Write] */ - (connection_instance->connection_state == CONNECTION_STATE_DISCARDING)) + (connection->connection_state == CONNECTION_STATE_DISCARDING)) { - xio_close(connection_instance->io, NULL, NULL); - } + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + } else { CLOSE_HANDLE close_handle; /* Codes_SRS_CONNECTION_01_012: [A close frame MAY be received on any channel up to the maximum channel number negotiated in open.] */ - if (channel > connection_instance->channel_max) + if (channel > connection->channel_max) { - close_connection_with_error(connection_instance, "amqp:invalid-field", "connection_endpoint_frame_received::failed parsing CLOSE frame"); - } + close_connection_with_error(connection, "amqp:invalid-field", "connection_endpoint_frame_received::failed parsing CLOSE frame"); + LogError("connection_endpoint_frame_received::failed parsing CLOSE frame"); + } else { if (amqpvalue_get_close(performative, &close_handle) != 0) { - close_connection_with_error(connection_instance, "amqp:invalid-field", "connection_endpoint_frame_received::failed parsing CLOSE frame"); - } + close_connection_with_error(connection, "amqp:invalid-field", "connection_endpoint_frame_received::failed parsing CLOSE frame"); + LogError("connection_endpoint_frame_received::failed parsing CLOSE frame"); + } else { close_destroy(close_handle); - connection_set_state(connection_instance, CONNECTION_STATE_CLOSE_RCVD); + connection_set_state(connection, CONNECTION_STATE_CLOSE_RCVD); - (void)send_close_frame(connection_instance, NULL); + if (send_close_frame(connection, NULL) != 0) + { + LogError("Cannot send CLOSE frame"); + } + /* Codes_SRS_CONNECTION_01_214: [If the close frame cannot be constructed or sent, the connection shall be closed and set to the END state.] */ - (void)xio_close(connection_instance->io, NULL, NULL); + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } - connection_set_state(connection_instance, CONNECTION_STATE_END); + connection_set_state(connection, CONNECTION_STATE_END); } } } @@ -792,18 +918,17 @@ switch (performative_ulong) { default: - LOG(AZ_LOG_ERROR, LOG_LINE, "Bad performative: %02x", performative); + LogError("Bad performative: %02x", performative); break; case AMQP_BEGIN: { BEGIN_HANDLE begin; - amqpvalue_get_begin(performative, &begin); - if (begin == NULL) + if (amqpvalue_get_begin(performative, &begin) != 0) { - /* error */ - } + LogError("Cannot get begin performative"); + } else { uint16_t remote_channel; @@ -812,11 +937,11 @@ if (begin_get_remote_channel(begin, &remote_channel) != 0) { - remote_begin = true; - if (connection_instance->on_new_endpoint != NULL) + remote_begin = true; + if (connection->on_new_endpoint != NULL) { - new_endpoint = connection_create_endpoint(connection_instance); - if (!connection_instance->on_new_endpoint(connection_instance->on_new_endpoint_callback_context, new_endpoint)) + new_endpoint = connection_create_endpoint(connection); + if (!connection->on_new_endpoint(connection->on_new_endpoint_callback_context, new_endpoint)) { connection_destroy_endpoint(new_endpoint); new_endpoint = NULL; @@ -826,11 +951,11 @@ if (!remote_begin) { - ENDPOINT_INSTANCE* session_endpoint = find_session_endpoint_by_outgoing_channel(connection_instance, remote_channel); + ENDPOINT_INSTANCE* session_endpoint = find_session_endpoint_by_outgoing_channel(connection, remote_channel); if (session_endpoint == NULL) { - /* error */ - } + LogError("Cannot create session endpoint"); + } else { session_endpoint->incoming_channel = channel; @@ -859,11 +984,11 @@ case AMQP_ATTACH: case AMQP_DETACH: { - ENDPOINT_INSTANCE* session_endpoint = find_session_endpoint_by_incoming_channel(connection_instance, channel); + ENDPOINT_INSTANCE* session_endpoint = find_session_endpoint_by_incoming_channel(connection, channel); if (session_endpoint == NULL) { - /* error */ - } + LogError("Cannot find session endpoint for channel %u", (unsigned int)channel); + } else { session_endpoint->on_endpoint_frame_received(session_endpoint->callback_context, performative, payload_size, payload_bytes); @@ -888,8 +1013,11 @@ /* Codes_SRS_CONNECTION_01_234: [CLOSE_RCVD * - TCP Close for Read] */ case CONNECTION_STATE_END: /* Codes_SRS_CONNECTION_01_237: [END - - TCP Close] */ - xio_close(connection_instance->io, NULL, NULL); - break; + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + break; } } } @@ -899,14 +1027,16 @@ { /* Bug: some error handling should happen here Filed: uAMQP: frame_codec error and amqp_frame_codec_error should handle the errors */ - (void)context; + LogError("A frame_codec_error occured"); + (void)context; } static void amqp_frame_codec_error(void* context) { /* Bug: some error handling should happen here Filed: uAMQP: frame_codec error and amqp_frame_codec_error should handle the errors */ - (void)context; + LogError("An amqp_frame_codec_error occured"); + (void)context; } /* Codes_SRS_CONNECTION_01_001: [connection_create shall open a new connection to a specified host/port.] */ @@ -919,20 +1049,26 @@ /* Codes_SRS_CONNECTION_22_002: [connection_create shall allow registering connections state and io error callbacks.] */ CONNECTION_HANDLE connection_create2(XIO_HANDLE xio, const char* hostname, const char* container_id, ON_NEW_ENDPOINT on_new_endpoint, void* callback_context, ON_CONNECTION_STATE_CHANGED on_connection_state_changed, void* on_connection_state_changed_context, ON_IO_ERROR on_io_error, void* on_io_error_context) { - CONNECTION_INSTANCE* result; + CONNECTION_HANDLE result; if ((xio == NULL) || (container_id == NULL)) { /* Codes_SRS_CONNECTION_01_071: [If xio or container_id is NULL, connection_create shall return NULL.] */ - result = NULL; + LogError("Bad arguments: xio = %p, container_id = %p", + xio, container_id); + result = NULL; } else { - result = (CONNECTION_INSTANCE*)malloc(sizeof(CONNECTION_INSTANCE)); + result = (CONNECTION_HANDLE)malloc(sizeof(CONNECTION_INSTANCE)); /* Codes_SRS_CONNECTION_01_081: [If allocating the memory for the connection fails then connection_create shall return NULL.] */ - if (result != NULL) - { + if (result == NULL) + { + LogError("Cannot allocate memory for connection"); + } + else + { result->io = xio; /* Codes_SRS_CONNECTION_01_082: [connection_create shall allocate a new frame_codec instance to be used for frame encoding/decoding.] */ @@ -940,7 +1076,8 @@ if (result->frame_codec == NULL) { /* Codes_SRS_CONNECTION_01_083: [If frame_codec_create fails then connection_create shall return NULL.] */ - free(result); + LogError("Cannot create frame_codec"); + free(result); result = NULL; } else @@ -949,7 +1086,8 @@ if (result->amqp_frame_codec == NULL) { /* Codes_SRS_CONNECTION_01_108: [If amqp_frame_codec_create fails, connection_create shall return NULL.] */ - frame_codec_destroy(result->frame_codec); + LogError("Cannot create amqp_frame_codec"); + frame_codec_destroy(result->frame_codec); free(result); result = NULL; } @@ -962,7 +1100,8 @@ if (result->host_name == NULL) { /* Codes_SRS_CONNECTION_01_081: [If allocating the memory for the connection fails then connection_create shall return NULL.] */ - amqp_frame_codec_destroy(result->amqp_frame_codec); + LogError("Cannot allocate memory for host name"); + amqp_frame_codec_destroy(result->amqp_frame_codec); frame_codec_destroy(result->frame_codec); free(result); result = NULL; @@ -984,7 +1123,8 @@ if (result->container_id == NULL) { /* Codes_SRS_CONNECTION_01_081: [If allocating the memory for the connection fails then connection_create shall return NULL.] */ - free(result->host_name); + LogError("Cannot allocate memory for container_id"); + free(result->host_name); amqp_frame_codec_destroy(result->amqp_frame_codec); frame_codec_destroy(result->frame_codec); free(result); @@ -995,7 +1135,8 @@ result->tick_counter = tickcounter_create(); if (result->tick_counter == NULL) { - free(result->container_id); + LogError("Cannot create tick counter"); + free(result->container_id); free(result->host_name); amqp_frame_codec_destroy(result->amqp_frame_codec); frame_codec_destroy(result->frame_codec); @@ -1068,12 +1209,16 @@ void connection_destroy(CONNECTION_HANDLE connection) { /* Codes_SRS_CONNECTION_01_079: [If handle is NULL, connection_destroy shall do nothing.] */ - if (connection != NULL) - { + if (connection == NULL) + { + LogError("NULL connection"); + } + else + { /* Codes_SRS_CONNECTION_01_073: [connection_destroy shall free all resources associated with a connection.] */ if (connection->is_underlying_io_open) { - connection_close(connection, NULL, NULL); + (void)connection_close(connection, NULL, NULL); } amqp_frame_codec_destroy(connection->amqp_frame_codec); @@ -1094,7 +1239,8 @@ if (connection == NULL) { - result = __FAILURE__; + LogError("NULL connection"); + result = __FAILURE__; } else { @@ -1102,7 +1248,8 @@ { if (xio_open(connection->io, connection_on_io_open_complete, connection, connection_on_bytes_received, connection, connection_on_io_error, connection) != 0) { - connection_set_state(connection, CONNECTION_STATE_END); + LogError("Opening the underlying IO failed"); + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } else @@ -1129,7 +1276,8 @@ if (connection == NULL) { - result = __FAILURE__; + LogError("NULL connection"); + result = __FAILURE__; } else { @@ -1137,7 +1285,8 @@ { if (xio_open(connection->io, connection_on_io_open_complete, connection, connection_on_bytes_received, connection, connection_on_io_error, connection) != 0) { - connection_set_state(connection, CONNECTION_STATE_END); + LogError("Opening the underlying IO failed"); + connection_set_state(connection, CONNECTION_STATE_END); result = __FAILURE__; } else @@ -1164,7 +1313,8 @@ if (connection == NULL) { - result = __FAILURE__; + LogError("NULL connection"); + result = __FAILURE__; } else { @@ -1174,12 +1324,20 @@ } else { - (void)send_close_frame(connection, NULL); + if (send_close_frame(connection, NULL) != 0) + { + LogError("Sending CLOSE frame failed"); + } + connection_set_state(connection, CONNECTION_STATE_END); } - (void)xio_close(connection->io, NULL, NULL); - connection->is_underlying_io_open = 1; + if (xio_close(connection->io, NULL, NULL) != 0) + { + LogError("xio_close failed"); + } + + connection->is_underlying_io_open = 1; result = 0; } @@ -1192,19 +1350,25 @@ int result; /* Codes_SRS_CONNECTION_01_163: [If connection is NULL, connection_set_max_frame_size shall fail and return a non-zero value.] */ - if ((connection == NULL) || - /* Codes_SRS_CONNECTION_01_150: [If the max_frame_size is invalid then connection_set_max_frame_size shall fail and return a non-zero value.] */ - /* Codes_SRS_CONNECTION_01_167: [Both peers MUST accept frames of up to 512 (MIN-MAX-FRAME-SIZE) octets.] */ - (max_frame_size < 512)) + if (connection == NULL) + { + LogError("NULL connection"); + result = __FAILURE__; + } + /* Codes_SRS_CONNECTION_01_150: [If the max_frame_size is invalid then connection_set_max_frame_size shall fail and return a non-zero value.] */ + /* Codes_SRS_CONNECTION_01_167: [Both peers MUST accept frames of up to 512 (MIN-MAX-FRAME-SIZE) octets.] */ + else if (max_frame_size < 512) { - result = __FAILURE__; + LogError("max_frame_size too small"); + result = __FAILURE__; } else { /* Codes_SRS_CONNECTION_01_157: [If connection_set_max_frame_size is called after the initial Open frame has been sent, it shall fail and return a non-zero value.] */ if (connection->connection_state != CONNECTION_STATE_START) { - result = __FAILURE__; + LogError("Connection already open"); + result = __FAILURE__; } else { @@ -1228,7 +1392,9 @@ if ((connection == NULL) || (max_frame_size == NULL)) { - result = __FAILURE__; + LogError("Bad arguments: connection = %p, max_frame_size = %p", + connection, max_frame_size); + result = __FAILURE__; } else { @@ -1249,14 +1415,16 @@ /* Codes_SRS_CONNECTION_01_181: [If connection is NULL then connection_set_channel_max shall fail and return a non-zero value.] */ if (connection == NULL) { - result = __FAILURE__; + LogError("NULL connection"); + result = __FAILURE__; } else { /* Codes_SRS_CONNECTION_01_156: [If connection_set_channel_max is called after the initial Open frame has been sent, it shall fail and return a non-zero value.] */ if (connection->connection_state != CONNECTION_STATE_START) { - result = __FAILURE__; + LogError("Connection already open"); + result = __FAILURE__; } else { @@ -1280,7 +1448,9 @@ if ((connection == NULL) || (channel_max == NULL)) { - result = __FAILURE__; + LogError("Bad arguments: connection = %p, channel_max = %p", + connection, channel_max); + result = __FAILURE__; } else { @@ -1301,14 +1471,16 @@ /* Codes_SRS_CONNECTION_01_191: [If connection is NULL, connection_set_idle_timeout shall fail and return a non-zero value.] */ if (connection == NULL) { - result = __FAILURE__; + LogError("NULL connection"); + result = __FAILURE__; } else { /* Codes_SRS_CONNECTION_01_158: [If connection_set_idle_timeout is called after the initial Open frame has been sent, it shall fail and return a non-zero value.] */ if (connection->connection_state != CONNECTION_STATE_START) { - result = __FAILURE__; + LogError("Connection already open"); + result = __FAILURE__; } else { @@ -1333,7 +1505,9 @@ if ((connection == NULL) || (idle_timeout == NULL)) { - result = __FAILURE__; + LogError("Bad arguments: connection = %p, idle_timeout = %p", + connection, idle_timeout); + result = __FAILURE__; } else { @@ -1354,7 +1528,9 @@ if ((connection == NULL) || (remote_max_frame_size == NULL)) { - result = __FAILURE__; + LogError("Bad arguments: connection = %p, remote_max_frame_size = %p", + connection, remote_max_frame_size); + result = __FAILURE__; } else { @@ -1371,13 +1547,18 @@ uint64_t local_deadline = (uint64_t )-1; uint64_t remote_deadline = (uint64_t)-1; - if (connection != NULL) - { + if (connection == NULL) + { + LogError("NULL connection"); + } + else + { tickcounter_ms_t current_ms; if (tickcounter_get_current_ms(connection->tick_counter, ¤t_ms) != 0) { - close_connection_with_error(connection, "amqp:internal-error", "Could not get tick count"); + LogError("Could not get tick counter value"); + close_connection_with_error(connection, "amqp:internal-error", "Could not get tick count"); } else { @@ -1415,8 +1596,9 @@ connection->on_send_complete = NULL; if (amqp_frame_codec_encode_empty_frame(connection->amqp_frame_codec, 0, on_bytes_encoded, connection) != 0) { - /* close connection */ - close_connection_with_error(connection, "amqp:internal-error", "Cannot send empty frame"); + LogError("Encoding the empty frame failed"); + /* close connection */ + close_connection_with_error(connection, "amqp:internal-error", "Cannot send empty frame"); } else { @@ -1441,8 +1623,12 @@ void connection_dowork(CONNECTION_HANDLE connection) { /* Codes_SRS_CONNECTION_01_078: [If handle is NULL, connection_dowork shall do nothing.] */ - if (connection != NULL) - { + if (connection == NULL) + { + LogError("NULL connection"); + } + else + { if (connection_handle_deadlines(connection) > 0) { /* Codes_SRS_CONNECTION_01_076: [connection_dowork shall schedule the underlying IO interface to do its work by calling xio_dowork.] */ @@ -1453,13 +1639,14 @@ ENDPOINT_HANDLE connection_create_endpoint(CONNECTION_HANDLE connection) { - ENDPOINT_INSTANCE* result; + ENDPOINT_HANDLE result; /* Codes_SRS_CONNECTION_01_113: [If connection, on_endpoint_frame_received or on_connection_state_changed is NULL, connection_create_endpoint shall fail and return NULL.] */ /* Codes_SRS_CONNECTION_01_193: [The context argument shall be allowed to be NULL.] */ if (connection == NULL) { - result = NULL; + LogError("NULL connection"); + result = NULL; } else { @@ -1483,11 +1670,15 @@ } /* Codes_SRS_CONNECTION_01_127: [On success, connection_create_endpoint shall return a non-NULL handle to the newly created endpoint.] */ - result = (ENDPOINT_INSTANCE*)malloc(sizeof(ENDPOINT_INSTANCE)); + result = (ENDPOINT_HANDLE)malloc(sizeof(ENDPOINT_INSTANCE)); /* Codes_SRS_CONNECTION_01_196: [If memory cannot be allocated for the new endpoint, connection_create_endpoint shall fail and return NULL.] */ - if (result != NULL) - { - ENDPOINT_INSTANCE** new_endpoints; + if (result == NULL) + { + LogError("Cannot allocate memory for endpoint"); + } + else + { + ENDPOINT_HANDLE* new_endpoints; result->on_endpoint_frame_received = NULL; result->on_connection_state_changed = NULL; @@ -1496,11 +1687,12 @@ result->connection = connection; /* Codes_SRS_CONNECTION_01_197: [The newly created endpoint shall be added to the endpoints list, so that it can be tracked.] */ - new_endpoints = (ENDPOINT_INSTANCE**)realloc(connection->endpoints, sizeof(ENDPOINT_INSTANCE*) * (connection->endpoint_count + 1)); + new_endpoints = (ENDPOINT_HANDLE*)realloc(connection->endpoints, sizeof(ENDPOINT_HANDLE) * (connection->endpoint_count + 1)); if (new_endpoints == NULL) { /* Tests_SRS_CONNECTION_01_198: [If adding the endpoint to the endpoints list tracked by the connection fails, connection_create_endpoint shall fail and return NULL.] */ - free(result); + LogError("Cannot reallocate memory for connection endpoints"); + free(result); result = NULL; } else @@ -1532,7 +1724,9 @@ (on_endpoint_frame_received == NULL) || (on_connection_state_changed == NULL)) { - result = __FAILURE__; + LogError("Bad arguments: endpoint = %p, on_endpoint_frame_received = %p, on_connection_state_changed = %p", + endpoint, on_endpoint_frame_received, on_connection_state_changed); + result = __FAILURE__; } else { @@ -1553,7 +1747,9 @@ if ((endpoint == NULL) || (incoming_channel == NULL)) { - result = __FAILURE__; + LogError("Bad arguments: endpoint = %p, incoming_channel = %p", + endpoint, incoming_channel); + result = __FAILURE__; } else { @@ -1567,9 +1763,13 @@ /* Codes_SRS_CONNECTION_01_129: [connection_destroy_endpoint shall free all resources associated with an endpoint created by connection_create_endpoint.] */ void connection_destroy_endpoint(ENDPOINT_HANDLE endpoint) { - if (endpoint != NULL) - { - CONNECTION_INSTANCE* connection = (CONNECTION_INSTANCE*)endpoint->connection; + if (endpoint == NULL) + { + LogError("NULL endpoint"); + } + else + { + CONNECTION_HANDLE connection = (CONNECTION_HANDLE)endpoint->connection; size_t i; for (i = 0; i < connection->endpoint_count; i++) @@ -1582,11 +1782,11 @@ /* Codes_SRS_CONNECTION_01_130: [The outgoing channel associated with the endpoint shall be released by removing the endpoint from the endpoint list.] */ /* Codes_SRS_CONNECTION_01_131: [Any incoming channel number associated with the endpoint shall be released.] */ - if (i < connection->endpoint_count && i > 0) + if ((i < connection->endpoint_count) && (i > 0)) { - (void)memmove(connection->endpoints + i, connection->endpoints + i + 1, sizeof(ENDPOINT_INSTANCE*) * (connection->endpoint_count - i - 1)); + (void)memmove(connection->endpoints + i, connection->endpoints + i + 1, sizeof(ENDPOINT_HANDLE) * (connection->endpoint_count - i - 1)); - ENDPOINT_INSTANCE** new_endpoints = (ENDPOINT_INSTANCE**)realloc(connection->endpoints, (connection->endpoint_count - 1) * sizeof(ENDPOINT_INSTANCE*)); + ENDPOINT_HANDLE* new_endpoints = (ENDPOINT_HANDLE*)realloc(connection->endpoints, (connection->endpoint_count - 1) * sizeof(ENDPOINT_HANDLE)); if (new_endpoints != NULL) { connection->endpoints = new_endpoints; @@ -1614,17 +1814,20 @@ if ((endpoint == NULL) || (performative == NULL)) { - result = __FAILURE__; + LogError("Bad arguments: endpoint = %p, performative = %p", + endpoint, performative); + result = __FAILURE__; } else { - CONNECTION_INSTANCE* connection = (CONNECTION_INSTANCE*)endpoint->connection; + CONNECTION_HANDLE connection = (CONNECTION_HANDLE)endpoint->connection; AMQP_FRAME_CODEC_HANDLE amqp_frame_codec = connection->amqp_frame_codec; /* Codes_SRS_CONNECTION_01_254: [If connection_encode_frame is called before the connection is in the OPENED state, connection_encode_frame shall fail and return a non-zero value.] */ if (connection->connection_state != CONNECTION_STATE_OPENED) { - result = __FAILURE__; + LogError("Connection not open"); + result = __FAILURE__; } else { @@ -1637,7 +1840,8 @@ if (amqp_frame_codec_encode_frame(amqp_frame_codec, endpoint->outgoing_channel, performative, payloads, payload_count, on_bytes_encoded, connection) != 0) { /* Codes_SRS_CONNECTION_01_253: [If amqp_frame_codec_begin_encode_frame or amqp_frame_codec_encode_payload_bytes fails, then connection_encode_frame shall fail and return a non-zero value.] */ - result = __FAILURE__; + LogError("Encoding AMQP frame failed"); + result = __FAILURE__; } else { @@ -1648,7 +1852,8 @@ if (tickcounter_get_current_ms(connection->tick_counter, &connection->last_frame_sent_time) != 0) { - result = __FAILURE__; + LogError("Getting tick counter value failed"); + result = __FAILURE__; } else { @@ -1662,12 +1867,16 @@ return result; } -void connection_set_trace(CONNECTION_HANDLE connection, bool traceOn) +void connection_set_trace(CONNECTION_HANDLE connection, bool trace_on) { /* Codes_SRS_CONNECTION_07_002: [If connection is NULL then connection_set_trace shall do nothing.] */ - if (connection != NULL) - { + if (connection == NULL) + { + LogError("NULL connection"); + } + else + { /* Codes_SRS_CONNECTION_07_001: [connection_set_trace shall set the ability to turn on and off trace logging.] */ - connection->is_trace_on = traceOn ? 1 : 0; + connection->is_trace_on = trace_on ? 1 : 0; } }