Azure IoT
A small memory footprint AMQP implimentation
Dependents: iothub_client_sample_amqp remote_monitoring simplesample_amqp
Diff: connection.c
- Revision:
- 35:d0bed2404ee9
- Parent:
- 30:0407b2db334c
- Child:
- 38:7631b92cc772
--- a/connection.c Sat Oct 21 20:12:19 2017 +0000
+++ b/connection.c Fri Nov 03 13:18:57 2017 -0700
@@ -69,6 +69,8 @@
uint16_t channel_max;
milliseconds idle_timeout;
milliseconds remote_idle_timeout;
+ milliseconds remote_idle_timeout_send_frame_millisecond;
+ double idle_timeout_empty_frame_send_ratio;
tickcounter_ms_t last_frame_received_time;
tickcounter_ms_t last_frame_sent_time;
@@ -816,7 +818,12 @@
}
else
{
- (void)open_get_idle_time_out(open_handle, &connection->remote_idle_timeout);
+ if (open_get_idle_time_out(open_handle, &connection->remote_idle_timeout) == 0)
+ {
+ /* since we obtained the remote_idle_timeout, compute at what millisecond we should send the empty frame */
+ connection->remote_idle_timeout_send_frame_millisecond = (milliseconds)(connection->idle_timeout_empty_frame_send_ratio * 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->remote_max_frame_size < 512))
@@ -1051,7 +1058,7 @@
/* 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_HANDLE result;
+ CONNECTION_HANDLE connection;
if ((xio == NULL) ||
(container_id == NULL))
@@ -1059,143 +1066,145 @@
/* Codes_SRS_CONNECTION_01_071: [If xio or container_id is NULL, connection_create shall return NULL.] */
LogError("Bad arguments: xio = %p, container_id = %p",
xio, container_id);
- result = NULL;
+ connection = NULL;
}
else
{
- result = (CONNECTION_HANDLE)malloc(sizeof(CONNECTION_INSTANCE));
+ connection = (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 (connection == NULL)
{
LogError("Cannot allocate memory for connection");
}
else
{
- result->io = xio;
+ connection->io = xio;
/* Codes_SRS_CONNECTION_01_082: [connection_create shall allocate a new frame_codec instance to be used for frame encoding/decoding.] */
- result->frame_codec = frame_codec_create(frame_codec_error, result);
- if (result->frame_codec == NULL)
+ connection->frame_codec = frame_codec_create(frame_codec_error, connection);
+ if (connection->frame_codec == NULL)
{
/* Codes_SRS_CONNECTION_01_083: [If frame_codec_create fails then connection_create shall return NULL.] */
LogError("Cannot create frame_codec");
- free(result);
- result = NULL;
+ free(connection);
+ connection = NULL;
}
else
{
- result->amqp_frame_codec = amqp_frame_codec_create(result->frame_codec, on_amqp_frame_received, on_empty_amqp_frame_received, amqp_frame_codec_error, result);
- if (result->amqp_frame_codec == NULL)
+ connection->amqp_frame_codec = amqp_frame_codec_create(connection->frame_codec, on_amqp_frame_received, on_empty_amqp_frame_received, amqp_frame_codec_error, connection);
+ if (connection->amqp_frame_codec == NULL)
{
/* Codes_SRS_CONNECTION_01_108: [If amqp_frame_codec_create fails, connection_create shall return NULL.] */
LogError("Cannot create amqp_frame_codec");
- frame_codec_destroy(result->frame_codec);
- free(result);
- result = NULL;
+ frame_codec_destroy(connection->frame_codec);
+ free(connection);
+ connection = NULL;
}
else
{
if (hostname != NULL)
{
size_t hostname_length = strlen(hostname);
- result->host_name = (char*)malloc(hostname_length + 1);
- if (result->host_name == NULL)
+ connection->host_name = (char*)malloc(hostname_length + 1);
+ if (connection->host_name == NULL)
{
/* Codes_SRS_CONNECTION_01_081: [If allocating the memory for the connection fails then connection_create shall return NULL.] */
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;
+ amqp_frame_codec_destroy(connection->amqp_frame_codec);
+ frame_codec_destroy(connection->frame_codec);
+ free(connection);
+ connection = NULL;
}
else
{
- (void)memcpy(result->host_name, hostname, hostname_length + 1);
+ (void)memcpy(connection->host_name, hostname, hostname_length + 1);
}
}
else
{
- result->host_name = NULL;
+ connection->host_name = NULL;
}
- if (result != NULL)
+ if (connection != NULL)
{
size_t container_id_length = strlen(container_id);
- result->container_id = (char*)malloc(container_id_length + 1);
- if (result->container_id == NULL)
+ connection->container_id = (char*)malloc(container_id_length + 1);
+ if (connection->container_id == NULL)
{
/* Codes_SRS_CONNECTION_01_081: [If allocating the memory for the connection fails then connection_create shall return NULL.] */
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);
- result = NULL;
+ free(connection->host_name);
+ amqp_frame_codec_destroy(connection->amqp_frame_codec);
+ frame_codec_destroy(connection->frame_codec);
+ free(connection);
+ connection = NULL;
}
else
{
- result->tick_counter = tickcounter_create();
- if (result->tick_counter == NULL)
+ connection->tick_counter = tickcounter_create();
+ if (connection->tick_counter == NULL)
{
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);
- free(result);
- result = NULL;
+ free(connection->container_id);
+ free(connection->host_name);
+ amqp_frame_codec_destroy(connection->amqp_frame_codec);
+ frame_codec_destroy(connection->frame_codec);
+ free(connection);
+ connection = NULL;
}
else
{
- (void)memcpy(result->container_id, container_id, container_id_length + 1);
+ (void)memcpy(connection->container_id, container_id, container_id_length + 1);
/* Codes_SRS_CONNECTION_01_173: [<field name="max-frame-size" type="uint" default="4294967295"/>] */
- result->max_frame_size = 4294967295u;
+ connection->max_frame_size = 4294967295u;
/* Codes: [<field name="channel-max" type="ushort" default="65535"/>] */
- result->channel_max = 65535;
+ connection->channel_max = 65535;
/* Codes_SRS_CONNECTION_01_175: [<field name="idle-time-out" type="milliseconds"/>] */
/* Codes_SRS_CONNECTION_01_192: [A value of zero is the same as if it was not set (null).] */
- result->idle_timeout = 0;
- result->remote_idle_timeout = 0;
+ connection->idle_timeout = 0;
+ connection->remote_idle_timeout = 0;
+ connection->remote_idle_timeout_send_frame_millisecond = 0;
+ connection->idle_timeout_empty_frame_send_ratio = 0.5;
- result->endpoint_count = 0;
- result->endpoints = NULL;
- result->header_bytes_received = 0;
- result->is_remote_frame_received = 0;
+ connection->endpoint_count = 0;
+ connection->endpoints = NULL;
+ connection->header_bytes_received = 0;
+ connection->is_remote_frame_received = 0;
- result->is_underlying_io_open = 0;
- result->remote_max_frame_size = 512;
- result->is_trace_on = 0;
+ connection->is_underlying_io_open = 0;
+ connection->remote_max_frame_size = 512;
+ connection->is_trace_on = 0;
/* Mark that settings have not yet been set by the user */
- result->idle_timeout_specified = 0;
+ connection->idle_timeout_specified = 0;
- result->on_new_endpoint = on_new_endpoint;
- result->on_new_endpoint_callback_context = callback_context;
+ connection->on_new_endpoint = on_new_endpoint;
+ connection->on_new_endpoint_callback_context = callback_context;
- result->on_io_error = on_io_error;
- result->on_io_error_callback_context = on_io_error_context;
- result->on_connection_state_changed = on_connection_state_changed;
- result->on_connection_state_changed_callback_context = on_connection_state_changed_context;
+ connection->on_io_error = on_io_error;
+ connection->on_io_error_callback_context = on_io_error_context;
+ connection->on_connection_state_changed = on_connection_state_changed;
+ connection->on_connection_state_changed_callback_context = on_connection_state_changed_context;
- if (tickcounter_get_current_ms(result->tick_counter, &result->last_frame_received_time) != 0)
+ if (tickcounter_get_current_ms(connection->tick_counter, &connection->last_frame_received_time) != 0)
{
LogError("Could not retrieve time for last frame received time");
- tickcounter_destroy(result->tick_counter);
- free(result->container_id);
- free(result->host_name);
- amqp_frame_codec_destroy(result->amqp_frame_codec);
- frame_codec_destroy(result->frame_codec);
- free(result);
- result = NULL;
+ tickcounter_destroy(connection->tick_counter);
+ free(connection->container_id);
+ free(connection->host_name);
+ amqp_frame_codec_destroy(connection->amqp_frame_codec);
+ frame_codec_destroy(connection->frame_codec);
+ free(connection);
+ connection = NULL;
}
else
{
- result->last_frame_sent_time = result->last_frame_received_time;
+ connection->last_frame_sent_time = connection->last_frame_received_time;
/* Codes_SRS_CONNECTION_01_072: [When connection_create succeeds, the state of the connection shall be CONNECTION_STATE_START.] */
- connection_set_state(result, CONNECTION_STATE_START);
+ connection_set_state(connection, CONNECTION_STATE_START);
}
}
}
@@ -1205,7 +1214,7 @@
}
}
- return result;
+ return connection;
}
void connection_destroy(CONNECTION_HANDLE connection)
@@ -1546,7 +1555,7 @@
uint64_t connection_handle_deadlines(CONNECTION_HANDLE connection)
{
- uint64_t local_deadline = (uint64_t )-1;
+ uint64_t local_deadline = (uint64_t)-1;
uint64_t remote_deadline = (uint64_t)-1;
if (connection == NULL)
@@ -1586,7 +1595,7 @@
{
/* Calculate time until remote idle timeout expires */
- uint64_t remote_idle_timeout = (connection->remote_idle_timeout / 2);
+ uint64_t remote_idle_timeout = connection->remote_idle_timeout_send_frame_millisecond;
uint64_t time_since_last_sent = current_ms - connection->last_frame_sent_time;
if (time_since_last_sent < remote_idle_timeout)
@@ -1883,3 +1892,24 @@
connection->is_trace_on = trace_on ? 1 : 0;
}
}
+
+int connection_set_remote_idle_timeout_empty_frame_send_ratio(CONNECTION_HANDLE connection, double idle_timeout_empty_frame_send_ratio)
+{
+ int result;
+
+ if ((connection == NULL) ||
+ (idle_timeout_empty_frame_send_ratio <= 0.0) ||
+ (idle_timeout_empty_frame_send_ratio > 1.0))
+ {
+ LogError("Bad arguments: connection = %p, idle_timeout_empty_frame_send_ratio = %f",
+ connection, idle_timeout_empty_frame_send_ratio);
+ result = __FAILURE__;
+ }
+ else
+ {
+ connection->idle_timeout_empty_frame_send_ratio = idle_timeout_empty_frame_send_ratio;
+ result = 0;
+ }
+
+ return result;
+}
