Azure IoT
/
iothub_amqp_transport
Important changes to repositories hosted on mbed.com
Mbed hosted mercurial repositories are deprecated and are due to be permanently deleted in July 2026.
To keep a copy of this software download the repository Zip archive or clone locally using Mercurial.
It is also possible to export all your personal repositories from the account settings page.
Dependents: sht15_remote_monitoring RobotArmDemo iothub_client_sample_amqp iothub_client_sample_amqp ... more
Revision 36:f78f9a56869e, committed 2017-06-16
- Comitter:
- AzureIoTClient
- Date:
- Fri Jun 16 16:12:09 2017 -0700
- Parent:
- 35:e8f259604ec3
- Child:
- 37:abd16824f63b
- Commit message:
- 1.1.17
Changed in this revision
--- a/iothubtransport_amqp_device.c Fri Jun 02 15:52:19 2017 -0700
+++ b/iothubtransport_amqp_device.c Fri Jun 16 16:12:09 2017 -0700
@@ -2,7 +2,7 @@
// Licensed under the MIT license. See LICENSE file in the project root for full license information.
#include <stdlib.h>
-#include "iothubtransport_amqp_messenger.h"
+#include "iothubtransport_amqp_telemetry_messenger.h"
#include "azure_c_shared_utility/optimize_size.h"
#include "azure_c_shared_utility/gballoc.h"
#include "azure_c_shared_utility/agenttime.h"
@@ -33,8 +33,8 @@
time_t auth_state_last_changed_time;
size_t auth_state_change_timeout_secs;
- MESSENGER_HANDLE messenger_handle;
- MESSENGER_STATE msgr_state;
+ TELEMETRY_MESSENGER_HANDLE messenger_handle;
+ TELEMETRY_MESSENGER_STATE msgr_state;
time_t msgr_state_last_changed_time;
size_t msgr_state_change_timeout_secs;
@@ -100,25 +100,25 @@
}
// Callback Handlers
-static D2C_EVENT_SEND_RESULT get_d2c_event_send_result_from(MESSENGER_EVENT_SEND_COMPLETE_RESULT result)
+static D2C_EVENT_SEND_RESULT get_d2c_event_send_result_from(TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT result)
{
D2C_EVENT_SEND_RESULT d2c_esr;
switch (result)
{
- case MESSENGER_EVENT_SEND_COMPLETE_RESULT_OK:
+ case TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_OK:
d2c_esr = D2C_EVENT_SEND_COMPLETE_RESULT_OK;
break;
- case MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE:
+ case TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE:
d2c_esr = D2C_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE;
break;
- case MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING:
+ case TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING:
d2c_esr = D2C_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING;
break;
- case MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT:
+ case TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT:
d2c_esr = D2C_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT;
break;
- case MESSENGER_EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED:
+ case TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED:
d2c_esr = D2C_EVENT_SEND_COMPLETE_RESULT_DEVICE_DESTROYED;
break;
default:
@@ -130,7 +130,7 @@
return d2c_esr;
}
-static void on_event_send_complete_messenger_callback(IOTHUB_MESSAGE_LIST* iothub_message, MESSENGER_EVENT_SEND_COMPLETE_RESULT ev_send_comp_result, void* context)
+static void on_event_send_complete_messenger_callback(IOTHUB_MESSAGE_LIST* iothub_message, TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT ev_send_comp_result, void* context)
{
if (iothub_message == NULL || context == NULL)
{
@@ -140,11 +140,11 @@
{
DEVICE_SEND_EVENT_TASK* send_task = (DEVICE_SEND_EVENT_TASK*)context;
- // Codes_SRS_DEVICE_09_059: [If `ev_send_comp_result` is MESSENGER_EVENT_SEND_COMPLETE_RESULT_OK, D2C_EVENT_SEND_COMPLETE_RESULT_OK shall be reported as `event_send_complete`]
- // Codes_SRS_DEVICE_09_060: [If `ev_send_comp_result` is MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE, D2C_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE shall be reported as `event_send_complete`]
- // Codes_SRS_DEVICE_09_061: [If `ev_send_comp_result` is MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING, D2C_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING shall be reported as `event_send_complete`]
- // Codes_SRS_DEVICE_09_062: [If `ev_send_comp_result` is MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT, D2C_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT shall be reported as `event_send_complete`]
- // Codes_SRS_DEVICE_09_063: [If `ev_send_comp_result` is MESSENGER_EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED, D2C_EVENT_SEND_COMPLETE_RESULT_DEVICE_DESTROYED shall be reported as `event_send_complete`]
+ // Codes_SRS_DEVICE_09_059: [If `ev_send_comp_result` is TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_OK, D2C_EVENT_SEND_COMPLETE_RESULT_OK shall be reported as `event_send_complete`]
+ // Codes_SRS_DEVICE_09_060: [If `ev_send_comp_result` is TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE, D2C_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE shall be reported as `event_send_complete`]
+ // Codes_SRS_DEVICE_09_061: [If `ev_send_comp_result` is TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING, D2C_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING shall be reported as `event_send_complete`]
+ // Codes_SRS_DEVICE_09_062: [If `ev_send_comp_result` is TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT, D2C_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT shall be reported as `event_send_complete`]
+ // Codes_SRS_DEVICE_09_063: [If `ev_send_comp_result` is TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED, D2C_EVENT_SEND_COMPLETE_RESULT_DEVICE_DESTROYED shall be reported as `event_send_complete`]
D2C_EVENT_SEND_RESULT device_send_result = get_d2c_event_send_result_from(ev_send_comp_result);
// Codes_SRS_DEVICE_09_064: [If provided, the user callback and context saved in `send_task` shall be invoked passing the device `event_send_complete`]
@@ -189,7 +189,7 @@
}
}
-static void on_messenger_state_changed_callback(void* context, MESSENGER_STATE previous_state, MESSENGER_STATE new_state)
+static void on_messenger_state_changed_callback(void* context, TELEMETRY_MESSENGER_STATE previous_state, TELEMETRY_MESSENGER_STATE new_state)
{
if (context == NULL)
{
@@ -207,7 +207,7 @@
}
}
-static DEVICE_MESSAGE_DISPOSITION_INFO* create_device_message_disposition_info_from(MESSENGER_MESSAGE_DISPOSITION_INFO* messenger_disposition_info)
+static DEVICE_MESSAGE_DISPOSITION_INFO* create_device_message_disposition_info_from(TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* messenger_disposition_info)
{
DEVICE_MESSAGE_DISPOSITION_INFO* device_disposition_info;
@@ -235,17 +235,17 @@
free(disposition_info);
}
-static MESSENGER_MESSAGE_DISPOSITION_INFO* create_messenger_disposition_info(DEVICE_MESSAGE_DISPOSITION_INFO* device_disposition_info)
+static TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* create_messenger_disposition_info(DEVICE_MESSAGE_DISPOSITION_INFO* device_disposition_info)
{
- MESSENGER_MESSAGE_DISPOSITION_INFO* messenger_disposition_info;
+ TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* messenger_disposition_info;
- if ((messenger_disposition_info = (MESSENGER_MESSAGE_DISPOSITION_INFO*)malloc(sizeof(MESSENGER_MESSAGE_DISPOSITION_INFO))) == NULL)
+ if ((messenger_disposition_info = (TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO*)malloc(sizeof(TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO))) == NULL)
{
- LogError("Failed creating MESSENGER_MESSAGE_DISPOSITION_INFO (malloc failed)");
+ LogError("Failed creating TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO (malloc failed)");
}
else if (mallocAndStrcpy_s(&messenger_disposition_info->source, device_disposition_info->source) != RESULT_OK)
{
- LogError("Failed creating MESSENGER_MESSAGE_DISPOSITION_INFO (mallocAndStrcpy_s failed)");
+ LogError("Failed creating TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO (mallocAndStrcpy_s failed)");
free(messenger_disposition_info);
messenger_disposition_info = NULL;
}
@@ -257,48 +257,48 @@
return messenger_disposition_info;
}
-static void destroy_messenger_disposition_info(MESSENGER_MESSAGE_DISPOSITION_INFO* messenger_disposition_info)
+static void destroy_messenger_disposition_info(TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* messenger_disposition_info)
{
free(messenger_disposition_info->source);
free(messenger_disposition_info);
}
-static MESSENGER_DISPOSITION_RESULT get_messenger_message_disposition_result_from(DEVICE_MESSAGE_DISPOSITION_RESULT device_disposition_result)
+static TELEMETRY_MESSENGER_DISPOSITION_RESULT get_messenger_message_disposition_result_from(DEVICE_MESSAGE_DISPOSITION_RESULT device_disposition_result)
{
- MESSENGER_DISPOSITION_RESULT messenger_disposition_result;
+ TELEMETRY_MESSENGER_DISPOSITION_RESULT messenger_disposition_result;
switch (device_disposition_result)
{
case DEVICE_MESSAGE_DISPOSITION_RESULT_NONE:
- messenger_disposition_result = MESSENGER_DISPOSITION_RESULT_NONE;
+ messenger_disposition_result = TELEMETRY_MESSENGER_DISPOSITION_RESULT_NONE;
break;
case DEVICE_MESSAGE_DISPOSITION_RESULT_ACCEPTED:
- messenger_disposition_result = MESSENGER_DISPOSITION_RESULT_ACCEPTED;
+ messenger_disposition_result = TELEMETRY_MESSENGER_DISPOSITION_RESULT_ACCEPTED;
break;
case DEVICE_MESSAGE_DISPOSITION_RESULT_REJECTED:
- messenger_disposition_result = MESSENGER_DISPOSITION_RESULT_REJECTED;
+ messenger_disposition_result = TELEMETRY_MESSENGER_DISPOSITION_RESULT_REJECTED;
break;
case DEVICE_MESSAGE_DISPOSITION_RESULT_RELEASED:
- messenger_disposition_result = MESSENGER_DISPOSITION_RESULT_RELEASED;
+ messenger_disposition_result = TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED;
break;
default:
- LogError("Failed to get the corresponding MESSENGER_DISPOSITION_RESULT (%d is not supported)", device_disposition_result);
- messenger_disposition_result = MESSENGER_DISPOSITION_RESULT_RELEASED;
+ LogError("Failed to get the corresponding TELEMETRY_MESSENGER_DISPOSITION_RESULT (%d is not supported)", device_disposition_result);
+ messenger_disposition_result = TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED;
break;
}
return messenger_disposition_result;
}
-static MESSENGER_DISPOSITION_RESULT on_messenger_message_received_callback(IOTHUB_MESSAGE_HANDLE iothub_message_handle, MESSENGER_MESSAGE_DISPOSITION_INFO* disposition_info, void* context)
+static TELEMETRY_MESSENGER_DISPOSITION_RESULT on_messenger_message_received_callback(IOTHUB_MESSAGE_HANDLE iothub_message_handle, TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* disposition_info, void* context)
{
- MESSENGER_DISPOSITION_RESULT msgr_disposition_result;
+ TELEMETRY_MESSENGER_DISPOSITION_RESULT msgr_disposition_result;
- // Codes_SRS_DEVICE_09_070: [If `iothub_message_handle` or `context` is NULL, on_messenger_message_received_callback shall return MESSENGER_DISPOSITION_RESULT_RELEASED]
+ // Codes_SRS_DEVICE_09_070: [If `iothub_message_handle` or `context` is NULL, on_messenger_message_received_callback shall return TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED]
if (iothub_message_handle == NULL || context == NULL)
{
LogError("Failed receiving incoming C2D message (message handle (%p) or context (%p) is NULL)", iothub_message_handle, context);
- msgr_disposition_result = MESSENGER_DISPOSITION_RESULT_RELEASED;
+ msgr_disposition_result = TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED;
}
else
{
@@ -307,7 +307,7 @@
if (device_instance->on_message_received_callback == NULL)
{
LogError("Device '%s' failed receiving incoming C2D message (callback is NULL)", device_instance->config->device_id);
- msgr_disposition_result = MESSENGER_DISPOSITION_RESULT_RELEASED;
+ msgr_disposition_result = TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED;
}
else
{
@@ -316,18 +316,18 @@
// Codes_SRS_DEVICE_09_119: [A DEVICE_MESSAGE_DISPOSITION_INFO instance shall be created containing a copy of `disposition_info->source` and `disposition_info->message_id`]
if ((device_message_disposition_info = create_device_message_disposition_info_from(disposition_info)) == NULL)
{
- // Codes_SRS_DEVICE_09_120: [If the DEVICE_MESSAGE_DISPOSITION_INFO instance fails to be created, on_messenger_message_received_callback shall return MESSENGER_DISPOSITION_RESULT_RELEASED]
+ // Codes_SRS_DEVICE_09_120: [If the DEVICE_MESSAGE_DISPOSITION_INFO instance fails to be created, on_messenger_message_received_callback shall return TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED]
LogError("Device '%s' failed receiving incoming C2D message (failed creating DEVICE_MESSAGE_DISPOSITION_INFO)", device_instance->config->device_id);
- msgr_disposition_result = MESSENGER_DISPOSITION_RESULT_RELEASED;
+ msgr_disposition_result = TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED;
}
else
{
// Codes_SRS_DEVICE_09_071: [The user callback shall be invoked, passing the context it provided]
DEVICE_MESSAGE_DISPOSITION_RESULT device_disposition_result = device_instance->on_message_received_callback(iothub_message_handle, device_message_disposition_info, device_instance->on_message_received_context);
- // Codes_SRS_DEVICE_09_072: [If the user callback returns DEVICE_MESSAGE_DISPOSITION_RESULT_ACCEPTED, on_messenger_message_received_callback shall return MESSENGER_DISPOSITION_RESULT_ACCEPTED]
- // Codes_SRS_DEVICE_09_073: [If the user callback returns DEVICE_MESSAGE_DISPOSITION_RESULT_REJECTED, on_messenger_message_received_callback shall return MESSENGER_DISPOSITION_RESULT_REJECTED]
- // Codes_SRS_DEVICE_09_074: [If the user callback returns DEVICE_MESSAGE_DISPOSITION_RESULT_RELEASED, on_messenger_message_received_callback shall return MESSENGER_DISPOSITION_RESULT_RELEASED]
+ // Codes_SRS_DEVICE_09_072: [If the user callback returns DEVICE_MESSAGE_DISPOSITION_RESULT_ACCEPTED, on_messenger_message_received_callback shall return TELEMETRY_MESSENGER_DISPOSITION_RESULT_ACCEPTED]
+ // Codes_SRS_DEVICE_09_073: [If the user callback returns DEVICE_MESSAGE_DISPOSITION_RESULT_REJECTED, on_messenger_message_received_callback shall return TELEMETRY_MESSENGER_DISPOSITION_RESULT_REJECTED]
+ // Codes_SRS_DEVICE_09_074: [If the user callback returns DEVICE_MESSAGE_DISPOSITION_RESULT_RELEASED, on_messenger_message_received_callback shall return TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED]
msgr_disposition_result = get_messenger_message_disposition_result_from(device_disposition_result);
// Codes_SRS_DEVICE_09_121: [on_messenger_message_received_callback shall release the memory allocated for DEVICE_MESSAGE_DISPOSITION_INFO]
@@ -414,7 +414,7 @@
{
if (instance->messenger_handle != NULL)
{
- messenger_destroy(instance->messenger_handle);
+ telemetry_messenger_destroy(instance->messenger_handle);
}
if (instance->authentication_handle != NULL)
@@ -451,15 +451,15 @@
{
int result;
- MESSENGER_CONFIG messenger_config;
+ TELEMETRY_MESSENGER_CONFIG messenger_config;
messenger_config.device_id = instance->config->device_id;
messenger_config.iothub_host_fqdn = instance->config->iothub_host_fqdn;
messenger_config.on_state_changed_callback = on_messenger_state_changed_callback;
messenger_config.on_state_changed_context = instance;
- if ((instance->messenger_handle = messenger_create(&messenger_config, pi)) == NULL)
+ if ((instance->messenger_handle = telemetry_messenger_create(&messenger_config, pi)) == NULL)
{
- LogError("Failed creating the MESSENGER_HANDLE (messenger_create failed)");
+ LogError("Failed creating the TELEMETRY_MESSENGER_HANDLE (messenger_create failed)");
result = __FAILURE__;
}
else
@@ -573,17 +573,17 @@
LogError("Failed creating the device instance for device '%s' (failed creating the authentication instance)", instance->config->device_id);
result = __FAILURE__;
}
- // Codes_SRS_DEVICE_09_008: [`instance->messenger_handle` shall be set using messenger_create()]
+ // Codes_SRS_DEVICE_09_008: [`instance->messenger_handle` shall be set using telemetry_messenger_create()]
else if (create_messenger_instance(instance, config->product_info) != RESULT_OK)
{
- // Codes_SRS_DEVICE_09_009: [If the MESSENGER_HANDLE fails to be created, device_create shall fail and return NULL]
+ // Codes_SRS_DEVICE_09_009: [If the TELEMETRY_MESSENGER_HANDLE fails to be created, device_create shall fail and return NULL]
LogError("Failed creating the device instance for device '%s' (failed creating the messenger instance)", instance->config->device_id);
result = __FAILURE__;
}
else
{
instance->auth_state = AUTHENTICATION_STATE_STOPPED;
- instance->msgr_state = MESSENGER_STATE_STOPPED;
+ instance->msgr_state = TELEMETRY_MESSENGER_STATE_STOPPED;
instance->state = DEVICE_STATE_STOPPED;
instance->auth_state_last_changed_time = INDEFINITE_TIME;
instance->auth_state_change_timeout_secs = DEFAULT_AUTH_STATE_CHANGED_TIMEOUT_SECS;
@@ -683,10 +683,10 @@
// Codes_SRS_DEVICE_09_026: [The device state shall be updated to DEVICE_STATE_STOPPING, and state changed callback invoked]
update_state(instance, DEVICE_STATE_STOPPING);
- // Codes_SRS_DEVICE_09_027: [If `instance->messenger_handle` state is not MESSENGER_STATE_STOPPED, messenger_stop shall be invoked]
- if (instance->msgr_state != MESSENGER_STATE_STOPPED &&
- instance->msgr_state != MESSENGER_STATE_STOPPING &&
- messenger_stop(instance->messenger_handle) != RESULT_OK)
+ // Codes_SRS_DEVICE_09_027: [If `instance->messenger_handle` state is not TELEMETRY_MESSENGER_STATE_STOPPED, messenger_stop shall be invoked]
+ if (instance->msgr_state != TELEMETRY_MESSENGER_STATE_STOPPED &&
+ instance->msgr_state != TELEMETRY_MESSENGER_STATE_STOPPING &&
+ telemetry_messenger_stop(instance->messenger_handle) != RESULT_OK)
{
// Codes_SRS_DEVICE_09_028: [If messenger_stop fails, the `instance` state shall be updated to DEVICE_STATE_ERROR_MSG and the function shall return non-zero result]
LogError("Failed stopping device '%s' (messenger_stop failed)", instance->config->device_id);
@@ -780,19 +780,19 @@
// Codes_SRS_DEVICE_09_040: [Messenger shall not be started if using CBS authentication and authentication start has not completed yet]
if (instance->config->authentication_mode == DEVICE_AUTH_MODE_X509 || instance->auth_state == AUTHENTICATION_STATE_STARTED)
{
- // Codes_SRS_DEVICE_09_041: [If messenger state is MESSENGER_STATE_STOPPED, messenger_start shall be invoked]
- if (instance->msgr_state == MESSENGER_STATE_STOPPED)
+ // Codes_SRS_DEVICE_09_041: [If messenger state is TELEMETRY_MESSENGER_STATE_STOPPED, messenger_start shall be invoked]
+ if (instance->msgr_state == TELEMETRY_MESSENGER_STATE_STOPPED)
{
// Codes_SRS_DEVICE_09_042: [If messenger_start fails, the device state shall be updated to DEVICE_STATE_ERROR_MSG]
- if (messenger_start(instance->messenger_handle, instance->session_handle) != RESULT_OK)
+ if (telemetry_messenger_start(instance->messenger_handle, instance->session_handle) != RESULT_OK)
{
LogError("Device '%s' messenger failed to be started (messenger_start failed)", instance->config->device_id);
update_state(instance, DEVICE_STATE_ERROR_MSG);
}
}
- // Codes_SRS_DEVICE_09_043: [If messenger state is MESSENGER_STATE_STARTING, the device shall track the time since last event change and timeout if needed]
- else if (instance->msgr_state == MESSENGER_STATE_STARTING)
+ // Codes_SRS_DEVICE_09_043: [If messenger state is TELEMETRY_MESSENGER_STATE_STARTING, the device shall track the time since last event change and timeout if needed]
+ else if (instance->msgr_state == TELEMETRY_MESSENGER_STATE_STARTING)
{
int is_timed_out;
if (is_timeout_reached(instance->msgr_state_last_changed_time, instance->msgr_state_change_timeout_secs, &is_timed_out) != RESULT_OK)
@@ -809,15 +809,15 @@
update_state(instance, DEVICE_STATE_ERROR_MSG);
}
}
- // Codes_SRS_DEVICE_09_045: [If messenger state is MESSENGER_STATE_ERROR, the device state shall be updated to DEVICE_STATE_ERROR_MSG]
- else if (instance->msgr_state == MESSENGER_STATE_ERROR)
+ // Codes_SRS_DEVICE_09_045: [If messenger state is TELEMETRY_MESSENGER_STATE_ERROR, the device state shall be updated to DEVICE_STATE_ERROR_MSG]
+ else if (instance->msgr_state == TELEMETRY_MESSENGER_STATE_ERROR)
{
LogError("Device '%s' messenger failed to be started (messenger got into error state)", instance->config->device_id);
update_state(instance, DEVICE_STATE_ERROR_MSG);
}
- // Codes_SRS_DEVICE_09_046: [If messenger state is MESSENGER_STATE_STARTED, the device state shall be updated to DEVICE_STATE_STARTED]
- else if (instance->msgr_state == MESSENGER_STATE_STARTED)
+ // Codes_SRS_DEVICE_09_046: [If messenger state is TELEMETRY_MESSENGER_STATE_STARTED, the device state shall be updated to DEVICE_STATE_STARTED]
+ else if (instance->msgr_state == TELEMETRY_MESSENGER_STATE_STARTED)
{
update_state(instance, DEVICE_STATE_STARTED);
}
@@ -849,8 +849,8 @@
}
else
{
- // Codes_SRS_DEVICE_09_048: [If messenger state is not MESSENGER_STATE_STARTED, the device state shall be updated to DEVICE_STATE_ERROR_MSG]
- if (instance->msgr_state != MESSENGER_STATE_STARTED)
+ // Codes_SRS_DEVICE_09_048: [If messenger state is not TELEMETRY_MESSENGER_STATE_STARTED, the device state shall be updated to DEVICE_STATE_ERROR_MSG]
+ if (instance->msgr_state != TELEMETRY_MESSENGER_STATE_STARTED)
{
LogError("Device '%s' is started but messenger reported unexpected state %d", instance->config->device_id, instance->msgr_state);
update_state(instance, DEVICE_STATE_ERROR_MSG);
@@ -868,10 +868,10 @@
}
}
- if (instance->msgr_state != MESSENGER_STATE_STOPPED && instance->msgr_state != MESSENGER_STATE_ERROR)
+ if (instance->msgr_state != TELEMETRY_MESSENGER_STATE_STOPPED && instance->msgr_state != TELEMETRY_MESSENGER_STATE_ERROR)
{
// Codes_SRS_DEVICE_09_050: [If `instance->messenger_handle` state is not STOPPED or ERROR, authentication_do_work shall be invoked]
- messenger_do_work(instance->messenger_handle);
+ telemetry_messenger_do_work(instance->messenger_handle);
}
}
}
@@ -892,7 +892,7 @@
(void)device_stop((DEVICE_HANDLE)instance);
}
- // Codes_SRS_DEVICE_09_014: [`instance->messenger_handle shall be destroyed using messenger_destroy()`]
+ // Codes_SRS_DEVICE_09_014: [`instance->messenger_handle shall be destroyed using telemetry_messenger_destroy()`]
// Codes_SRS_DEVICE_09_015: [If created, `instance->authentication_handle` shall be destroyed using authentication_destroy()`]
// Codes_SRS_DEVICE_09_016: [The contents of `instance->config` shall be detroyed and then it shall be freed]
internal_destroy_device((DEVICE_INSTANCE*)handle);
@@ -929,11 +929,11 @@
send_task->on_event_send_complete_callback = on_device_d2c_event_send_complete_callback;
send_task->on_event_send_complete_context = context;
- // Codes_SRS_DEVICE_09_055: [The message shall be sent using messenger_send_async, passing `on_event_send_complete_messenger_callback` and `send_task`]
- if (messenger_send_async(instance->messenger_handle, message, on_event_send_complete_messenger_callback, (void*)send_task) != RESULT_OK)
+ // Codes_SRS_DEVICE_09_055: [The message shall be sent using telemetry_messenger_send_async, passing `on_event_send_complete_messenger_callback` and `send_task`]
+ if (telemetry_messenger_send_async(instance->messenger_handle, message, on_event_send_complete_messenger_callback, (void*)send_task) != RESULT_OK)
{
- // Codes_SRS_DEVICE_09_056: [If messenger_send_async fails, device_send_event_async shall return a non-zero value]
- LogError("Failed sending event (messenger_send_async failed)");
+ // Codes_SRS_DEVICE_09_056: [If telemetry_messenger_send_async fails, device_send_event_async shall return a non-zero value]
+ LogError("Failed sending event (telemetry_messenger_send_async failed)");
// Codes_SRS_DEVICE_09_057: [If any failures occur, device_send_event_async shall release all memory it has allocated]
free(send_task);
result = __FAILURE__;
@@ -963,24 +963,24 @@
else
{
DEVICE_INSTANCE* instance = (DEVICE_INSTANCE*)handle;
- MESSENGER_SEND_STATUS messenger_send_status;
+ TELEMETRY_MESSENGER_SEND_STATUS messenger_send_status;
- // Codes_SRS_DEVICE_09_106: [The status of `instance->messenger_handle` shall be obtained using messenger_get_send_status]
- if (messenger_get_send_status(instance->messenger_handle, &messenger_send_status) != RESULT_OK)
+ // Codes_SRS_DEVICE_09_106: [The status of `instance->messenger_handle` shall be obtained using telemetry_messenger_get_send_status]
+ if (telemetry_messenger_get_send_status(instance->messenger_handle, &messenger_send_status) != RESULT_OK)
{
- // Codes_SRS_DEVICE_09_107: [If messenger_get_send_status fails, device_get_send_status shall return a non-zero result]
- LogError("Failed getting the device messenger send status (messenger_get_send_status failed)");
+ // Codes_SRS_DEVICE_09_107: [If telemetry_messenger_get_send_status fails, device_get_send_status shall return a non-zero result]
+ LogError("Failed getting the device messenger send status (telemetry_messenger_get_send_status failed)");
result = __FAILURE__;
}
else
{
- // Codes_SRS_DEVICE_09_108: [If messenger_get_send_status returns MESSENGER_SEND_STATUS_IDLE, device_get_send_status return status DEVICE_SEND_STATUS_IDLE]
- if (messenger_send_status == MESSENGER_SEND_STATUS_IDLE)
+ // Codes_SRS_DEVICE_09_108: [If telemetry_messenger_get_send_status returns TELEMETRY_MESSENGER_SEND_STATUS_IDLE, device_get_send_status return status DEVICE_SEND_STATUS_IDLE]
+ if (messenger_send_status == TELEMETRY_MESSENGER_SEND_STATUS_IDLE)
{
*send_status = DEVICE_SEND_STATUS_IDLE;
}
- // Codes_SRS_DEVICE_09_109: [If messenger_get_send_status returns MESSENGER_SEND_STATUS_BUSY, device_get_send_status return status DEVICE_SEND_STATUS_BUSY]
- else // i.e., messenger_send_status == MESSENGER_SEND_STATUS_BUSY
+ // Codes_SRS_DEVICE_09_109: [If telemetry_messenger_get_send_status returns TELEMETRY_MESSENGER_SEND_STATUS_BUSY, device_get_send_status return status DEVICE_SEND_STATUS_BUSY]
+ else // i.e., messenger_send_status == TELEMETRY_MESSENGER_SEND_STATUS_BUSY
{
*send_status = DEVICE_SEND_STATUS_BUSY;
}
@@ -1008,11 +1008,11 @@
{
DEVICE_INSTANCE* instance = (DEVICE_INSTANCE*)handle;
- // Codes_SRS_DEVICE_09_067: [messenger_subscribe_for_messages shall be invoked passing `on_messenger_message_received_callback` and the user callback and context]
- if (messenger_subscribe_for_messages(instance->messenger_handle, on_messenger_message_received_callback, handle) != RESULT_OK)
+ // Codes_SRS_DEVICE_09_067: [telemetry_messenger_subscribe_for_messages shall be invoked passing `on_messenger_message_received_callback` and the user callback and context]
+ if (telemetry_messenger_subscribe_for_messages(instance->messenger_handle, on_messenger_message_received_callback, handle) != RESULT_OK)
{
- // Codes_SRS_DEVICE_09_068: [If messenger_subscribe_for_messages fails, device_subscribe_message shall return a non-zero result]
- LogError("Failed subscribing to C2D messages (messenger_subscribe_for_messages failed)");
+ // Codes_SRS_DEVICE_09_068: [If telemetry_messenger_subscribe_for_messages fails, device_subscribe_message shall return a non-zero result]
+ LogError("Failed subscribing to C2D messages (telemetry_messenger_subscribe_for_messages failed)");
result = __FAILURE__;
}
else
@@ -1042,11 +1042,11 @@
{
DEVICE_INSTANCE* instance = (DEVICE_INSTANCE*)handle;
- // Codes_SRS_DEVICE_09_077: [messenger_unsubscribe_for_messages shall be invoked passing `instance->messenger_handle`]
- if (messenger_unsubscribe_for_messages(instance->messenger_handle) != RESULT_OK)
+ // Codes_SRS_DEVICE_09_077: [telemetry_messenger_unsubscribe_for_messages shall be invoked passing `instance->messenger_handle`]
+ if (telemetry_messenger_unsubscribe_for_messages(instance->messenger_handle) != RESULT_OK)
{
- // Codes_SRS_DEVICE_09_078: [If messenger_unsubscribe_for_messages fails, device_unsubscribe_message shall return a non-zero result]
- LogError("Failed unsubscribing to C2D messages (messenger_unsubscribe_for_messages failed)");
+ // Codes_SRS_DEVICE_09_078: [If telemetry_messenger_unsubscribe_for_messages fails, device_unsubscribe_message shall return a non-zero result]
+ LogError("Failed unsubscribing to C2D messages (telemetry_messenger_unsubscribe_for_messages failed)");
result = __FAILURE__;
}
else
@@ -1077,24 +1077,24 @@
else
{
DEVICE_INSTANCE* device = (DEVICE_INSTANCE*)device_handle;
- MESSENGER_MESSAGE_DISPOSITION_INFO* messenger_disposition_info;
+ TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* messenger_disposition_info;
- // Codes_SRS_DEVICE_09_113: [A MESSENGER_MESSAGE_DISPOSITION_INFO instance shall be created with a copy of the `source` and `message_id` contained in `disposition_info`]
+ // Codes_SRS_DEVICE_09_113: [A TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO instance shall be created with a copy of the `source` and `message_id` contained in `disposition_info`]
if ((messenger_disposition_info = create_messenger_disposition_info(disposition_info)) == NULL)
{
- // Codes_SRS_DEVICE_09_114: [If the MESSENGER_MESSAGE_DISPOSITION_INFO fails to be created, device_send_message_disposition() shall fail and return __FAILURE__]
- LogError("Failed sending message disposition (failed to create MESSENGER_MESSAGE_DISPOSITION_INFO)");
+ // Codes_SRS_DEVICE_09_114: [If the TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO fails to be created, device_send_message_disposition() shall fail and return __FAILURE__]
+ LogError("Failed sending message disposition (failed to create TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO)");
result = __FAILURE__;
}
else
{
- MESSENGER_DISPOSITION_RESULT messenger_disposition_result = get_messenger_message_disposition_result_from(disposition_result);
+ TELEMETRY_MESSENGER_DISPOSITION_RESULT messenger_disposition_result = get_messenger_message_disposition_result_from(disposition_result);
- // Codes_SRS_DEVICE_09_115: [`messenger_send_message_disposition()` shall be invoked passing the MESSENGER_MESSAGE_DISPOSITION_INFO instance and the corresponding MESSENGER_DISPOSITION_RESULT]
- if (messenger_send_message_disposition(device->messenger_handle, messenger_disposition_info, messenger_disposition_result) != RESULT_OK)
+ // Codes_SRS_DEVICE_09_115: [`telemetry_messenger_send_message_disposition()` shall be invoked passing the TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO instance and the corresponding TELEMETRY_MESSENGER_DISPOSITION_RESULT]
+ if (telemetry_messenger_send_message_disposition(device->messenger_handle, messenger_disposition_info, messenger_disposition_result) != RESULT_OK)
{
- // Codes_SRS_DEVICE_09_116: [If `messenger_send_message_disposition()` fails, device_send_message_disposition() shall fail and return __FAILURE__]
- LogError("Failed sending message disposition (messenger_send_message_disposition failed)");
+ // Codes_SRS_DEVICE_09_116: [If `telemetry_messenger_send_message_disposition()` fails, device_send_message_disposition() shall fail and return __FAILURE__]
+ LogError("Failed sending message disposition (telemetry_messenger_send_message_disposition failed)");
result = __FAILURE__;
}
else
@@ -1103,7 +1103,7 @@
result = RESULT_OK;
}
- // Codes_SRS_DEVICE_09_117: [device_send_message_disposition() shall destroy the MESSENGER_MESSAGE_DISPOSITION_INFO instance]
+ // Codes_SRS_DEVICE_09_117: [device_send_message_disposition() shall destroy the TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO instance]
destroy_messenger_disposition_info(messenger_disposition_info);
}
}
@@ -1172,10 +1172,10 @@
}
else if (strcmp(DEVICE_OPTION_EVENT_SEND_TIMEOUT_SECS, name) == 0)
{
- // Codes_SRS_DEVICE_09_086: [If `name` refers to messenger module, it shall be passed along with `value` to messenger_set_option]
- if (messenger_set_option(instance->messenger_handle, name, value) != RESULT_OK)
+ // Codes_SRS_DEVICE_09_086: [If `name` refers to messenger module, it shall be passed along with `value` to telemetry_messenger_set_option]
+ if (telemetry_messenger_set_option(instance->messenger_handle, MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS, value) != RESULT_OK)
{
- // Codes_SRS_DEVICE_09_087: [If messenger_set_option fails, device_set_option shall return a non-zero result]
+ // Codes_SRS_DEVICE_09_087: [If telemetry_messenger_set_option fails, device_set_option shall return a non-zero result]
LogError("failed setting option for device '%s' (failed setting messenger option '%s')", instance->config->device_id, name);
result = __FAILURE__;
}
@@ -1285,10 +1285,10 @@
LogError("Failed to retrieve options from device '%s' (OptionHandler_AddOption failed for option '%s')", instance->config->device_id, DEVICE_OPTION_SAVED_AUTH_OPTIONS);
result = NULL;
}
- // Codes_SRS_DEVICE_09_099: [`instance->messenger_handle` options shall be retrieved using messenger_retrieve_options]
- else if ((dependency_options = messenger_retrieve_options(instance->messenger_handle)) == NULL)
+ // Codes_SRS_DEVICE_09_099: [`instance->messenger_handle` options shall be retrieved using telemetry_messenger_retrieve_options]
+ else if ((dependency_options = telemetry_messenger_retrieve_options(instance->messenger_handle)) == NULL)
{
- // Codes_SRS_DEVICE_09_100: [If messenger_retrieve_options fails, device_retrieve_options shall return NULL]
+ // Codes_SRS_DEVICE_09_100: [If telemetry_messenger_retrieve_options fails, device_retrieve_options shall return NULL]
LogError("Failed to retrieve options from device '%s' (failed to retrieve options from messenger instance)", instance->config->device_id);
result = NULL;
}
--- a/iothubtransport_amqp_messenger.c Fri Jun 02 15:52:19 2017 -0700
+++ /dev/null Thu Jan 01 00:00:00 1970 +0000
@@ -1,1898 +0,0 @@
-// Copyright (c) Microsoft. All rights reserved.
-// Licensed under the MIT license. See LICENSE file in the project root for full license information.
-
-#include <stdlib.h>
-#include <stdbool.h>
-#include "azure_c_shared_utility/optimize_size.h"
-#include "azure_c_shared_utility/crt_abstractions.h"
-#include "azure_c_shared_utility/gballoc.h"
-#include "azure_c_shared_utility/agenttime.h"
-#include "azure_c_shared_utility/xlogging.h"
-#include "azure_c_shared_utility/uniqueid.h"
-#include "azure_c_shared_utility/singlylinkedlist.h"
-#include "azure_uamqp_c/link.h"
-#include "azure_uamqp_c/messaging.h"
-#include "azure_uamqp_c/message_sender.h"
-#include "azure_uamqp_c/message_receiver.h"
-#include "uamqp_messaging.h"
-#include "iothub_client_private.h"
-#include "iothub_client_version.h"
-#include "iothubtransport_amqp_messenger.h"
-
-#define RESULT_OK 0
-#define INDEFINITE_TIME ((time_t)(-1))
-
-#define IOTHUB_DEVICES_PATH_FMT "%s/devices/%s"
-#define IOTHUB_EVENT_SEND_ADDRESS_FMT "amqps://%s/messages/events"
-#define IOTHUB_MESSAGE_RECEIVE_ADDRESS_FMT "amqps://%s/messages/devicebound"
-#define MESSAGE_SENDER_LINK_NAME_PREFIX "link-snd"
-#define MESSAGE_SENDER_MAX_LINK_SIZE UINT64_MAX
-#define MESSAGE_RECEIVER_LINK_NAME_PREFIX "link-rcv"
-#define MESSAGE_RECEIVER_MAX_LINK_SIZE 65536
-#define DEFAULT_EVENT_SEND_RETRY_LIMIT 10
-#define DEFAULT_EVENT_SEND_TIMEOUT_SECS 600
-#define MAX_MESSAGE_SENDER_STATE_CHANGE_TIMEOUT_SECS 300
-#define MAX_MESSAGE_RECEIVER_STATE_CHANGE_TIMEOUT_SECS 300
-#define UNIQUE_ID_BUFFER_SIZE 37
-#define STRING_NULL_TERMINATOR '\0'
-
-typedef struct MESSENGER_INSTANCE_TAG
-{
- STRING_HANDLE device_id;
- STRING_HANDLE product_info;
- STRING_HANDLE iothub_host_fqdn;
- SINGLYLINKEDLIST_HANDLE waiting_to_send;
- SINGLYLINKEDLIST_HANDLE in_progress_list;
- MESSENGER_STATE state;
-
- ON_MESSENGER_STATE_CHANGED_CALLBACK on_state_changed_callback;
- void* on_state_changed_context;
-
- bool receive_messages;
- ON_MESSENGER_MESSAGE_RECEIVED on_message_received_callback;
- void* on_message_received_context;
-
- SESSION_HANDLE session_handle;
- LINK_HANDLE sender_link;
- MESSAGE_SENDER_HANDLE message_sender;
- MESSAGE_SENDER_STATE message_sender_current_state;
- MESSAGE_SENDER_STATE message_sender_previous_state;
- LINK_HANDLE receiver_link;
- MESSAGE_RECEIVER_HANDLE message_receiver;
- MESSAGE_RECEIVER_STATE message_receiver_current_state;
- MESSAGE_RECEIVER_STATE message_receiver_previous_state;
-
- size_t event_send_retry_limit;
- size_t event_send_error_count;
- size_t event_send_timeout_secs;
- time_t last_message_sender_state_change_time;
- time_t last_message_receiver_state_change_time;
-} MESSENGER_INSTANCE;
-
-typedef struct MESSENGER_SEND_EVENT_TASK_TAG
-{
- IOTHUB_MESSAGE_LIST* message;
- ON_MESSENGER_EVENT_SEND_COMPLETE on_event_send_complete_callback;
- void* context;
- time_t send_time;
- MESSENGER_INSTANCE *messenger;
- bool is_timed_out;
-} MESSENGER_SEND_EVENT_TASK;
-
-// @brief
-// Evaluates if the ammount of time since start_time is greater or lesser than timeout_in_secs.
-// @param is_timed_out
-// Set to 1 if a timeout has been reached, 0 otherwise. Not set if any failure occurs.
-// @returns
-// 0 if no failures occur, non-zero otherwise.
-static int is_timeout_reached(time_t start_time, size_t timeout_in_secs, int *is_timed_out)
-{
- int result;
-
- if (start_time == INDEFINITE_TIME)
- {
- LogError("Failed to verify timeout (start_time is INDEFINITE)");
- result = __FAILURE__;
- }
- else
- {
- time_t current_time;
-
- if ((current_time = get_time(NULL)) == INDEFINITE_TIME)
- {
- LogError("Failed to verify timeout (get_time failed)");
- result = __FAILURE__;
- }
- else
- {
- if (get_difftime(current_time, start_time) >= timeout_in_secs)
- {
- *is_timed_out = 1;
- }
- else
- {
- *is_timed_out = 0;
- }
-
- result = RESULT_OK;
- }
- }
-
- return result;
-}
-
-static STRING_HANDLE create_devices_path(STRING_HANDLE iothub_host_fqdn, STRING_HANDLE device_id)
-{
- STRING_HANDLE devices_path;
-
- if ((devices_path = STRING_new()) == NULL)
- {
- LogError("Failed creating devices_path (STRING_new failed)");
- }
- else
- {
- const char* iothub_host_fqdn_char_ptr = STRING_c_str(iothub_host_fqdn);
- const char* device_id_char_ptr = STRING_c_str(device_id);
- if (STRING_sprintf(devices_path, IOTHUB_DEVICES_PATH_FMT, iothub_host_fqdn_char_ptr, device_id_char_ptr) != RESULT_OK)
- {
- STRING_delete(devices_path);
- devices_path = NULL;
- LogError("Failed creating devices_path (STRING_sprintf failed)");
- }
- }
-
- return devices_path;
-}
-
-static STRING_HANDLE create_event_send_address(STRING_HANDLE devices_path)
-{
- STRING_HANDLE event_send_address;
-
- if ((event_send_address = STRING_new()) == NULL)
- {
- LogError("Failed creating the event_send_address (STRING_new failed)");
- }
- else
- {
- const char* devices_path_char_ptr = STRING_c_str(devices_path);
- if (STRING_sprintf(event_send_address, IOTHUB_EVENT_SEND_ADDRESS_FMT, devices_path_char_ptr) != RESULT_OK)
- {
- STRING_delete(event_send_address);
- event_send_address = NULL;
- LogError("Failed creating the event_send_address (STRING_sprintf failed)");
- }
- }
-
- return event_send_address;
-}
-
-static STRING_HANDLE create_event_sender_source_name(STRING_HANDLE link_name)
-{
- STRING_HANDLE source_name;
-
- if ((source_name = STRING_new()) == NULL)
- {
- LogError("Failed creating the source_name (STRING_new failed)");
- }
- else
- {
- const char* link_name_char_ptr = STRING_c_str(link_name);
- if (STRING_sprintf(source_name, "%s-source", link_name_char_ptr) != RESULT_OK)
- {
- STRING_delete(source_name);
- source_name = NULL;
- LogError("Failed creating the source_name (STRING_sprintf failed)");
- }
- }
-
- return source_name;
-}
-
-static STRING_HANDLE create_message_receive_address(STRING_HANDLE devices_path)
-{
- STRING_HANDLE message_receive_address;
-
- if ((message_receive_address = STRING_new()) == NULL)
- {
- LogError("Failed creating the message_receive_address (STRING_new failed)");
- }
- else
- {
- const char* devices_path_char_ptr = STRING_c_str(devices_path);
- if (STRING_sprintf(message_receive_address, IOTHUB_MESSAGE_RECEIVE_ADDRESS_FMT, devices_path_char_ptr) != RESULT_OK)
- {
- STRING_delete(message_receive_address);
- message_receive_address = NULL;
- LogError("Failed creating the message_receive_address (STRING_sprintf failed)");
- }
- }
-
- return message_receive_address;
-}
-
-static STRING_HANDLE create_message_receiver_target_name(STRING_HANDLE link_name)
-{
- STRING_HANDLE target_name;
-
- if ((target_name = STRING_new()) == NULL)
- {
- LogError("Failed creating the target_name (STRING_new failed)");
- }
- else
- {
- const char* link_name_char_ptr = STRING_c_str(link_name);
- if (STRING_sprintf(target_name, "%s-target", link_name_char_ptr) != RESULT_OK)
- {
- STRING_delete(target_name);
- target_name = NULL;
- LogError("Failed creating the target_name (STRING_sprintf failed)");
- }
- }
-
- return target_name;
-}
-
-static STRING_HANDLE create_link_name(const char* prefix, const char* infix)
-{
- char* unique_id;
- STRING_HANDLE tag = NULL;
-
- if ((unique_id = (char*)malloc(sizeof(char) * UNIQUE_ID_BUFFER_SIZE + 1)) == NULL)
- {
- LogError("Failed generating an unique tag (malloc failed)");
- }
- else
- {
- memset(unique_id, 0, sizeof(char) * UNIQUE_ID_BUFFER_SIZE + 1);
-
- if (UniqueId_Generate(unique_id, UNIQUE_ID_BUFFER_SIZE) != UNIQUEID_OK)
- {
- LogError("Failed generating an unique tag (UniqueId_Generate failed)");
- }
- else if ((tag = STRING_new()) == NULL)
- {
- LogError("Failed generating an unique tag (STRING_new failed)");
- }
- else if (STRING_sprintf(tag, "%s-%s-%s", prefix, infix, unique_id) != RESULT_OK)
- {
- STRING_delete(tag);
- tag = NULL;
- LogError("Failed generating an unique tag (STRING_sprintf failed)");
- }
-
- free(unique_id);
- }
-
- return tag;
-}
-
-static void update_messenger_state(MESSENGER_INSTANCE* instance, MESSENGER_STATE new_state)
-{
- if (new_state != instance->state)
- {
- MESSENGER_STATE previous_state = instance->state;
- instance->state = new_state;
-
- if (instance->on_state_changed_callback != NULL)
- {
- instance->on_state_changed_callback(instance->on_state_changed_context, previous_state, new_state);
- }
- }
-}
-
-static void attach_device_client_type_to_link(LINK_HANDLE link, STRING_HANDLE product_info)
-{
- fields attach_properties;
- AMQP_VALUE device_client_type_key_name;
- AMQP_VALUE device_client_type_value;
- int result;
-
- if ((attach_properties = amqpvalue_create_map()) == NULL)
- {
- LogError("Failed to create the map for device client type.");
- }
- else
- {
- if ((device_client_type_key_name = amqpvalue_create_symbol("com.microsoft:client-version")) == NULL)
- {
- LogError("Failed to create the key name for the device client type.");
- }
- else
- {
- if ((device_client_type_value = amqpvalue_create_string(STRING_c_str(product_info))) == NULL)
- {
- LogError("Failed to create the key value for the device client type.");
- }
- else
- {
- if ((result = amqpvalue_set_map_value(attach_properties, device_client_type_key_name, device_client_type_value)) != 0)
- {
- LogError("Failed to set the property map for the device client type (error code is: %d)", result);
- }
- else if ((result = link_set_attach_properties(link, attach_properties)) != 0)
- {
- LogError("Unable to attach the device client type to the link properties (error code is: %d)", result);
- }
-
- amqpvalue_destroy(device_client_type_value);
- }
-
- amqpvalue_destroy(device_client_type_key_name);
- }
-
- amqpvalue_destroy(attach_properties);
- }
-}
-
-static void destroy_event_sender(MESSENGER_INSTANCE* instance)
-{
- if (instance->message_sender != NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_060: [`instance->message_sender` shall be destroyed using messagesender_destroy()]
- messagesender_destroy(instance->message_sender);
- instance->message_sender = NULL;
- instance->message_sender_current_state = MESSAGE_SENDER_STATE_IDLE;
- instance->message_sender_previous_state = MESSAGE_SENDER_STATE_IDLE;
- instance->last_message_sender_state_change_time = INDEFINITE_TIME;
- }
-
- if (instance->sender_link != NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_063: [`instance->sender_link` shall be destroyed using link_destroy()]
- link_destroy(instance->sender_link);
- instance->sender_link = NULL;
- }
-}
-
-static void on_event_sender_state_changed_callback(void* context, MESSAGE_SENDER_STATE new_state, MESSAGE_SENDER_STATE previous_state)
-{
- if (context == NULL)
- {
- LogError("on_event_sender_state_changed_callback was invoked with a NULL context; although unexpected, this failure will be ignored");
- }
- else
- {
- if (new_state != previous_state)
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)context;
- instance->message_sender_current_state = new_state;
- instance->message_sender_previous_state = previous_state;
- instance->last_message_sender_state_change_time = get_time(NULL);
- }
- }
-}
-
-static int create_event_sender(MESSENGER_INSTANCE* instance)
-{
- int result;
-
- STRING_HANDLE link_name = NULL;
- STRING_HANDLE source_name = NULL;
- AMQP_VALUE source = NULL;
- AMQP_VALUE target = NULL;
- STRING_HANDLE devices_path = NULL;
- STRING_HANDLE event_send_address = NULL;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_033: [A variable, named `devices_path`, shall be created concatenating `instance->iothub_host_fqdn`, "/devices/" and `instance->device_id`]
- if ((devices_path = create_devices_path(instance->iothub_host_fqdn, instance->device_id)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_034: [If `devices_path` fails to be created, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message sender (failed creating the 'devices_path')");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_035: [A variable, named `event_send_address`, shall be created concatenating "amqps://", `devices_path` and "/messages/events"]
- else if ((event_send_address = create_event_send_address(devices_path)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_036: [If `event_send_address` fails to be created, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message sender (failed creating the 'event_send_address')");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_037: [A `link_name` variable shall be created using an unique string label per AMQP session]
- else if ((link_name = create_link_name(MESSAGE_SENDER_LINK_NAME_PREFIX, STRING_c_str(instance->device_id))) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_038: [If `link_name` fails to be created, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message sender (failed creating an unique link name)");
- }
- else if ((source_name = create_event_sender_source_name(link_name)) == NULL)
- {
- result = __FAILURE__;
- LogError("Failed creating the message sender (failed creating an unique source name)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_039: [A `source` variable shall be created with messaging_create_source() using an unique string label per AMQP session]
- else if ((source = messaging_create_source(STRING_c_str(source_name))) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_040: [If `source` fails to be created, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message sender (messaging_create_source failed)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_041: [A `target` variable shall be created with messaging_create_target() using `event_send_address`]
- else if ((target = messaging_create_target(STRING_c_str(event_send_address))) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_042: [If `target` fails to be created, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message sender (messaging_create_target failed)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_043: [`instance->sender_link` shall be set using link_create(), passing `instance->session_handle`, `link_name`, "role_sender", `source` and `target` as parameters]
- else if ((instance->sender_link = link_create(instance->session_handle, STRING_c_str(link_name), role_sender, source, target)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_044: [If link_create() fails, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message sender (link_create failed)");
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_047: [`instance->sender_link` maximum message size shall be set to UINT64_MAX using link_set_max_message_size()]
- if (link_set_max_message_size(instance->sender_link, MESSAGE_SENDER_MAX_LINK_SIZE) != RESULT_OK)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_048: [If link_set_max_message_size() fails, it shall be logged and ignored.]
- LogError("Failed setting message sender link max message size.");
- }
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_049: [`instance->sender_link` should have a property "com.microsoft:client-version" set as `CLIENT_DEVICE_TYPE_PREFIX/IOTHUB_SDK_VERSION`, using amqpvalue_set_map_value() and link_set_attach_properties()]
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_050: [If amqpvalue_set_map_value() or link_set_attach_properties() fail, the failure shall be ignored]
- attach_device_client_type_to_link(instance->sender_link, instance->product_info);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_051: [`instance->message_sender` shall be created using messagesender_create(), passing the `instance->sender_link` and `on_event_sender_state_changed_callback`]
- if ((instance->message_sender = messagesender_create(instance->sender_link, on_event_sender_state_changed_callback, (void*)instance)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_052: [If messagesender_create() fails, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- link_destroy(instance->sender_link);
- instance->sender_link = NULL;
- LogError("Failed creating the message sender (messagesender_create failed)");
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_053: [`instance->message_sender` shall be opened using messagesender_open()]
- if (messagesender_open(instance->message_sender) != RESULT_OK)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_054: [If messagesender_open() fails, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- messagesender_destroy(instance->message_sender);
- instance->message_sender = NULL;
- link_destroy(instance->sender_link);
- instance->sender_link = NULL;
- LogError("Failed opening the AMQP message sender.");
- }
- else
- {
- result = RESULT_OK;
- }
- }
- }
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_055: [Before returning, messenger_do_work() shall release all the temporary memory it has allocated]
- if (link_name != NULL)
- STRING_delete(link_name);
- if (source_name != NULL)
- STRING_delete(source_name);
- if (source != NULL)
- amqpvalue_destroy(source);
- if (target != NULL)
- amqpvalue_destroy(target);
- if (devices_path != NULL)
- STRING_delete(devices_path);
- if (event_send_address != NULL)
- STRING_delete(event_send_address);
-
- return result;
-}
-
-static void destroy_message_receiver(MESSENGER_INSTANCE* instance)
-{
- if (instance->message_receiver != NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_061: [`instance->message_receiver` shall be closed using messagereceiver_close()]
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_093: [`instance->message_receiver` shall be closed using messagereceiver_close()]
- if (messagereceiver_close(instance->message_receiver) != RESULT_OK)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_094: [If messagereceiver_close() fails, it shall be logged and ignored]
- LogError("Failed closing the AMQP message receiver (this failure will be ignored).");
- }
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_062: [`instance->message_receiver` shall be destroyed using messagereceiver_destroy()]
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_095: [`instance->message_receiver` shall be destroyed using messagereceiver_destroy()]
- messagereceiver_destroy(instance->message_receiver);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_096: [`instance->message_receiver` shall be set to NULL]
- instance->message_receiver = NULL;
- instance->message_receiver_current_state = MESSAGE_RECEIVER_STATE_IDLE;
- instance->message_receiver_previous_state = MESSAGE_RECEIVER_STATE_IDLE;
- instance->last_message_receiver_state_change_time = INDEFINITE_TIME;
- }
-
- if (instance->receiver_link != NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_064: [`instance->receiver_link` shall be destroyed using link_destroy()]
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_097: [`instance->receiver_link` shall be destroyed using link_destroy()]
- link_destroy(instance->receiver_link);
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_098: [`instance->receiver_link` shall be set to NULL]
- instance->receiver_link = NULL;
- }
-}
-
-static void on_message_receiver_state_changed_callback(const void* context, MESSAGE_RECEIVER_STATE new_state, MESSAGE_RECEIVER_STATE previous_state)
-{
- if (context == NULL)
- {
- LogError("on_message_receiver_state_changed_callback was invoked with a NULL context; although unexpected, this failure will be ignored");
- }
- else
- {
- if (new_state != previous_state)
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)context;
- instance->message_receiver_current_state = new_state;
- instance->message_receiver_previous_state = previous_state;
- instance->last_message_receiver_state_change_time = get_time(NULL);
- }
- }
-}
-
-static MESSENGER_MESSAGE_DISPOSITION_INFO* create_message_disposition_info(MESSENGER_INSTANCE* messenger)
-{
- MESSENGER_MESSAGE_DISPOSITION_INFO* result;
-
- if ((result = (MESSENGER_MESSAGE_DISPOSITION_INFO*)malloc(sizeof(MESSENGER_MESSAGE_DISPOSITION_INFO))) == NULL)
- {
- LogError("Failed creating MESSENGER_MESSAGE_DISPOSITION_INFO container (malloc failed)");
- result = NULL;
- }
- else
- {
- delivery_number message_id;
-
- if (messagereceiver_get_received_message_id(messenger->message_receiver, &message_id) != RESULT_OK)
- {
- LogError("Failed creating MESSENGER_MESSAGE_DISPOSITION_INFO container (messagereceiver_get_received_message_id failed)");
- free(result);
- result = NULL;
- }
- else
- {
- const char* link_name;
-
- if (messagereceiver_get_link_name(messenger->message_receiver, &link_name) != RESULT_OK)
- {
- LogError("Failed creating MESSENGER_MESSAGE_DISPOSITION_INFO container (messagereceiver_get_link_name failed)");
- free(result);
- result = NULL;
- }
- else if (mallocAndStrcpy_s(&result->source, link_name) != RESULT_OK)
- {
- LogError("Failed creating MESSENGER_MESSAGE_DISPOSITION_INFO container (failed copying link name)");
- free(result);
- result = NULL;
- }
- else
- {
- result->message_id = message_id;
- }
- }
- }
-
- return result;
-}
-
-static void destroy_message_disposition_info(MESSENGER_MESSAGE_DISPOSITION_INFO* disposition_info)
-{
- free(disposition_info->source);
- free(disposition_info);
-}
-
-static AMQP_VALUE create_uamqp_disposition_result_from(MESSENGER_DISPOSITION_RESULT disposition_result)
-{
- AMQP_VALUE uamqp_disposition_result;
-
- if (disposition_result == MESSENGER_DISPOSITION_RESULT_NONE)
- {
- uamqp_disposition_result = NULL; // intentionally not sending an answer.
- }
- else if (disposition_result == MESSENGER_DISPOSITION_RESULT_ACCEPTED)
- {
- uamqp_disposition_result = messaging_delivery_accepted();
- }
- else if (disposition_result == MESSENGER_DISPOSITION_RESULT_RELEASED)
- {
- uamqp_disposition_result = messaging_delivery_released();
- }
- else if (disposition_result == MESSENGER_DISPOSITION_RESULT_REJECTED)
- {
- uamqp_disposition_result = messaging_delivery_rejected("Rejected by application", "Rejected by application");
- }
- else
- {
- LogError("Failed creating a disposition result for messagereceiver (result %d is not supported)", disposition_result);
- uamqp_disposition_result = NULL;
- }
-
- return uamqp_disposition_result;
-}
-
-static AMQP_VALUE on_message_received_internal_callback(const void* context, MESSAGE_HANDLE message)
-{
- AMQP_VALUE result;
- int api_call_result;
- IOTHUB_MESSAGE_HANDLE iothub_message;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_121: [An IOTHUB_MESSAGE_HANDLE shall be obtained from MESSAGE_HANDLE using IoTHubMessage_CreateFromUamqpMessage()]
- if ((api_call_result = IoTHubMessage_CreateFromUamqpMessage(message, &iothub_message)) != RESULT_OK)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_122: [If IoTHubMessage_CreateFromUamqpMessage() fails, on_message_received_internal_callback shall return the result of messaging_delivery_rejected()]
- result = messaging_delivery_rejected("Rejected due to failure reading AMQP message", "Failed reading AMQP message");
-
- LogError("on_message_received_internal_callback failed (IoTHubMessage_CreateFromUamqpMessage; error = %d).", api_call_result);
- }
- else
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)context;
- MESSENGER_MESSAGE_DISPOSITION_INFO* message_disposition_info;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_186: [A MESSENGER_MESSAGE_DISPOSITION_INFO instance shall be created containing the source link name and message delivery ID]
- if ((message_disposition_info = create_message_disposition_info(instance)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_187: [**If the MESSENGER_MESSAGE_DISPOSITION_INFO instance fails to be created, on_message_received_internal_callback shall return messaging_delivery_released()]
- LogError("on_message_received_internal_callback failed (failed creating MESSENGER_MESSAGE_DISPOSITION_INFO).");
- result = messaging_delivery_released();
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_123: [`instance->on_message_received_callback` shall be invoked passing the IOTHUB_MESSAGE_HANDLE and MESSENGER_MESSAGE_DISPOSITION_INFO instance]
- MESSENGER_DISPOSITION_RESULT disposition_result = instance->on_message_received_callback(iothub_message, message_disposition_info, instance->on_message_received_context);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_188: [The memory allocated for the MESSENGER_MESSAGE_DISPOSITION_INFO instance shall be released]
- destroy_message_disposition_info(message_disposition_info);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_125: [If `instance->on_message_received_callback` returns MESSENGER_DISPOSITION_RESULT_ACCEPTED, on_message_received_internal_callback shall return the result of messaging_delivery_accepted()]
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_126: [If `instance->on_message_received_callback` returns MESSENGER_DISPOSITION_RESULT_RELEASED, on_message_received_internal_callback shall return the result of messaging_delivery_released()]
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_127: [If `instance->on_message_received_callback` returns MESSENGER_DISPOSITION_RESULT_REJECTED, on_message_received_internal_callback shall return the result of messaging_delivery_rejected()]
- result = create_uamqp_disposition_result_from(disposition_result);
- }
- }
-
- return result;
-}
-
-static int create_message_receiver(MESSENGER_INSTANCE* instance)
-{
- int result;
-
- STRING_HANDLE devices_path = NULL;
- STRING_HANDLE message_receive_address = NULL;
- STRING_HANDLE link_name = NULL;
- STRING_HANDLE target_name = NULL;
- AMQP_VALUE source = NULL;
- AMQP_VALUE target = NULL;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_068: [A variable, named `devices_path`, shall be created concatenating `instance->iothub_host_fqdn`, "/devices/" and `instance->device_id`]
- if ((devices_path = create_devices_path(instance->iothub_host_fqdn, instance->device_id)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_069: [If `devices_path` fails to be created, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message receiver (failed creating the 'devices_path')");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_070: [A variable, named `message_receive_address`, shall be created concatenating "amqps://", `devices_path` and "/messages/devicebound"]
- else if ((message_receive_address = create_message_receive_address(devices_path)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_071: [If `message_receive_address` fails to be created, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message receiver (failed creating the 'message_receive_address')");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_072: [A `link_name` variable shall be created using an unique string label per AMQP session]
- else if ((link_name = create_link_name(MESSAGE_RECEIVER_LINK_NAME_PREFIX, STRING_c_str(instance->device_id))) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_073: [If `link_name` fails to be created, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message receiver (failed creating an unique link name)");
- }
- else if ((target_name = create_message_receiver_target_name(link_name)) == NULL)
- {
- result = __FAILURE__;
- LogError("Failed creating the message receiver (failed creating an unique target name)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_074: [A `target` variable shall be created with messaging_create_target() using an unique string label per AMQP session]
- else if ((target = messaging_create_target(STRING_c_str(target_name))) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_075: [If `target` fails to be created, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message receiver (messaging_create_target failed)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_076: [A `source` variable shall be created with messaging_create_source() using `message_receive_address`]
- else if ((source = messaging_create_source(STRING_c_str(message_receive_address))) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_077: [If `source` fails to be created, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message receiver (messaging_create_source failed)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_078: [`instance->receiver_link` shall be set using link_create(), passing `instance->session_handle`, `link_name`, "role_receiver", `source` and `target` as parameters]
- else if ((instance->receiver_link = link_create(instance->session_handle, STRING_c_str(link_name), role_receiver, source, target)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_079: [If link_create() fails, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message receiver (link_create failed)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_080: [`instance->receiver_link` settle mode shall be set to "receiver_settle_mode_first" using link_set_rcv_settle_mode(), ]
- else if (link_set_rcv_settle_mode(instance->receiver_link, receiver_settle_mode_first) != RESULT_OK)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_081: [If link_set_rcv_settle_mode() fails, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- LogError("Failed creating the message receiver (link_set_rcv_settle_mode failed)");
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_082: [`instance->receiver_link` maximum message size shall be set to 65536 using link_set_max_message_size()]
- if (link_set_max_message_size(instance->receiver_link, MESSAGE_RECEIVER_MAX_LINK_SIZE) != RESULT_OK)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_083: [If link_set_max_message_size() fails, it shall be logged and ignored.]
- LogError("Failed setting message receiver link max message size.");
- }
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_084: [`instance->receiver_link` should have a property "com.microsoft:client-version" set as `CLIENT_DEVICE_TYPE_PREFIX/IOTHUB_SDK_VERSION`, using amqpvalue_set_map_value() and link_set_attach_properties()]
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_085: [If amqpvalue_set_map_value() or link_set_attach_properties() fail, the failure shall be ignored]
- attach_device_client_type_to_link(instance->receiver_link, instance->product_info);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_086: [`instance->message_receiver` shall be created using messagereceiver_create(), passing the `instance->receiver_link` and `on_messagereceiver_state_changed_callback`]
- if ((instance->message_receiver = messagereceiver_create(instance->receiver_link, on_message_receiver_state_changed_callback, (void*)instance)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_087: [If messagereceiver_create() fails, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- link_destroy(instance->receiver_link);
- LogError("Failed creating the message receiver (messagereceiver_create failed)");
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_088: [`instance->message_receiver` shall be opened using messagereceiver_open(), passing `on_message_received_internal_callback`]
- if (messagereceiver_open(instance->message_receiver, on_message_received_internal_callback, (void*)instance) != RESULT_OK)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_089: [If messagereceiver_open() fails, messenger_do_work() shall fail and return]
- result = __FAILURE__;
- messagereceiver_destroy(instance->message_receiver);
- link_destroy(instance->receiver_link);
- LogError("Failed opening the AMQP message receiver.");
- }
- else
- {
- result = RESULT_OK;
- }
- }
- }
-
- if (devices_path != NULL)
- STRING_delete(devices_path);
- if (message_receive_address != NULL)
- STRING_delete(message_receive_address);
- if (link_name != NULL)
- STRING_delete(link_name);
- if (target_name != NULL)
- STRING_delete(target_name);
- if (source != NULL)
- amqpvalue_destroy(source);
- if (target != NULL)
- amqpvalue_destroy(target);
-
- return result;
-}
-
-static int move_event_to_in_progress_list(MESSENGER_SEND_EVENT_TASK* task)
-{
- int result;
-
- if (singlylinkedlist_add(task->messenger->in_progress_list, (void*)task) == NULL)
- {
- result = __FAILURE__;
- LogError("Failed moving event to in_progress list (singlylinkedlist_add failed)");
- }
- else
- {
- result = RESULT_OK;
- }
-
- return result;
-}
-
-static bool find_MESSENGER_SEND_EVENT_TASK_on_list(LIST_ITEM_HANDLE list_item, const void* match_context)
-{
- return (list_item != NULL && singlylinkedlist_item_get_value(list_item) == match_context);
-}
-
-static void remove_event_from_in_progress_list(MESSENGER_SEND_EVENT_TASK *task)
-{
- LIST_ITEM_HANDLE list_item = singlylinkedlist_find(task->messenger->in_progress_list, find_MESSENGER_SEND_EVENT_TASK_on_list, (void*)task);
-
- if (list_item != NULL)
- {
- if (singlylinkedlist_remove(task->messenger->in_progress_list, list_item) != RESULT_OK)
- {
- LogError("Failed removing event from in_progress list (singlylinkedlist_remove failed)");
- }
- }
-}
-
-static int copy_events_to_list(SINGLYLINKEDLIST_HANDLE from_list, SINGLYLINKEDLIST_HANDLE to_list)
-{
- int result;
- LIST_ITEM_HANDLE list_item;
-
- result = RESULT_OK;
- list_item = singlylinkedlist_get_head_item(from_list);
-
- while (list_item != NULL)
- {
- MESSENGER_SEND_EVENT_TASK *task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_item);
-
- if (singlylinkedlist_add(to_list, task) == NULL)
- {
- LogError("Failed copying event to destination list (singlylinkedlist_add failed)");
- result = __FAILURE__;
- break;
- }
- else
- {
- list_item = singlylinkedlist_get_next_item(list_item);
- }
- }
-
- return result;
-}
-
-static int singlylinkedlist_clear(SINGLYLINKEDLIST_HANDLE list)
-{
- int result;
- LIST_ITEM_HANDLE list_item;
-
- result = RESULT_OK;
-
- while ((list_item = singlylinkedlist_get_head_item(list)) != NULL)
- {
- if (singlylinkedlist_remove(list, list_item) != RESULT_OK)
- {
- LogError("Failed removing items from list (%d)", list);
- result = __FAILURE__;
- break;
- }
- }
-
- return result;
-}
-
-static int move_events_to_wait_to_send_list(MESSENGER_INSTANCE* instance)
-{
- int result;
- LIST_ITEM_HANDLE list_item;
-
- if ((list_item = singlylinkedlist_get_head_item(instance->in_progress_list)) == NULL)
- {
- result = RESULT_OK;
- }
- else
- {
- SINGLYLINKEDLIST_HANDLE new_wait_to_send_list;
-
- if ((new_wait_to_send_list = singlylinkedlist_create()) == NULL)
- {
- LogError("Failed moving events back to wait_to_send list (singlylinkedlist_create failed to create new wait_to_send_list)");
- result = __FAILURE__;
- }
- else
- {
- SINGLYLINKEDLIST_HANDLE new_in_progress_list;
-
- if (copy_events_to_list(instance->in_progress_list, new_wait_to_send_list) != RESULT_OK)
- {
- LogError("Failed moving events back to wait_to_send list (failed adding in_progress_list items to new_wait_to_send_list)");
- singlylinkedlist_destroy(new_wait_to_send_list);
- result = __FAILURE__;
- }
- else if (copy_events_to_list(instance->waiting_to_send, new_wait_to_send_list) != RESULT_OK)
- {
- LogError("Failed moving events back to wait_to_send list (failed adding wait_to_send items to new_wait_to_send_list)");
- singlylinkedlist_destroy(new_wait_to_send_list);
- result = __FAILURE__;
- }
- else if ((new_in_progress_list = singlylinkedlist_create()) == NULL)
- {
- LogError("Failed moving events back to wait_to_send list (singlylinkedlist_create failed to create new in_progress_list)");
- singlylinkedlist_destroy(new_wait_to_send_list);
- result = __FAILURE__;
- }
- else
- {
- singlylinkedlist_destroy(instance->waiting_to_send);
- singlylinkedlist_destroy(instance->in_progress_list);
- instance->waiting_to_send = new_wait_to_send_list;
- instance->in_progress_list = new_in_progress_list;
- result = RESULT_OK;
- }
- }
- }
-
- return result;
-}
-
-static void internal_on_event_send_complete_callback(void* context, MESSAGE_SEND_RESULT send_result)
-{
- if (context != NULL)
- {
- MESSENGER_SEND_EVENT_TASK* task = (MESSENGER_SEND_EVENT_TASK*)context;
-
- if (task->messenger->message_sender_current_state != MESSAGE_SENDER_STATE_ERROR)
- {
- if (task->is_timed_out == false)
- {
- MESSENGER_EVENT_SEND_COMPLETE_RESULT messenger_send_result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_107: [If no failure occurs, `task->on_event_send_complete_callback` shall be invoked with result EVENT_SEND_COMPLETE_RESULT_OK]
- if (send_result == MESSAGE_SEND_OK)
- {
- messenger_send_result = MESSENGER_EVENT_SEND_COMPLETE_RESULT_OK;
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_108: [If a failure occurred, `task->on_event_send_complete_callback` shall be invoked with result EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING]
- else
- {
- messenger_send_result = MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING;
- }
-
- task->on_event_send_complete_callback(task->message, messenger_send_result, (void*)task->context);
- }
- else
- {
- LogInfo("messenger on_event_send_complete_callback invoked for timed out event %p; not firing upper layer callback.", task->message);
- }
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_128: [`task` shall be removed from `instance->in_progress_list`]
- remove_event_from_in_progress_list(task);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_130: [`task` shall be destroyed using free()]
- free(task);
- }
- }
-}
-
-static MESSENGER_SEND_EVENT_TASK* get_next_event_to_send(MESSENGER_INSTANCE* instance)
-{
- MESSENGER_SEND_EVENT_TASK* task;
- LIST_ITEM_HANDLE list_item;
-
- if ((list_item = singlylinkedlist_get_head_item(instance->waiting_to_send)) == NULL)
- {
- task = NULL;
- }
- else
- {
- task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_item);
-
- if (singlylinkedlist_remove(instance->waiting_to_send, list_item) != RESULT_OK)
- {
- LogError("Failed removing item from waiting_to_send list (singlylinkedlist_remove failed)");
- }
- }
-
- return task;
-}
-
-static int send_pending_events(MESSENGER_INSTANCE* instance)
-{
- int result = RESULT_OK;
-
- MESSENGER_SEND_EVENT_TASK* task;
-
- while ((task = get_next_event_to_send(instance)) != NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_153: [messenger_do_work() shall move each event to be sent from `instance->wait_to_send_list` to `instance->in_progress_list`]
- if (move_event_to_in_progress_list(task) != RESULT_OK)
- {
- result = __FAILURE__;
- task->on_event_send_complete_callback(task->message, MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING, (void*)task->context);
- break;
- }
- else
- {
- int uamqp_result;
- MESSAGE_HANDLE amqp_message = NULL;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_154: [A MESSAGE_HANDLE shall be obtained out of the event's IOTHUB_MESSAGE_HANDLE instance by using message_create_from_iothub_message()]
- if ((uamqp_result = message_create_from_iothub_message(task->message->messageHandle, &amqp_message)) != RESULT_OK)
- {
- LogError("Failed sending event message (failed creating AMQP message; error: %d).", uamqp_result);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_155: [If message_create_from_iothub_message() fails, `task->on_event_send_complete_callback` shall be invoked with result EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE]
- task->on_event_send_complete_callback(task->message, MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE, (void*)task->context);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_160: [If any failure occurs the event shall be removed from `instance->in_progress_list` and destroyed]
- remove_event_from_in_progress_list(task);
- free(task);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_156: [If message_create_from_iothub_message() fails, messenger_do_work() shall skip to the next event to be sent]
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_157: [The MESSAGE_HANDLE shall be submitted for sending using messagesender_send(), passing `internal_on_event_send_complete_callback`]
- uamqp_result = messagesender_send(instance->message_sender, amqp_message, internal_on_event_send_complete_callback, task);
- task->send_time = get_time(NULL);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_159: [The MESSAGE_HANDLE shall be destroyed using message_destroy().]
- message_destroy(amqp_message);
-
- if (uamqp_result != RESULT_OK)
- {
- LogError("Failed sending event (messagesender_send failed; error: %d)", uamqp_result);
-
- result = __FAILURE__;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_158: [If messagesender_send() fails, `task->on_event_send_complete_callback` shall be invoked with result EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING]
- task->on_event_send_complete_callback(task->message, MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING, (void*)task->context);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_160: [If any failure occurs the event shall be removed from `instance->in_progress_list` and destroyed]
- remove_event_from_in_progress_list(task);
- free(task);
-
- break;
- }
- }
- }
- }
-
- return result;
-}
-
-// @brief
-// Goes through each task in in_progress_list and checks if the events timed out to be sent.
-// @remarks
-// If an event is timed out, it is marked as such but not removed, and the upper layer callback is invoked.
-// @returns
-// 0 if no failures occur, non-zero otherwise.
-static int process_event_send_timeouts(MESSENGER_INSTANCE* instance)
-{
- int result = RESULT_OK;
-
- if (instance->event_send_timeout_secs > 0)
- {
- LIST_ITEM_HANDLE list_item = singlylinkedlist_get_head_item(instance->in_progress_list);
-
- while (list_item != NULL)
- {
- MESSENGER_SEND_EVENT_TASK* task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_item);
-
- if (task->is_timed_out == false)
- {
- int is_timed_out;
-
- if (is_timeout_reached(task->send_time, instance->event_send_timeout_secs, &is_timed_out) == RESULT_OK)
- {
- if (is_timed_out)
- {
- task->is_timed_out = true;
-
- if (task->on_event_send_complete_callback != NULL)
- {
- task->on_event_send_complete_callback(task->message, MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT, task->context);
- }
- }
- }
- else
- {
- LogError("messenger failed to evaluate event send timeout of event %d", task->message);
- result = __FAILURE__;
- }
- }
-
- list_item = singlylinkedlist_get_next_item(list_item);
- }
- }
-
- return result;
-}
-
-// @brief
-// Removes all the timed out events from the in_progress_list, without invoking callbacks or detroying the messages.
-static void remove_timed_out_events(MESSENGER_INSTANCE* instance)
-{
- LIST_ITEM_HANDLE list_item = singlylinkedlist_get_head_item(instance->in_progress_list);
-
- while (list_item != NULL)
- {
- MESSENGER_SEND_EVENT_TASK* task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_item);
-
- if (task->is_timed_out == true)
- {
- remove_event_from_in_progress_list(task);
-
- free(task);
- }
-
- list_item = singlylinkedlist_get_next_item(list_item);
- }
-}
-
-
-// ---------- Set/Retrieve Options Helpers ----------//
-
-static void* messenger_clone_option(const char* name, const void* value)
-{
- void* result;
-
- if (name == NULL)
- {
- LogError("Failed to clone messenger option (name is NULL)");
- result = NULL;
- }
- else if (value == NULL)
- {
- LogError("Failed to clone messenger option (value is NULL)");
- result = NULL;
- }
- else
- {
- if (strcmp(MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS, name) == 0 ||
- strcmp(MESSENGER_OPTION_SAVED_OPTIONS, name) == 0)
- {
- result = (void*)value;
- }
- else
- {
- LogError("Failed to clone messenger option (option with name '%s' is not suppported)", name);
- result = NULL;
- }
- }
-
- return result;
-}
-
-static void messenger_destroy_option(const char* name, const void* value)
-{
- if (name == NULL)
- {
- LogError("Failed to destroy messenger option (name is NULL)");
- }
- else if (value == NULL)
- {
- LogError("Failed to destroy messenger option (value is NULL)");
- }
- else
- {
- // Nothing to be done for the supported options.
- }
-}
-
-
-// Public API:
-
-int messenger_subscribe_for_messages(MESSENGER_HANDLE messenger_handle, ON_MESSENGER_MESSAGE_RECEIVED on_message_received_callback, void* context)
-{
- int result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_016: [If `messenger_handle` is NULL, messenger_subscribe_for_messages() shall fail and return __FAILURE__]
- if (messenger_handle == NULL)
- {
- result = __FAILURE__;
- LogError("messenger_subscribe_for_messages failed (messenger_handle is NULL)");
- }
- else
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)messenger_handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_017: [If `instance->receive_messages` is already true, messenger_subscribe_for_messages() shall fail and return __FAILURE__]
- if (instance->receive_messages)
- {
- result = __FAILURE__;
- LogError("messenger_subscribe_for_messages failed (messenger already subscribed)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_018: [If `on_message_received_callback` is NULL, messenger_subscribe_for_messages() shall fail and return __FAILURE__]
- else if (on_message_received_callback == NULL)
- {
- result = __FAILURE__;
- LogError("messenger_subscribe_for_messages failed (on_message_received_callback is NULL)");
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_019: [`on_message_received_callback` shall be saved on `instance->on_message_received_callback`]
- instance->on_message_received_callback = on_message_received_callback;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_020: [`context` shall be saved on `instance->on_message_received_context`]
- instance->on_message_received_context = context;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_021: [messenger_subscribe_for_messages() shall set `instance->receive_messages` to true]
- instance->receive_messages = true;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_022: [If no failures occurr, messenger_subscribe_for_messages() shall return 0]
- result = RESULT_OK;
- }
- }
-
- return result;
-}
-
-int messenger_unsubscribe_for_messages(MESSENGER_HANDLE messenger_handle)
-{
- int result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_023: [If `messenger_handle` is NULL, messenger_unsubscribe_for_messages() shall fail and return __FAILURE__]
- if (messenger_handle == NULL)
- {
- result = __FAILURE__;
- LogError("messenger_unsubscribe_for_messages failed (messenger_handle is NULL)");
- }
- else
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)messenger_handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_024: [If `instance->receive_messages` is already false, messenger_unsubscribe_for_messages() shall fail and return __FAILURE__]
- if (instance->receive_messages == false)
- {
- result = __FAILURE__;
- LogError("messenger_unsubscribe_for_messages failed (messenger is not subscribed)");
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_025: [messenger_unsubscribe_for_messages() shall set `instance->receive_messages` to false]
- instance->receive_messages = false;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_026: [messenger_unsubscribe_for_messages() shall set `instance->on_message_received_callback` to NULL]
- instance->on_message_received_callback = NULL;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_027: [messenger_unsubscribe_for_messages() shall set `instance->on_message_received_context` to NULL]
- instance->on_message_received_context = NULL;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_028: [If no failures occurr, messenger_unsubscribe_for_messages() shall return 0]
- result = RESULT_OK;
- }
- }
-
- return result;
-}
-
-int messenger_send_message_disposition(MESSENGER_HANDLE messenger_handle, MESSENGER_MESSAGE_DISPOSITION_INFO* disposition_info, MESSENGER_DISPOSITION_RESULT disposition_result)
-{
- int result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_179: [If `messenger_handle` or `disposition_info` are NULL, messenger_send_message_disposition() shall fail and return __FAILURE__]
- if (messenger_handle == NULL || disposition_info == NULL)
- {
- LogError("Failed sending message disposition (either messenger_handle (%p) or disposition_info (%p) are NULL)", messenger_handle, disposition_info);
- result = __FAILURE__;
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_180: [If `disposition_info->source` is NULL, messenger_send_message_disposition() shall fail and return __FAILURE__]
- else if (disposition_info->source == NULL)
- {
- LogError("Failed sending message disposition (disposition_info->source is NULL)");
- result = __FAILURE__;
- }
- else
- {
- MESSENGER_INSTANCE* messenger = (MESSENGER_INSTANCE*)messenger_handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_189: [If `messenger_handle->message_receiver` is NULL, messenger_send_message_disposition() shall fail and return __FAILURE__]
- if (messenger->message_receiver == NULL)
- {
- LogError("Failed sending message disposition (message_receiver is not created; check if it is subscribed)");
- result = __FAILURE__;
- }
- else
- {
- AMQP_VALUE uamqp_disposition_result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_181: [An AMQP_VALUE disposition result shall be created corresponding to the `disposition_result` provided]
- if ((uamqp_disposition_result = create_uamqp_disposition_result_from(disposition_result)) == NULL)
- {
- LogError("Failed sending message disposition (disposition result %d is not supported)", disposition_result);
- result = __FAILURE__;
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_182: [`messagereceiver_send_message_disposition()` shall be invoked passing `disposition_info->source`, `disposition_info->message_id` and the corresponding AMQP_VALUE disposition result]
- if (messagereceiver_send_message_disposition(messenger->message_receiver, disposition_info->source, disposition_info->message_id, uamqp_disposition_result) != RESULT_OK)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_183: [If `messagereceiver_send_message_disposition()` fails, messenger_send_message_disposition() shall fail and return __FAILURE__]
- LogError("Failed sending message disposition (messagereceiver_send_message_disposition failed)");
- result = __FAILURE__;
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_185: [If no failures occurr, messenger_send_message_disposition() shall return 0]
- result = RESULT_OK;
- }
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_184: [messenger_send_message_disposition() shall destroy the AMQP_VALUE disposition result]
- amqpvalue_destroy(uamqp_disposition_result);
- }
- }
- }
-
- return result;
-}
-
-int messenger_send_async(MESSENGER_HANDLE messenger_handle, IOTHUB_MESSAGE_LIST* message, ON_MESSENGER_EVENT_SEND_COMPLETE on_messenger_event_send_complete_callback, void* context)
-{
- int result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_134: [If `messenger_handle` is NULL, messenger_send_async() shall fail and return a non-zero value]
- if (messenger_handle == NULL)
- {
- LogError("Failed sending event (messenger_handle is NULL)");
- result = __FAILURE__;
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_135: [If `message` is NULL, messenger_send_async() shall fail and return a non-zero value]
- else if (message == NULL)
- {
- LogError("Failed sending event (message is NULL)");
- result = __FAILURE__;
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_136: [If `on_event_send_complete_callback` is NULL, messenger_send_async() shall fail and return a non-zero value]
- else if (on_messenger_event_send_complete_callback == NULL)
- {
- LogError("Failed sending event (on_event_send_complete_callback is NULL)");
- result = __FAILURE__;
- }
- else
- {
- MESSENGER_SEND_EVENT_TASK *task;
- MESSENGER_INSTANCE *instance = (MESSENGER_INSTANCE*)messenger_handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_137: [messenger_send_async() shall allocate memory for a MESSENGER_SEND_EVENT_TASK structure (aka `task`)]
- if ((task = (MESSENGER_SEND_EVENT_TASK*)malloc(sizeof(MESSENGER_SEND_EVENT_TASK))) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_138: [If malloc() fails, messenger_send_async() shall fail and return a non-zero value]
- LogError("Failed sending event (failed to create struct for task; malloc failed)");
- result = __FAILURE__;
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_100: [`task` shall be added to `instance->waiting_to_send` using singlylinkedlist_add()]
- else if (singlylinkedlist_add(instance->waiting_to_send, task) == NULL)
- {
- LogError("Failed sending event (singlylinkedlist_add failed)");
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_139: [If singlylinkedlist_add() fails, messenger_send_async() shall fail and return a non-zero value]
- result = __FAILURE__;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_142: [If any failure occurs, messenger_send_async() shall free any memory it has allocated]
- free(task);
- }
- else
- {
- memset(task, 0, sizeof(MESSENGER_SEND_EVENT_TASK));
- task->message = message;
- task->on_event_send_complete_callback = on_messenger_event_send_complete_callback;
- task->context = context;
- task->send_time = INDEFINITE_TIME;
- task->messenger = instance;
- task->is_timed_out = false;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_143: [If no failures occur, messenger_send_async() shall return zero]
- result = RESULT_OK;
- }
- }
-
- return result;
-}
-
-int messenger_get_send_status(MESSENGER_HANDLE messenger_handle, MESSENGER_SEND_STATUS* send_status)
-{
- int result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_144: [If `messenger_handle` is NULL, messenger_get_send_status() shall fail and return a non-zero value]
- if (messenger_handle == NULL)
- {
- LogError("messenger_get_send_status failed (messenger_handle is NULL)");
- result = __FAILURE__;
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_145: [If `send_status` is NULL, messenger_get_send_status() shall fail and return a non-zero value]
- else if (send_status == NULL)
- {
- LogError("messenger_get_send_status failed (send_status is NULL)");
- result = __FAILURE__;
- }
- else
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)messenger_handle;
- LIST_ITEM_HANDLE wts_list_head = singlylinkedlist_get_head_item(instance->waiting_to_send);
- LIST_ITEM_HANDLE ip_list_head = singlylinkedlist_get_head_item(instance->in_progress_list);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_147: [If `instance->in_progress_list` and `instance->wait_to_send_list` are empty, send_status shall be set to MESSENGER_SEND_STATUS_IDLE]
- if (wts_list_head == NULL && ip_list_head == NULL)
- {
- *send_status = MESSENGER_SEND_STATUS_IDLE;
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_148: [Otherwise, send_status shall be set to MESSENGER_SEND_STATUS_BUSY]
- else
- {
- *send_status = MESSENGER_SEND_STATUS_BUSY;
- }
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_149: [If no failures occur, messenger_get_send_status() shall return 0]
- result = RESULT_OK;
- }
-
- return result;
-}
-
-int messenger_start(MESSENGER_HANDLE messenger_handle, SESSION_HANDLE session_handle)
-{
- int result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_029: [If `messenger_handle` is NULL, messenger_start() shall fail and return __FAILURE__]
- if (messenger_handle == NULL)
- {
- result = __FAILURE__;
- LogError("messenger_start failed (messenger_handle is NULL)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_030: [If `session_handle` is NULL, messenger_start() shall fail and return __FAILURE__]
- else if (session_handle == NULL)
- {
- result = __FAILURE__;
- LogError("messenger_start failed (session_handle is NULL)");
- }
- else
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)messenger_handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_031: [If `instance->state` is not MESSENGER_STATE_STOPPED, messenger_start() shall fail and return __FAILURE__]
- if (instance->state != MESSENGER_STATE_STOPPED)
- {
- result = __FAILURE__;
- LogError("messenger_start failed (current state is %d; expected MESSENGER_STATE_STOPPED)", instance->state);
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_032: [`session_handle` shall be saved on `instance->session_handle`]
- instance->session_handle = session_handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_115: [If no failures occurr, `instance->state` shall be set to MESSENGER_STATE_STARTING, and `instance->on_state_changed_callback` invoked if provided]
- update_messenger_state(instance, MESSENGER_STATE_STARTING);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_056: [If no failures occurr, messenger_start() shall return 0]
- result = RESULT_OK;
- }
- }
-
- return result;
-}
-
-int messenger_stop(MESSENGER_HANDLE messenger_handle)
-{
- int result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_057: [If `messenger_handle` is NULL, messenger_stop() shall fail and return a non-zero value]
- if (messenger_handle == NULL)
- {
- result = __FAILURE__;
- LogError("messenger_stop failed (messenger_handle is NULL)");
- }
- else
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)messenger_handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_058: [If `instance->state` is MESSENGER_STATE_STOPPED, messenger_stop() shall fail and return a non-zero value]
- if (instance->state == MESSENGER_STATE_STOPPED)
- {
- result = __FAILURE__;
- LogError("messenger_stop failed (messenger is already stopped)");
- }
- else
- {
- update_messenger_state(instance, MESSENGER_STATE_STOPPING);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_152: [messenger_stop() shall close and destroy `instance->message_sender` and `instance->message_receiver`]
- destroy_event_sender(instance);
- destroy_message_receiver(instance);
-
- remove_timed_out_events(instance);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_162: [messenger_stop() shall move all items from `instance->in_progress_list` to the beginning of `instance->wait_to_send_list`]
- if (move_events_to_wait_to_send_list(instance) != RESULT_OK)
- {
- LogError("Messenger failed to move events in progress back to wait_to_send list");
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_163: [If not all items from `instance->in_progress_list` can be moved back to `instance->wait_to_send_list`, `instance->state` shall be set to MESSENGER_STATE_ERROR, and `instance->on_state_changed_callback` invoked]
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- result = __FAILURE__;
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_164: [If all items get successfuly moved back to `instance->wait_to_send_list`, `instance->state` shall be set to MESSENGER_STATE_STOPPED, and `instance->on_state_changed_callback` invoked]
- update_messenger_state(instance, MESSENGER_STATE_STOPPED);
- result = RESULT_OK;
- }
- }
- }
-
- return result;
-}
-
-// @brief
-// Sets the messenger module state based on the state changes from messagesender and messagereceiver
-static void process_state_changes(MESSENGER_INSTANCE* instance)
-{
- // Note: messagesender and messagereceiver are still not created or already destroyed
- // when state is MESSENGER_STATE_STOPPED, so no checking is needed there.
-
- if (instance->state == MESSENGER_STATE_STARTED)
- {
- if (instance->message_sender_current_state != MESSAGE_SENDER_STATE_OPEN)
- {
- LogError("messagesender reported unexpected state %d while messenger was started", instance->message_sender_current_state);
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- }
- else if (instance->message_receiver != NULL && instance->message_receiver_current_state != MESSAGE_RECEIVER_STATE_OPEN)
- {
- if (instance->message_receiver_current_state == MESSAGE_RECEIVER_STATE_OPENING)
- {
- int is_timed_out;
- if (is_timeout_reached(instance->last_message_receiver_state_change_time, MAX_MESSAGE_RECEIVER_STATE_CHANGE_TIMEOUT_SECS, &is_timed_out) != RESULT_OK)
- {
- LogError("messenger got an error (failed to verify messagereceiver start timeout)");
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- }
- else if (is_timed_out == 1)
- {
- LogError("messenger got an error (messagereceiver failed to start within expected timeout (%d secs))", MAX_MESSAGE_RECEIVER_STATE_CHANGE_TIMEOUT_SECS);
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- }
- }
- else if (instance->message_receiver_current_state == MESSAGE_RECEIVER_STATE_ERROR ||
- instance->message_receiver_current_state == MESSAGE_RECEIVER_STATE_IDLE)
- {
- LogError("messagereceiver reported unexpected state %d while messenger is starting", instance->message_receiver_current_state);
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- }
- }
- }
- else
- {
- if (instance->state == MESSENGER_STATE_STARTING)
- {
- if (instance->message_sender_current_state == MESSAGE_SENDER_STATE_OPEN)
- {
- update_messenger_state(instance, MESSENGER_STATE_STARTED);
- }
- else if (instance->message_sender_current_state == MESSAGE_SENDER_STATE_OPENING)
- {
- int is_timed_out;
- if (is_timeout_reached(instance->last_message_sender_state_change_time, MAX_MESSAGE_SENDER_STATE_CHANGE_TIMEOUT_SECS, &is_timed_out) != RESULT_OK)
- {
- LogError("messenger failed to start (failed to verify messagesender start timeout)");
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- }
- else if (is_timed_out == 1)
- {
- LogError("messenger failed to start (messagesender failed to start within expected timeout (%d secs))", MAX_MESSAGE_SENDER_STATE_CHANGE_TIMEOUT_SECS);
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- }
- }
- // For this module, the only valid scenario where messagesender state is IDLE is if
- // the messagesender hasn't been created yet or already destroyed.
- else if ((instance->message_sender_current_state == MESSAGE_SENDER_STATE_ERROR) ||
- (instance->message_sender_current_state == MESSAGE_SENDER_STATE_CLOSING) ||
- (instance->message_sender_current_state == MESSAGE_SENDER_STATE_IDLE && instance->message_sender != NULL))
- {
- LogError("messagesender reported unexpected state %d while messenger is starting", instance->message_sender_current_state);
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- }
- }
- // message sender and receiver are stopped/destroyed synchronously, so no need for state control.
- }
-}
-
-void messenger_do_work(MESSENGER_HANDLE messenger_handle)
-{
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_065: [If `messenger_handle` is NULL, messenger_do_work() shall fail and return]
- if (messenger_handle == NULL)
- {
- LogError("messenger_do_work failed (messenger_handle is NULL)");
- }
- else
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)messenger_handle;
-
- process_state_changes(instance);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_151: [If `instance->state` is MESSENGER_STATE_STARTING, messenger_do_work() shall create and open `instance->message_sender`]
- if (instance->state == MESSENGER_STATE_STARTING)
- {
- if (instance->message_sender == NULL)
- {
- if (create_event_sender(instance) != RESULT_OK)
- {
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- }
- }
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_066: [If `instance->state` is not MESSENGER_STATE_STARTED, messenger_do_work() shall return]
- else if (instance->state == MESSENGER_STATE_STARTED)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_067: [If `instance->receive_messages` is true and `instance->message_receiver` is NULL, a message_receiver shall be created]
- if (instance->receive_messages == true &&
- instance->message_receiver == NULL &&
- create_message_receiver(instance) != RESULT_OK)
- {
- LogError("messenger_do_work warning (failed creating the message receiver [%s])", STRING_c_str(instance->device_id));
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_092: [If `instance->receive_messages` is false and `instance->message_receiver` is not NULL, it shall be destroyed]
- else if (instance->receive_messages == false && instance->message_receiver != NULL)
- {
- destroy_message_receiver(instance);
- }
-
- if (process_event_send_timeouts(instance) != RESULT_OK)
- {
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- }
- else if (send_pending_events(instance) != RESULT_OK && instance->event_send_retry_limit > 0)
- {
- instance->event_send_error_count++;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_161: [If messenger_do_work() fail sending events for `instance->event_send_retry_limit` times in a row, it shall invoke `instance->on_state_changed_callback`, if provided, with error code MESSENGER_STATE_ERROR]
- if (instance->event_send_error_count >= instance->event_send_retry_limit)
- {
- LogError("messenger_do_work failed (failed sending events; reached max number of consecutive attempts)");
- update_messenger_state(instance, MESSENGER_STATE_ERROR);
- }
- }
- else
- {
- instance->event_send_error_count = 0;
- }
- }
- }
-}
-
-void messenger_destroy(MESSENGER_HANDLE messenger_handle)
-{
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_109: [If `messenger_handle` is NULL, messenger_destroy() shall fail and return]
- if (messenger_handle == NULL)
- {
- LogError("messenger_destroy failed (messenger_handle is NULL)");
- }
- else
- {
- LIST_ITEM_HANDLE list_node;
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)messenger_handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_110: [If the `instance->state` is not MESSENGER_STATE_STOPPED, messenger_destroy() shall invoke messenger_stop()]
- if (instance->state != MESSENGER_STATE_STOPPED)
- {
- (void)messenger_stop(messenger_handle);
- }
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_111: [All elements of `instance->in_progress_list` and `instance->wait_to_send_list` shall be removed, invoking `task->on_event_send_complete_callback` for each with EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED]
-
- // Note: yes messenger_stop() tried to move all events from in_progress_list to wait_to_send_list,
- // but we need to iterate through in case any events failed to be moved.
- while ((list_node = singlylinkedlist_get_head_item(instance->in_progress_list)) != NULL)
- {
- MESSENGER_SEND_EVENT_TASK* task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_node);
-
- (void)singlylinkedlist_remove(instance->in_progress_list, list_node);
-
- if (task != NULL)
- {
- task->on_event_send_complete_callback(task->message, MESSENGER_EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED, (void*)task->context);
- free(task);
- }
- }
-
- while ((list_node = singlylinkedlist_get_head_item(instance->waiting_to_send)) != NULL)
- {
- MESSENGER_SEND_EVENT_TASK* task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_node);
-
- (void)singlylinkedlist_remove(instance->waiting_to_send, list_node);
-
- if (task != NULL)
- {
- task->on_event_send_complete_callback(task->message, MESSENGER_EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED, (void*)task->context);
- free(task);
- }
- }
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_150: [`instance->in_progress_list` and `instance->wait_to_send_list` shall be destroyed using singlylinkedlist_destroy()]
- singlylinkedlist_destroy(instance->waiting_to_send);
- singlylinkedlist_destroy(instance->in_progress_list);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_112: [`instance->iothub_host_fqdn` shall be destroyed using STRING_delete()]
- STRING_delete(instance->iothub_host_fqdn);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_113: [`instance->device_id` shall be destroyed using STRING_delete()]
- STRING_delete(instance->device_id);
-
- STRING_delete(instance->product_info);
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_114: [messenger_destroy() shall destroy `instance` with free()]
- (void)free(instance);
- }
-}
-
-MESSENGER_HANDLE messenger_create(const MESSENGER_CONFIG* messenger_config, const char* product_info)
-{
- MESSENGER_HANDLE handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_001: [If parameter `messenger_config` is NULL, messenger_create() shall return NULL]
- if (messenger_config == NULL)
- {
- handle = NULL;
- LogError("messenger_create failed (messenger_config is NULL)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_002: [If `messenger_config->device_id` is NULL, messenger_create() shall return NULL]
- else if (messenger_config->device_id == NULL)
- {
- handle = NULL;
- LogError("messenger_create failed (messenger_config->device_id is NULL)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_003: [If `messenger_config->iothub_host_fqdn` is NULL, messenger_create() shall return NULL]
- else if (messenger_config->iothub_host_fqdn == NULL)
- {
- handle = NULL;
- LogError("messenger_create failed (messenger_config->iothub_host_fqdn is NULL)");
- }
- else
- {
- MESSENGER_INSTANCE* instance;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_006: [messenger_create() shall allocate memory for the messenger instance structure (aka `instance`)]
- if ((instance = (MESSENGER_INSTANCE*)malloc(sizeof(MESSENGER_INSTANCE))) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_007: [If malloc() fails, messenger_create() shall fail and return NULL]
- handle = NULL;
- LogError("messenger_create failed (messenger_config->wait_to_send_list is NULL)");
- }
- else
- {
- memset(instance, 0, sizeof(MESSENGER_INSTANCE));
- instance->state = MESSENGER_STATE_STOPPED;
- instance->message_sender_current_state = MESSAGE_SENDER_STATE_IDLE;
- instance->message_sender_previous_state = MESSAGE_SENDER_STATE_IDLE;
- instance->message_receiver_current_state = MESSAGE_RECEIVER_STATE_IDLE;
- instance->message_receiver_previous_state = MESSAGE_RECEIVER_STATE_IDLE;
- instance->event_send_retry_limit = DEFAULT_EVENT_SEND_RETRY_LIMIT;
- instance->event_send_timeout_secs = DEFAULT_EVENT_SEND_TIMEOUT_SECS;
- instance->last_message_sender_state_change_time = INDEFINITE_TIME;
- instance->last_message_receiver_state_change_time = INDEFINITE_TIME;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_008: [messenger_create() shall save a copy of `messenger_config->device_id` into `instance->device_id`]
- if ((instance->device_id = STRING_construct(messenger_config->device_id)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_009: [If STRING_construct() fails, messenger_create() shall fail and return NULL]
- handle = NULL;
- LogError("messenger_create failed (device_id could not be copied; STRING_construct failed)");
- }
- else if ((instance->product_info = STRING_construct(product_info)) == NULL)
- {
- handle = NULL;
- LogError("messenger_create failed (product_info could not be copied; STRING_construct failed)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_010: [messenger_create() shall save a copy of `messenger_config->iothub_host_fqdn` into `instance->iothub_host_fqdn`]
- else if ((instance->iothub_host_fqdn = STRING_construct(messenger_config->iothub_host_fqdn)) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_011: [If STRING_construct() fails, messenger_create() shall fail and return NULL]
- handle = NULL;
- LogError("messenger_create failed (iothub_host_fqdn could not be copied; STRING_construct failed)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_165: [`instance->wait_to_send_list` shall be set using singlylinkedlist_create()]
- else if ((instance->waiting_to_send = singlylinkedlist_create()) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_166: [If singlylinkedlist_create() fails, messenger_create() shall fail and return NULL]
- handle = NULL;
- LogError("messenger_create failed (singlylinkedlist_create failed to create wait_to_send_list)");
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_132: [`instance->in_progress_list` shall be set using singlylinkedlist_create()]
- else if ((instance->in_progress_list = singlylinkedlist_create()) == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_133: [If singlylinkedlist_create() fails, messenger_create() shall fail and return NULL]
- handle = NULL;
- LogError("messenger_create failed (singlylinkedlist_create failed to create in_progress_list)");
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_013: [`messenger_config->on_state_changed_callback` shall be saved into `instance->on_state_changed_callback`]
- instance->on_state_changed_callback = messenger_config->on_state_changed_callback;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_014: [`messenger_config->on_state_changed_context` shall be saved into `instance->on_state_changed_context`]
- instance->on_state_changed_context = messenger_config->on_state_changed_context;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_015: [If no failures occurr, messenger_create() shall return a handle to `instance`]
- handle = (MESSENGER_HANDLE)instance;
- }
- }
-
- if (handle == NULL)
- {
- messenger_destroy((MESSENGER_HANDLE)instance);
- }
- }
-
- return handle;
-}
-
-int messenger_set_option(MESSENGER_HANDLE messenger_handle, const char* name, void* value)
-{
- int result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_167: [If `messenger_handle` or `name` or `value` is NULL, messenger_set_option shall fail and return a non-zero value]
- if (messenger_handle == NULL || name == NULL || value == NULL)
- {
- LogError("messenger_set_option failed (one of the followin are NULL: messenger_handle=%p, name=%p, value=%p)",
- messenger_handle, name, value);
- result = __FAILURE__;
- }
- else
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)messenger_handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_168: [If name matches MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS, `value` shall be saved on `instance->event_send_timeout_secs`]
- if (strcmp(MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS, name) == 0)
- {
- instance->event_send_timeout_secs = *((size_t*)value);
- result = RESULT_OK;
- }
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_169: [If name matches MESSENGER_OPTION_SAVED_OPTIONS, `value` shall be applied using OptionHandler_FeedOptions]
- else if (strcmp(MESSENGER_OPTION_SAVED_OPTIONS, name) == 0)
- {
- if (OptionHandler_FeedOptions((OPTIONHANDLER_HANDLE)value, messenger_handle) != OPTIONHANDLER_OK)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_170: [If OptionHandler_FeedOptions fails, messenger_set_option shall fail and return a non-zero value]
- LogError("messenger_set_option failed (OptionHandler_FeedOptions failed)");
- result = __FAILURE__;
- }
- else
- {
- result = RESULT_OK;
- }
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_171: [If name does not match any supported option, authentication_set_option shall fail and return a non-zero value]
- LogError("messenger_set_option failed (option with name '%s' is not suppported)", name);
- result = __FAILURE__;
- }
- }
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_172: [If no errors occur, messenger_set_option shall return 0]
- return result;
-}
-
-OPTIONHANDLER_HANDLE messenger_retrieve_options(MESSENGER_HANDLE messenger_handle)
-{
- OPTIONHANDLER_HANDLE result;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_173: [If `messenger_handle` is NULL, messenger_retrieve_options shall fail and return NULL]
- if (messenger_handle == NULL)
- {
- LogError("Failed to retrieve options from messenger instance (messenger_handle is NULL)");
- result = NULL;
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_174: [An OPTIONHANDLER_HANDLE instance shall be created using OptionHandler_Create]
- OPTIONHANDLER_HANDLE options = OptionHandler_Create(messenger_clone_option, messenger_destroy_option, (pfSetOption)messenger_set_option);
-
- if (options == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_175: [If an OPTIONHANDLER_HANDLE instance fails to be created, messenger_retrieve_options shall fail and return NULL]
- LogError("Failed to retrieve options from messenger instance (OptionHandler_Create failed)");
- result = NULL;
- }
- else
- {
- MESSENGER_INSTANCE* instance = (MESSENGER_INSTANCE*)messenger_handle;
-
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_176: [Each option of `instance` shall be added to the OPTIONHANDLER_HANDLE instance using OptionHandler_AddOption]
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_177: [If OptionHandler_AddOption fails, messenger_retrieve_options shall fail and return NULL]
- if (OptionHandler_AddOption(options, MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS, (void*)&instance->event_send_timeout_secs) != OPTIONHANDLER_OK)
- {
- LogError("Failed to retrieve options from messenger instance (OptionHandler_Create failed for option '%s')", MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS);
- result = NULL;
- }
- else
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_179: [If no failures occur, messenger_retrieve_options shall return the OPTIONHANDLER_HANDLE instance]
- result = options;
- }
-
- if (result == NULL)
- {
- // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_178: [If messenger_retrieve_options fails, any allocated memory shall be freed]
- OptionHandler_Destroy(options);
- }
- }
- }
-
- return result;
-}
--- a/iothubtransport_amqp_messenger.h Fri Jun 02 15:52:19 2017 -0700
+++ /dev/null Thu Jan 01 00:00:00 1970 +0000
@@ -1,91 +0,0 @@
-// Copyright (c) Microsoft. All rights reserved.
-// Licensed under the MIT license. See LICENSE file in the project root for full license information.
-
-#ifndef IOTHUBTRANSPORT_AMQP_MESSENGER
-#define IOTHUBTRANSPORT_AMQP_MESSENGER
-
-#include "azure_c_shared_utility/umock_c_prod.h"
-#include "azure_c_shared_utility/optionhandler.h"
-#include "azure_uamqp_c/session.h"
-#include "iothub_client_private.h"
-
-#ifdef __cplusplus
-extern "C"
-{
-#endif
-
-
-static const char* MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS = "event_send_timeout_secs";
-static const char* MESSENGER_OPTION_SAVED_OPTIONS = "saved_messenger_options";
-
-typedef struct MESSENGER_INSTANCE* MESSENGER_HANDLE;
-
-typedef enum MESSENGER_SEND_STATUS_TAG
-{
- MESSENGER_SEND_STATUS_IDLE,
- MESSENGER_SEND_STATUS_BUSY
-} MESSENGER_SEND_STATUS;
-
-typedef enum MESSENGER_EVENT_SEND_COMPLETE_RESULT_TAG
-{
- MESSENGER_EVENT_SEND_COMPLETE_RESULT_OK,
- MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE,
- MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING,
- MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT,
- MESSENGER_EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED
-} MESSENGER_EVENT_SEND_COMPLETE_RESULT;
-
-typedef enum MESSENGER_DISPOSITION_RESULT_TAG
-{
- MESSENGER_DISPOSITION_RESULT_NONE,
- MESSENGER_DISPOSITION_RESULT_ACCEPTED,
- MESSENGER_DISPOSITION_RESULT_REJECTED,
- MESSENGER_DISPOSITION_RESULT_RELEASED
-} MESSENGER_DISPOSITION_RESULT;
-
-typedef enum MESSENGER_STATE_TAG
-{
- MESSENGER_STATE_STARTING,
- MESSENGER_STATE_STARTED,
- MESSENGER_STATE_STOPPING,
- MESSENGER_STATE_STOPPED,
- MESSENGER_STATE_ERROR
-} MESSENGER_STATE;
-
-typedef struct MESSENGER_MESSAGE_DISPOSITION_INFO_TAG
-{
- delivery_number message_id;
- char* source;
-} MESSENGER_MESSAGE_DISPOSITION_INFO;
-
-typedef void(*ON_MESSENGER_EVENT_SEND_COMPLETE)(IOTHUB_MESSAGE_LIST* iothub_message_list, MESSENGER_EVENT_SEND_COMPLETE_RESULT messenger_event_send_complete_result, void* context);
-typedef void(*ON_MESSENGER_STATE_CHANGED_CALLBACK)(void* context, MESSENGER_STATE previous_state, MESSENGER_STATE new_state);
-typedef MESSENGER_DISPOSITION_RESULT(*ON_MESSENGER_MESSAGE_RECEIVED)(IOTHUB_MESSAGE_HANDLE message, MESSENGER_MESSAGE_DISPOSITION_INFO* disposition_info, void* context);
-
-typedef struct MESSENGER_CONFIG_TAG
-{
- const char* device_id;
- char* iothub_host_fqdn;
- ON_MESSENGER_STATE_CHANGED_CALLBACK on_state_changed_callback;
- void* on_state_changed_context;
-} MESSENGER_CONFIG;
-
-MOCKABLE_FUNCTION(, MESSENGER_HANDLE, messenger_create, const MESSENGER_CONFIG*, messenger_config, const char*, product_info);
-MOCKABLE_FUNCTION(, int, messenger_send_async, MESSENGER_HANDLE, messenger_handle, IOTHUB_MESSAGE_LIST*, message, ON_MESSENGER_EVENT_SEND_COMPLETE, on_messenger_event_send_complete_callback, void*, context);
-MOCKABLE_FUNCTION(, int, messenger_subscribe_for_messages, MESSENGER_HANDLE, messenger_handle, ON_MESSENGER_MESSAGE_RECEIVED, on_message_received_callback, void*, context);
-MOCKABLE_FUNCTION(, int, messenger_unsubscribe_for_messages, MESSENGER_HANDLE, messenger_handle);
-MOCKABLE_FUNCTION(, int, messenger_send_message_disposition, MESSENGER_HANDLE, messenger_handle, MESSENGER_MESSAGE_DISPOSITION_INFO*, disposition_info, MESSENGER_DISPOSITION_RESULT, disposition_result);
-MOCKABLE_FUNCTION(, int, messenger_get_send_status, MESSENGER_HANDLE, messenger_handle, MESSENGER_SEND_STATUS*, send_status);
-MOCKABLE_FUNCTION(, int, messenger_start, MESSENGER_HANDLE, messenger_handle, SESSION_HANDLE, session_handle);
-MOCKABLE_FUNCTION(, int, messenger_stop, MESSENGER_HANDLE, messenger_handle);
-MOCKABLE_FUNCTION(, void, messenger_do_work, MESSENGER_HANDLE, messenger_handle);
-MOCKABLE_FUNCTION(, void, messenger_destroy, MESSENGER_HANDLE, messenger_handle);
-MOCKABLE_FUNCTION(, int, messenger_set_option, MESSENGER_HANDLE, messenger_handle, const char*, name, void*, value);
-MOCKABLE_FUNCTION(, OPTIONHANDLER_HANDLE, messenger_retrieve_options, MESSENGER_HANDLE, messenger_handle);
-
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /*IOTHUBTRANSPORT_AMQP_MESSENGER*/
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/iothubtransport_amqp_telemetry_messenger.c Fri Jun 16 16:12:09 2017 -0700
@@ -0,0 +1,1898 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+#include <stdlib.h>
+#include <stdbool.h>
+#include "azure_c_shared_utility/optimize_size.h"
+#include "azure_c_shared_utility/crt_abstractions.h"
+#include "azure_c_shared_utility/gballoc.h"
+#include "azure_c_shared_utility/agenttime.h"
+#include "azure_c_shared_utility/xlogging.h"
+#include "azure_c_shared_utility/uniqueid.h"
+#include "azure_c_shared_utility/singlylinkedlist.h"
+#include "azure_uamqp_c/link.h"
+#include "azure_uamqp_c/messaging.h"
+#include "azure_uamqp_c/message_sender.h"
+#include "azure_uamqp_c/message_receiver.h"
+#include "uamqp_messaging.h"
+#include "iothub_client_private.h"
+#include "iothub_client_version.h"
+#include "iothubtransport_amqp_telemetry_messenger.h"
+
+#define RESULT_OK 0
+#define INDEFINITE_TIME ((time_t)(-1))
+
+#define IOTHUB_DEVICES_PATH_FMT "%s/devices/%s"
+#define IOTHUB_EVENT_SEND_ADDRESS_FMT "amqps://%s/messages/events"
+#define IOTHUB_MESSAGE_RECEIVE_ADDRESS_FMT "amqps://%s/messages/devicebound"
+#define MESSAGE_SENDER_LINK_NAME_PREFIX "link-snd"
+#define MESSAGE_SENDER_MAX_LINK_SIZE UINT64_MAX
+#define MESSAGE_RECEIVER_LINK_NAME_PREFIX "link-rcv"
+#define MESSAGE_RECEIVER_MAX_LINK_SIZE 65536
+#define DEFAULT_EVENT_SEND_RETRY_LIMIT 10
+#define DEFAULT_EVENT_SEND_TIMEOUT_SECS 600
+#define MAX_MESSAGE_SENDER_STATE_CHANGE_TIMEOUT_SECS 300
+#define MAX_MESSAGE_RECEIVER_STATE_CHANGE_TIMEOUT_SECS 300
+#define UNIQUE_ID_BUFFER_SIZE 37
+#define STRING_NULL_TERMINATOR '\0'
+
+typedef struct TELEMETRY_MESSENGER_INSTANCE_TAG
+{
+ STRING_HANDLE device_id;
+ STRING_HANDLE product_info;
+ STRING_HANDLE iothub_host_fqdn;
+ SINGLYLINKEDLIST_HANDLE waiting_to_send;
+ SINGLYLINKEDLIST_HANDLE in_progress_list;
+ TELEMETRY_MESSENGER_STATE state;
+
+ ON_TELEMETRY_MESSENGER_STATE_CHANGED_CALLBACK on_state_changed_callback;
+ void* on_state_changed_context;
+
+ bool receive_messages;
+ ON_TELEMETRY_MESSENGER_MESSAGE_RECEIVED on_message_received_callback;
+ void* on_message_received_context;
+
+ SESSION_HANDLE session_handle;
+ LINK_HANDLE sender_link;
+ MESSAGE_SENDER_HANDLE message_sender;
+ MESSAGE_SENDER_STATE message_sender_current_state;
+ MESSAGE_SENDER_STATE message_sender_previous_state;
+ LINK_HANDLE receiver_link;
+ MESSAGE_RECEIVER_HANDLE message_receiver;
+ MESSAGE_RECEIVER_STATE message_receiver_current_state;
+ MESSAGE_RECEIVER_STATE message_receiver_previous_state;
+
+ size_t event_send_retry_limit;
+ size_t event_send_error_count;
+ size_t event_send_timeout_secs;
+ time_t last_message_sender_state_change_time;
+ time_t last_message_receiver_state_change_time;
+} TELEMETRY_MESSENGER_INSTANCE;
+
+typedef struct MESSENGER_SEND_EVENT_TASK_TAG
+{
+ IOTHUB_MESSAGE_LIST* message;
+ ON_TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE on_event_send_complete_callback;
+ void* context;
+ time_t send_time;
+ TELEMETRY_MESSENGER_INSTANCE *messenger;
+ bool is_timed_out;
+} MESSENGER_SEND_EVENT_TASK;
+
+// @brief
+// Evaluates if the ammount of time since start_time is greater or lesser than timeout_in_secs.
+// @param is_timed_out
+// Set to 1 if a timeout has been reached, 0 otherwise. Not set if any failure occurs.
+// @returns
+// 0 if no failures occur, non-zero otherwise.
+static int is_timeout_reached(time_t start_time, size_t timeout_in_secs, int *is_timed_out)
+{
+ int result;
+
+ if (start_time == INDEFINITE_TIME)
+ {
+ LogError("Failed to verify timeout (start_time is INDEFINITE)");
+ result = __FAILURE__;
+ }
+ else
+ {
+ time_t current_time;
+
+ if ((current_time = get_time(NULL)) == INDEFINITE_TIME)
+ {
+ LogError("Failed to verify timeout (get_time failed)");
+ result = __FAILURE__;
+ }
+ else
+ {
+ if (get_difftime(current_time, start_time) >= timeout_in_secs)
+ {
+ *is_timed_out = 1;
+ }
+ else
+ {
+ *is_timed_out = 0;
+ }
+
+ result = RESULT_OK;
+ }
+ }
+
+ return result;
+}
+
+static STRING_HANDLE create_devices_path(STRING_HANDLE iothub_host_fqdn, STRING_HANDLE device_id)
+{
+ STRING_HANDLE devices_path;
+
+ if ((devices_path = STRING_new()) == NULL)
+ {
+ LogError("Failed creating devices_path (STRING_new failed)");
+ }
+ else
+ {
+ const char* iothub_host_fqdn_char_ptr = STRING_c_str(iothub_host_fqdn);
+ const char* device_id_char_ptr = STRING_c_str(device_id);
+ if (STRING_sprintf(devices_path, IOTHUB_DEVICES_PATH_FMT, iothub_host_fqdn_char_ptr, device_id_char_ptr) != RESULT_OK)
+ {
+ STRING_delete(devices_path);
+ devices_path = NULL;
+ LogError("Failed creating devices_path (STRING_sprintf failed)");
+ }
+ }
+
+ return devices_path;
+}
+
+static STRING_HANDLE create_event_send_address(STRING_HANDLE devices_path)
+{
+ STRING_HANDLE event_send_address;
+
+ if ((event_send_address = STRING_new()) == NULL)
+ {
+ LogError("Failed creating the event_send_address (STRING_new failed)");
+ }
+ else
+ {
+ const char* devices_path_char_ptr = STRING_c_str(devices_path);
+ if (STRING_sprintf(event_send_address, IOTHUB_EVENT_SEND_ADDRESS_FMT, devices_path_char_ptr) != RESULT_OK)
+ {
+ STRING_delete(event_send_address);
+ event_send_address = NULL;
+ LogError("Failed creating the event_send_address (STRING_sprintf failed)");
+ }
+ }
+
+ return event_send_address;
+}
+
+static STRING_HANDLE create_event_sender_source_name(STRING_HANDLE link_name)
+{
+ STRING_HANDLE source_name;
+
+ if ((source_name = STRING_new()) == NULL)
+ {
+ LogError("Failed creating the source_name (STRING_new failed)");
+ }
+ else
+ {
+ const char* link_name_char_ptr = STRING_c_str(link_name);
+ if (STRING_sprintf(source_name, "%s-source", link_name_char_ptr) != RESULT_OK)
+ {
+ STRING_delete(source_name);
+ source_name = NULL;
+ LogError("Failed creating the source_name (STRING_sprintf failed)");
+ }
+ }
+
+ return source_name;
+}
+
+static STRING_HANDLE create_message_receive_address(STRING_HANDLE devices_path)
+{
+ STRING_HANDLE message_receive_address;
+
+ if ((message_receive_address = STRING_new()) == NULL)
+ {
+ LogError("Failed creating the message_receive_address (STRING_new failed)");
+ }
+ else
+ {
+ const char* devices_path_char_ptr = STRING_c_str(devices_path);
+ if (STRING_sprintf(message_receive_address, IOTHUB_MESSAGE_RECEIVE_ADDRESS_FMT, devices_path_char_ptr) != RESULT_OK)
+ {
+ STRING_delete(message_receive_address);
+ message_receive_address = NULL;
+ LogError("Failed creating the message_receive_address (STRING_sprintf failed)");
+ }
+ }
+
+ return message_receive_address;
+}
+
+static STRING_HANDLE create_message_receiver_target_name(STRING_HANDLE link_name)
+{
+ STRING_HANDLE target_name;
+
+ if ((target_name = STRING_new()) == NULL)
+ {
+ LogError("Failed creating the target_name (STRING_new failed)");
+ }
+ else
+ {
+ const char* link_name_char_ptr = STRING_c_str(link_name);
+ if (STRING_sprintf(target_name, "%s-target", link_name_char_ptr) != RESULT_OK)
+ {
+ STRING_delete(target_name);
+ target_name = NULL;
+ LogError("Failed creating the target_name (STRING_sprintf failed)");
+ }
+ }
+
+ return target_name;
+}
+
+static STRING_HANDLE create_link_name(const char* prefix, const char* infix)
+{
+ char* unique_id;
+ STRING_HANDLE tag = NULL;
+
+ if ((unique_id = (char*)malloc(sizeof(char) * UNIQUE_ID_BUFFER_SIZE + 1)) == NULL)
+ {
+ LogError("Failed generating an unique tag (malloc failed)");
+ }
+ else
+ {
+ memset(unique_id, 0, sizeof(char) * UNIQUE_ID_BUFFER_SIZE + 1);
+
+ if (UniqueId_Generate(unique_id, UNIQUE_ID_BUFFER_SIZE) != UNIQUEID_OK)
+ {
+ LogError("Failed generating an unique tag (UniqueId_Generate failed)");
+ }
+ else if ((tag = STRING_new()) == NULL)
+ {
+ LogError("Failed generating an unique tag (STRING_new failed)");
+ }
+ else if (STRING_sprintf(tag, "%s-%s-%s", prefix, infix, unique_id) != RESULT_OK)
+ {
+ STRING_delete(tag);
+ tag = NULL;
+ LogError("Failed generating an unique tag (STRING_sprintf failed)");
+ }
+
+ free(unique_id);
+ }
+
+ return tag;
+}
+
+static void update_messenger_state(TELEMETRY_MESSENGER_INSTANCE* instance, TELEMETRY_MESSENGER_STATE new_state)
+{
+ if (new_state != instance->state)
+ {
+ TELEMETRY_MESSENGER_STATE previous_state = instance->state;
+ instance->state = new_state;
+
+ if (instance->on_state_changed_callback != NULL)
+ {
+ instance->on_state_changed_callback(instance->on_state_changed_context, previous_state, new_state);
+ }
+ }
+}
+
+static void attach_device_client_type_to_link(LINK_HANDLE link, STRING_HANDLE product_info)
+{
+ fields attach_properties;
+ AMQP_VALUE device_client_type_key_name;
+ AMQP_VALUE device_client_type_value;
+ int result;
+
+ if ((attach_properties = amqpvalue_create_map()) == NULL)
+ {
+ LogError("Failed to create the map for device client type.");
+ }
+ else
+ {
+ if ((device_client_type_key_name = amqpvalue_create_symbol("com.microsoft:client-version")) == NULL)
+ {
+ LogError("Failed to create the key name for the device client type.");
+ }
+ else
+ {
+ if ((device_client_type_value = amqpvalue_create_string(STRING_c_str(product_info))) == NULL)
+ {
+ LogError("Failed to create the key value for the device client type.");
+ }
+ else
+ {
+ if ((result = amqpvalue_set_map_value(attach_properties, device_client_type_key_name, device_client_type_value)) != 0)
+ {
+ LogError("Failed to set the property map for the device client type (error code is: %d)", result);
+ }
+ else if ((result = link_set_attach_properties(link, attach_properties)) != 0)
+ {
+ LogError("Unable to attach the device client type to the link properties (error code is: %d)", result);
+ }
+
+ amqpvalue_destroy(device_client_type_value);
+ }
+
+ amqpvalue_destroy(device_client_type_key_name);
+ }
+
+ amqpvalue_destroy(attach_properties);
+ }
+}
+
+static void destroy_event_sender(TELEMETRY_MESSENGER_INSTANCE* instance)
+{
+ if (instance->message_sender != NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_060: [`instance->message_sender` shall be destroyed using messagesender_destroy()]
+ messagesender_destroy(instance->message_sender);
+ instance->message_sender = NULL;
+ instance->message_sender_current_state = MESSAGE_SENDER_STATE_IDLE;
+ instance->message_sender_previous_state = MESSAGE_SENDER_STATE_IDLE;
+ instance->last_message_sender_state_change_time = INDEFINITE_TIME;
+ }
+
+ if (instance->sender_link != NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_063: [`instance->sender_link` shall be destroyed using link_destroy()]
+ link_destroy(instance->sender_link);
+ instance->sender_link = NULL;
+ }
+}
+
+static void on_event_sender_state_changed_callback(void* context, MESSAGE_SENDER_STATE new_state, MESSAGE_SENDER_STATE previous_state)
+{
+ if (context == NULL)
+ {
+ LogError("on_event_sender_state_changed_callback was invoked with a NULL context; although unexpected, this failure will be ignored");
+ }
+ else
+ {
+ if (new_state != previous_state)
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)context;
+ instance->message_sender_current_state = new_state;
+ instance->message_sender_previous_state = previous_state;
+ instance->last_message_sender_state_change_time = get_time(NULL);
+ }
+ }
+}
+
+static int create_event_sender(TELEMETRY_MESSENGER_INSTANCE* instance)
+{
+ int result;
+
+ STRING_HANDLE link_name = NULL;
+ STRING_HANDLE source_name = NULL;
+ AMQP_VALUE source = NULL;
+ AMQP_VALUE target = NULL;
+ STRING_HANDLE devices_path = NULL;
+ STRING_HANDLE event_send_address = NULL;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_033: [A variable, named `devices_path`, shall be created concatenating `instance->iothub_host_fqdn`, "/devices/" and `instance->device_id`]
+ if ((devices_path = create_devices_path(instance->iothub_host_fqdn, instance->device_id)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_034: [If `devices_path` fails to be created, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message sender (failed creating the 'devices_path')");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_035: [A variable, named `event_send_address`, shall be created concatenating "amqps://", `devices_path` and "/messages/events"]
+ else if ((event_send_address = create_event_send_address(devices_path)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_036: [If `event_send_address` fails to be created, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message sender (failed creating the 'event_send_address')");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_037: [A `link_name` variable shall be created using an unique string label per AMQP session]
+ else if ((link_name = create_link_name(MESSAGE_SENDER_LINK_NAME_PREFIX, STRING_c_str(instance->device_id))) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_038: [If `link_name` fails to be created, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message sender (failed creating an unique link name)");
+ }
+ else if ((source_name = create_event_sender_source_name(link_name)) == NULL)
+ {
+ result = __FAILURE__;
+ LogError("Failed creating the message sender (failed creating an unique source name)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_039: [A `source` variable shall be created with messaging_create_source() using an unique string label per AMQP session]
+ else if ((source = messaging_create_source(STRING_c_str(source_name))) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_040: [If `source` fails to be created, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message sender (messaging_create_source failed)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_041: [A `target` variable shall be created with messaging_create_target() using `event_send_address`]
+ else if ((target = messaging_create_target(STRING_c_str(event_send_address))) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_042: [If `target` fails to be created, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message sender (messaging_create_target failed)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_043: [`instance->sender_link` shall be set using link_create(), passing `instance->session_handle`, `link_name`, "role_sender", `source` and `target` as parameters]
+ else if ((instance->sender_link = link_create(instance->session_handle, STRING_c_str(link_name), role_sender, source, target)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_044: [If link_create() fails, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message sender (link_create failed)");
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_047: [`instance->sender_link` maximum message size shall be set to UINT64_MAX using link_set_max_message_size()]
+ if (link_set_max_message_size(instance->sender_link, MESSAGE_SENDER_MAX_LINK_SIZE) != RESULT_OK)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_048: [If link_set_max_message_size() fails, it shall be logged and ignored.]
+ LogError("Failed setting message sender link max message size.");
+ }
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_049: [`instance->sender_link` should have a property "com.microsoft:client-version" set as `CLIENT_DEVICE_TYPE_PREFIX/IOTHUB_SDK_VERSION`, using amqpvalue_set_map_value() and link_set_attach_properties()]
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_050: [If amqpvalue_set_map_value() or link_set_attach_properties() fail, the failure shall be ignored]
+ attach_device_client_type_to_link(instance->sender_link, instance->product_info);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_051: [`instance->message_sender` shall be created using messagesender_create(), passing the `instance->sender_link` and `on_event_sender_state_changed_callback`]
+ if ((instance->message_sender = messagesender_create(instance->sender_link, on_event_sender_state_changed_callback, (void*)instance)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_052: [If messagesender_create() fails, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ link_destroy(instance->sender_link);
+ instance->sender_link = NULL;
+ LogError("Failed creating the message sender (messagesender_create failed)");
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_053: [`instance->message_sender` shall be opened using messagesender_open()]
+ if (messagesender_open(instance->message_sender) != RESULT_OK)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_054: [If messagesender_open() fails, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ messagesender_destroy(instance->message_sender);
+ instance->message_sender = NULL;
+ link_destroy(instance->sender_link);
+ instance->sender_link = NULL;
+ LogError("Failed opening the AMQP message sender.");
+ }
+ else
+ {
+ result = RESULT_OK;
+ }
+ }
+ }
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_055: [Before returning, telemetry_messenger_do_work() shall release all the temporary memory it has allocated]
+ if (link_name != NULL)
+ STRING_delete(link_name);
+ if (source_name != NULL)
+ STRING_delete(source_name);
+ if (source != NULL)
+ amqpvalue_destroy(source);
+ if (target != NULL)
+ amqpvalue_destroy(target);
+ if (devices_path != NULL)
+ STRING_delete(devices_path);
+ if (event_send_address != NULL)
+ STRING_delete(event_send_address);
+
+ return result;
+}
+
+static void destroy_message_receiver(TELEMETRY_MESSENGER_INSTANCE* instance)
+{
+ if (instance->message_receiver != NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_061: [`instance->message_receiver` shall be closed using messagereceiver_close()]
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_093: [`instance->message_receiver` shall be closed using messagereceiver_close()]
+ if (messagereceiver_close(instance->message_receiver) != RESULT_OK)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_094: [If messagereceiver_close() fails, it shall be logged and ignored]
+ LogError("Failed closing the AMQP message receiver (this failure will be ignored).");
+ }
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_062: [`instance->message_receiver` shall be destroyed using messagereceiver_destroy()]
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_095: [`instance->message_receiver` shall be destroyed using messagereceiver_destroy()]
+ messagereceiver_destroy(instance->message_receiver);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_096: [`instance->message_receiver` shall be set to NULL]
+ instance->message_receiver = NULL;
+ instance->message_receiver_current_state = MESSAGE_RECEIVER_STATE_IDLE;
+ instance->message_receiver_previous_state = MESSAGE_RECEIVER_STATE_IDLE;
+ instance->last_message_receiver_state_change_time = INDEFINITE_TIME;
+ }
+
+ if (instance->receiver_link != NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_064: [`instance->receiver_link` shall be destroyed using link_destroy()]
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_097: [`instance->receiver_link` shall be destroyed using link_destroy()]
+ link_destroy(instance->receiver_link);
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_098: [`instance->receiver_link` shall be set to NULL]
+ instance->receiver_link = NULL;
+ }
+}
+
+static void on_message_receiver_state_changed_callback(const void* context, MESSAGE_RECEIVER_STATE new_state, MESSAGE_RECEIVER_STATE previous_state)
+{
+ if (context == NULL)
+ {
+ LogError("on_message_receiver_state_changed_callback was invoked with a NULL context; although unexpected, this failure will be ignored");
+ }
+ else
+ {
+ if (new_state != previous_state)
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)context;
+ instance->message_receiver_current_state = new_state;
+ instance->message_receiver_previous_state = previous_state;
+ instance->last_message_receiver_state_change_time = get_time(NULL);
+ }
+ }
+}
+
+static TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* create_message_disposition_info(TELEMETRY_MESSENGER_INSTANCE* messenger)
+{
+ TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* result;
+
+ if ((result = (TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO*)malloc(sizeof(TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO))) == NULL)
+ {
+ LogError("Failed creating TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO container (malloc failed)");
+ result = NULL;
+ }
+ else
+ {
+ delivery_number message_id;
+
+ if (messagereceiver_get_received_message_id(messenger->message_receiver, &message_id) != RESULT_OK)
+ {
+ LogError("Failed creating TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO container (messagereceiver_get_received_message_id failed)");
+ free(result);
+ result = NULL;
+ }
+ else
+ {
+ const char* link_name;
+
+ if (messagereceiver_get_link_name(messenger->message_receiver, &link_name) != RESULT_OK)
+ {
+ LogError("Failed creating TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO container (messagereceiver_get_link_name failed)");
+ free(result);
+ result = NULL;
+ }
+ else if (mallocAndStrcpy_s(&result->source, link_name) != RESULT_OK)
+ {
+ LogError("Failed creating TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO container (failed copying link name)");
+ free(result);
+ result = NULL;
+ }
+ else
+ {
+ result->message_id = message_id;
+ }
+ }
+ }
+
+ return result;
+}
+
+static void destroy_message_disposition_info(TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* disposition_info)
+{
+ free(disposition_info->source);
+ free(disposition_info);
+}
+
+static AMQP_VALUE create_uamqp_disposition_result_from(TELEMETRY_MESSENGER_DISPOSITION_RESULT disposition_result)
+{
+ AMQP_VALUE uamqp_disposition_result;
+
+ if (disposition_result == TELEMETRY_MESSENGER_DISPOSITION_RESULT_NONE)
+ {
+ uamqp_disposition_result = NULL; // intentionally not sending an answer.
+ }
+ else if (disposition_result == TELEMETRY_MESSENGER_DISPOSITION_RESULT_ACCEPTED)
+ {
+ uamqp_disposition_result = messaging_delivery_accepted();
+ }
+ else if (disposition_result == TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED)
+ {
+ uamqp_disposition_result = messaging_delivery_released();
+ }
+ else if (disposition_result == TELEMETRY_MESSENGER_DISPOSITION_RESULT_REJECTED)
+ {
+ uamqp_disposition_result = messaging_delivery_rejected("Rejected by application", "Rejected by application");
+ }
+ else
+ {
+ LogError("Failed creating a disposition result for messagereceiver (result %d is not supported)", disposition_result);
+ uamqp_disposition_result = NULL;
+ }
+
+ return uamqp_disposition_result;
+}
+
+static AMQP_VALUE on_message_received_internal_callback(const void* context, MESSAGE_HANDLE message)
+{
+ AMQP_VALUE result;
+ int api_call_result;
+ IOTHUB_MESSAGE_HANDLE iothub_message;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_121: [An IOTHUB_MESSAGE_HANDLE shall be obtained from MESSAGE_HANDLE using IoTHubMessage_CreateFromUamqpMessage()]
+ if ((api_call_result = IoTHubMessage_CreateFromUamqpMessage(message, &iothub_message)) != RESULT_OK)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_122: [If IoTHubMessage_CreateFromUamqpMessage() fails, on_message_received_internal_callback shall return the result of messaging_delivery_rejected()]
+ result = messaging_delivery_rejected("Rejected due to failure reading AMQP message", "Failed reading AMQP message");
+
+ LogError("on_message_received_internal_callback failed (IoTHubMessage_CreateFromUamqpMessage; error = %d).", api_call_result);
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)context;
+ TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* message_disposition_info;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_186: [A TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO instance shall be created containing the source link name and message delivery ID]
+ if ((message_disposition_info = create_message_disposition_info(instance)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_187: [**If the TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO instance fails to be created, on_message_received_internal_callback shall return messaging_delivery_released()]
+ LogError("on_message_received_internal_callback failed (failed creating TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO).");
+ result = messaging_delivery_released();
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_123: [`instance->on_message_received_callback` shall be invoked passing the IOTHUB_MESSAGE_HANDLE and TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO instance]
+ TELEMETRY_MESSENGER_DISPOSITION_RESULT disposition_result = instance->on_message_received_callback(iothub_message, message_disposition_info, instance->on_message_received_context);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_188: [The memory allocated for the TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO instance shall be released]
+ destroy_message_disposition_info(message_disposition_info);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_125: [If `instance->on_message_received_callback` returns TELEMETRY_MESSENGER_DISPOSITION_RESULT_ACCEPTED, on_message_received_internal_callback shall return the result of messaging_delivery_accepted()]
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_126: [If `instance->on_message_received_callback` returns TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED, on_message_received_internal_callback shall return the result of messaging_delivery_released()]
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_127: [If `instance->on_message_received_callback` returns TELEMETRY_MESSENGER_DISPOSITION_RESULT_REJECTED, on_message_received_internal_callback shall return the result of messaging_delivery_rejected()]
+ result = create_uamqp_disposition_result_from(disposition_result);
+ }
+ }
+
+ return result;
+}
+
+static int create_message_receiver(TELEMETRY_MESSENGER_INSTANCE* instance)
+{
+ int result;
+
+ STRING_HANDLE devices_path = NULL;
+ STRING_HANDLE message_receive_address = NULL;
+ STRING_HANDLE link_name = NULL;
+ STRING_HANDLE target_name = NULL;
+ AMQP_VALUE source = NULL;
+ AMQP_VALUE target = NULL;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_068: [A variable, named `devices_path`, shall be created concatenating `instance->iothub_host_fqdn`, "/devices/" and `instance->device_id`]
+ if ((devices_path = create_devices_path(instance->iothub_host_fqdn, instance->device_id)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_069: [If `devices_path` fails to be created, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message receiver (failed creating the 'devices_path')");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_070: [A variable, named `message_receive_address`, shall be created concatenating "amqps://", `devices_path` and "/messages/devicebound"]
+ else if ((message_receive_address = create_message_receive_address(devices_path)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_071: [If `message_receive_address` fails to be created, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message receiver (failed creating the 'message_receive_address')");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_072: [A `link_name` variable shall be created using an unique string label per AMQP session]
+ else if ((link_name = create_link_name(MESSAGE_RECEIVER_LINK_NAME_PREFIX, STRING_c_str(instance->device_id))) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_073: [If `link_name` fails to be created, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message receiver (failed creating an unique link name)");
+ }
+ else if ((target_name = create_message_receiver_target_name(link_name)) == NULL)
+ {
+ result = __FAILURE__;
+ LogError("Failed creating the message receiver (failed creating an unique target name)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_074: [A `target` variable shall be created with messaging_create_target() using an unique string label per AMQP session]
+ else if ((target = messaging_create_target(STRING_c_str(target_name))) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_075: [If `target` fails to be created, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message receiver (messaging_create_target failed)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_076: [A `source` variable shall be created with messaging_create_source() using `message_receive_address`]
+ else if ((source = messaging_create_source(STRING_c_str(message_receive_address))) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_077: [If `source` fails to be created, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message receiver (messaging_create_source failed)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_078: [`instance->receiver_link` shall be set using link_create(), passing `instance->session_handle`, `link_name`, "role_receiver", `source` and `target` as parameters]
+ else if ((instance->receiver_link = link_create(instance->session_handle, STRING_c_str(link_name), role_receiver, source, target)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_079: [If link_create() fails, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message receiver (link_create failed)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_080: [`instance->receiver_link` settle mode shall be set to "receiver_settle_mode_first" using link_set_rcv_settle_mode(), ]
+ else if (link_set_rcv_settle_mode(instance->receiver_link, receiver_settle_mode_first) != RESULT_OK)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_081: [If link_set_rcv_settle_mode() fails, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ LogError("Failed creating the message receiver (link_set_rcv_settle_mode failed)");
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_082: [`instance->receiver_link` maximum message size shall be set to 65536 using link_set_max_message_size()]
+ if (link_set_max_message_size(instance->receiver_link, MESSAGE_RECEIVER_MAX_LINK_SIZE) != RESULT_OK)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_083: [If link_set_max_message_size() fails, it shall be logged and ignored.]
+ LogError("Failed setting message receiver link max message size.");
+ }
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_084: [`instance->receiver_link` should have a property "com.microsoft:client-version" set as `CLIENT_DEVICE_TYPE_PREFIX/IOTHUB_SDK_VERSION`, using amqpvalue_set_map_value() and link_set_attach_properties()]
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_085: [If amqpvalue_set_map_value() or link_set_attach_properties() fail, the failure shall be ignored]
+ attach_device_client_type_to_link(instance->receiver_link, instance->product_info);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_086: [`instance->message_receiver` shall be created using messagereceiver_create(), passing the `instance->receiver_link` and `on_messagereceiver_state_changed_callback`]
+ if ((instance->message_receiver = messagereceiver_create(instance->receiver_link, on_message_receiver_state_changed_callback, (void*)instance)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_087: [If messagereceiver_create() fails, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ link_destroy(instance->receiver_link);
+ LogError("Failed creating the message receiver (messagereceiver_create failed)");
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_088: [`instance->message_receiver` shall be opened using messagereceiver_open(), passing `on_message_received_internal_callback`]
+ if (messagereceiver_open(instance->message_receiver, on_message_received_internal_callback, (void*)instance) != RESULT_OK)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_089: [If messagereceiver_open() fails, telemetry_messenger_do_work() shall fail and return]
+ result = __FAILURE__;
+ messagereceiver_destroy(instance->message_receiver);
+ link_destroy(instance->receiver_link);
+ LogError("Failed opening the AMQP message receiver.");
+ }
+ else
+ {
+ result = RESULT_OK;
+ }
+ }
+ }
+
+ if (devices_path != NULL)
+ STRING_delete(devices_path);
+ if (message_receive_address != NULL)
+ STRING_delete(message_receive_address);
+ if (link_name != NULL)
+ STRING_delete(link_name);
+ if (target_name != NULL)
+ STRING_delete(target_name);
+ if (source != NULL)
+ amqpvalue_destroy(source);
+ if (target != NULL)
+ amqpvalue_destroy(target);
+
+ return result;
+}
+
+static int move_event_to_in_progress_list(MESSENGER_SEND_EVENT_TASK* task)
+{
+ int result;
+
+ if (singlylinkedlist_add(task->messenger->in_progress_list, (void*)task) == NULL)
+ {
+ result = __FAILURE__;
+ LogError("Failed moving event to in_progress list (singlylinkedlist_add failed)");
+ }
+ else
+ {
+ result = RESULT_OK;
+ }
+
+ return result;
+}
+
+static bool find_MESSENGER_SEND_EVENT_TASK_on_list(LIST_ITEM_HANDLE list_item, const void* match_context)
+{
+ return (list_item != NULL && singlylinkedlist_item_get_value(list_item) == match_context);
+}
+
+static void remove_event_from_in_progress_list(MESSENGER_SEND_EVENT_TASK *task)
+{
+ LIST_ITEM_HANDLE list_item = singlylinkedlist_find(task->messenger->in_progress_list, find_MESSENGER_SEND_EVENT_TASK_on_list, (void*)task);
+
+ if (list_item != NULL)
+ {
+ if (singlylinkedlist_remove(task->messenger->in_progress_list, list_item) != RESULT_OK)
+ {
+ LogError("Failed removing event from in_progress list (singlylinkedlist_remove failed)");
+ }
+ }
+}
+
+static int copy_events_to_list(SINGLYLINKEDLIST_HANDLE from_list, SINGLYLINKEDLIST_HANDLE to_list)
+{
+ int result;
+ LIST_ITEM_HANDLE list_item;
+
+ result = RESULT_OK;
+ list_item = singlylinkedlist_get_head_item(from_list);
+
+ while (list_item != NULL)
+ {
+ MESSENGER_SEND_EVENT_TASK *task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_item);
+
+ if (singlylinkedlist_add(to_list, task) == NULL)
+ {
+ LogError("Failed copying event to destination list (singlylinkedlist_add failed)");
+ result = __FAILURE__;
+ break;
+ }
+ else
+ {
+ list_item = singlylinkedlist_get_next_item(list_item);
+ }
+ }
+
+ return result;
+}
+
+static int singlylinkedlist_clear(SINGLYLINKEDLIST_HANDLE list)
+{
+ int result;
+ LIST_ITEM_HANDLE list_item;
+
+ result = RESULT_OK;
+
+ while ((list_item = singlylinkedlist_get_head_item(list)) != NULL)
+ {
+ if (singlylinkedlist_remove(list, list_item) != RESULT_OK)
+ {
+ LogError("Failed removing items from list (%d)", list);
+ result = __FAILURE__;
+ break;
+ }
+ }
+
+ return result;
+}
+
+static int move_events_to_wait_to_send_list(TELEMETRY_MESSENGER_INSTANCE* instance)
+{
+ int result;
+ LIST_ITEM_HANDLE list_item;
+
+ if ((list_item = singlylinkedlist_get_head_item(instance->in_progress_list)) == NULL)
+ {
+ result = RESULT_OK;
+ }
+ else
+ {
+ SINGLYLINKEDLIST_HANDLE new_wait_to_send_list;
+
+ if ((new_wait_to_send_list = singlylinkedlist_create()) == NULL)
+ {
+ LogError("Failed moving events back to wait_to_send list (singlylinkedlist_create failed to create new wait_to_send_list)");
+ result = __FAILURE__;
+ }
+ else
+ {
+ SINGLYLINKEDLIST_HANDLE new_in_progress_list;
+
+ if (copy_events_to_list(instance->in_progress_list, new_wait_to_send_list) != RESULT_OK)
+ {
+ LogError("Failed moving events back to wait_to_send list (failed adding in_progress_list items to new_wait_to_send_list)");
+ singlylinkedlist_destroy(new_wait_to_send_list);
+ result = __FAILURE__;
+ }
+ else if (copy_events_to_list(instance->waiting_to_send, new_wait_to_send_list) != RESULT_OK)
+ {
+ LogError("Failed moving events back to wait_to_send list (failed adding wait_to_send items to new_wait_to_send_list)");
+ singlylinkedlist_destroy(new_wait_to_send_list);
+ result = __FAILURE__;
+ }
+ else if ((new_in_progress_list = singlylinkedlist_create()) == NULL)
+ {
+ LogError("Failed moving events back to wait_to_send list (singlylinkedlist_create failed to create new in_progress_list)");
+ singlylinkedlist_destroy(new_wait_to_send_list);
+ result = __FAILURE__;
+ }
+ else
+ {
+ singlylinkedlist_destroy(instance->waiting_to_send);
+ singlylinkedlist_destroy(instance->in_progress_list);
+ instance->waiting_to_send = new_wait_to_send_list;
+ instance->in_progress_list = new_in_progress_list;
+ result = RESULT_OK;
+ }
+ }
+ }
+
+ return result;
+}
+
+static void internal_on_event_send_complete_callback(void* context, MESSAGE_SEND_RESULT send_result)
+{
+ if (context != NULL)
+ {
+ MESSENGER_SEND_EVENT_TASK* task = (MESSENGER_SEND_EVENT_TASK*)context;
+
+ if (task->messenger->message_sender_current_state != MESSAGE_SENDER_STATE_ERROR)
+ {
+ if (task->is_timed_out == false)
+ {
+ TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT messenger_send_result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_107: [If no failure occurs, `task->on_event_send_complete_callback` shall be invoked with result EVENT_SEND_COMPLETE_RESULT_OK]
+ if (send_result == MESSAGE_SEND_OK)
+ {
+ messenger_send_result = TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_OK;
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_108: [If a failure occurred, `task->on_event_send_complete_callback` shall be invoked with result EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING]
+ else
+ {
+ messenger_send_result = TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING;
+ }
+
+ task->on_event_send_complete_callback(task->message, messenger_send_result, (void*)task->context);
+ }
+ else
+ {
+ LogInfo("messenger on_event_send_complete_callback invoked for timed out event %p; not firing upper layer callback.", task->message);
+ }
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_128: [`task` shall be removed from `instance->in_progress_list`]
+ remove_event_from_in_progress_list(task);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_130: [`task` shall be destroyed using free()]
+ free(task);
+ }
+ }
+}
+
+static MESSENGER_SEND_EVENT_TASK* get_next_event_to_send(TELEMETRY_MESSENGER_INSTANCE* instance)
+{
+ MESSENGER_SEND_EVENT_TASK* task;
+ LIST_ITEM_HANDLE list_item;
+
+ if ((list_item = singlylinkedlist_get_head_item(instance->waiting_to_send)) == NULL)
+ {
+ task = NULL;
+ }
+ else
+ {
+ task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_item);
+
+ if (singlylinkedlist_remove(instance->waiting_to_send, list_item) != RESULT_OK)
+ {
+ LogError("Failed removing item from waiting_to_send list (singlylinkedlist_remove failed)");
+ }
+ }
+
+ return task;
+}
+
+static int send_pending_events(TELEMETRY_MESSENGER_INSTANCE* instance)
+{
+ int result = RESULT_OK;
+
+ MESSENGER_SEND_EVENT_TASK* task;
+
+ while ((task = get_next_event_to_send(instance)) != NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_153: [telemetry_messenger_do_work() shall move each event to be sent from `instance->wait_to_send_list` to `instance->in_progress_list`]
+ if (move_event_to_in_progress_list(task) != RESULT_OK)
+ {
+ result = __FAILURE__;
+ task->on_event_send_complete_callback(task->message, TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING, (void*)task->context);
+ break;
+ }
+ else
+ {
+ int uamqp_result;
+ MESSAGE_HANDLE amqp_message = NULL;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_154: [A MESSAGE_HANDLE shall be obtained out of the event's IOTHUB_MESSAGE_HANDLE instance by using message_create_from_iothub_message()]
+ if ((uamqp_result = message_create_from_iothub_message(task->message->messageHandle, &amqp_message)) != RESULT_OK)
+ {
+ LogError("Failed sending event message (failed creating AMQP message; error: %d).", uamqp_result);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_155: [If message_create_from_iothub_message() fails, `task->on_event_send_complete_callback` shall be invoked with result EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE]
+ task->on_event_send_complete_callback(task->message, TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE, (void*)task->context);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_160: [If any failure occurs the event shall be removed from `instance->in_progress_list` and destroyed]
+ remove_event_from_in_progress_list(task);
+ free(task);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_156: [If message_create_from_iothub_message() fails, telemetry_messenger_do_work() shall skip to the next event to be sent]
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_157: [The MESSAGE_HANDLE shall be submitted for sending using messagesender_send(), passing `internal_on_event_send_complete_callback`]
+ uamqp_result = messagesender_send(instance->message_sender, amqp_message, internal_on_event_send_complete_callback, task);
+ task->send_time = get_time(NULL);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_159: [The MESSAGE_HANDLE shall be destroyed using message_destroy().]
+ message_destroy(amqp_message);
+
+ if (uamqp_result != RESULT_OK)
+ {
+ LogError("Failed sending event (messagesender_send failed; error: %d)", uamqp_result);
+
+ result = __FAILURE__;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_158: [If messagesender_send() fails, `task->on_event_send_complete_callback` shall be invoked with result EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING]
+ task->on_event_send_complete_callback(task->message, TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING, (void*)task->context);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_160: [If any failure occurs the event shall be removed from `instance->in_progress_list` and destroyed]
+ remove_event_from_in_progress_list(task);
+ free(task);
+
+ break;
+ }
+ }
+ }
+ }
+
+ return result;
+}
+
+// @brief
+// Goes through each task in in_progress_list and checks if the events timed out to be sent.
+// @remarks
+// If an event is timed out, it is marked as such but not removed, and the upper layer callback is invoked.
+// @returns
+// 0 if no failures occur, non-zero otherwise.
+static int process_event_send_timeouts(TELEMETRY_MESSENGER_INSTANCE* instance)
+{
+ int result = RESULT_OK;
+
+ if (instance->event_send_timeout_secs > 0)
+ {
+ LIST_ITEM_HANDLE list_item = singlylinkedlist_get_head_item(instance->in_progress_list);
+
+ while (list_item != NULL)
+ {
+ MESSENGER_SEND_EVENT_TASK* task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_item);
+
+ if (task->is_timed_out == false)
+ {
+ int is_timed_out;
+
+ if (is_timeout_reached(task->send_time, instance->event_send_timeout_secs, &is_timed_out) == RESULT_OK)
+ {
+ if (is_timed_out)
+ {
+ task->is_timed_out = true;
+
+ if (task->on_event_send_complete_callback != NULL)
+ {
+ task->on_event_send_complete_callback(task->message, TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT, task->context);
+ }
+ }
+ }
+ else
+ {
+ LogError("messenger failed to evaluate event send timeout of event %d", task->message);
+ result = __FAILURE__;
+ }
+ }
+
+ list_item = singlylinkedlist_get_next_item(list_item);
+ }
+ }
+
+ return result;
+}
+
+// @brief
+// Removes all the timed out events from the in_progress_list, without invoking callbacks or detroying the messages.
+static void remove_timed_out_events(TELEMETRY_MESSENGER_INSTANCE* instance)
+{
+ LIST_ITEM_HANDLE list_item = singlylinkedlist_get_head_item(instance->in_progress_list);
+
+ while (list_item != NULL)
+ {
+ MESSENGER_SEND_EVENT_TASK* task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_item);
+
+ if (task->is_timed_out == true)
+ {
+ remove_event_from_in_progress_list(task);
+
+ free(task);
+ }
+
+ list_item = singlylinkedlist_get_next_item(list_item);
+ }
+}
+
+
+// ---------- Set/Retrieve Options Helpers ----------//
+
+static void* telemetry_messenger_clone_option(const char* name, const void* value)
+{
+ void* result;
+
+ if (name == NULL)
+ {
+ LogError("Failed to clone messenger option (name is NULL)");
+ result = NULL;
+ }
+ else if (value == NULL)
+ {
+ LogError("Failed to clone messenger option (value is NULL)");
+ result = NULL;
+ }
+ else
+ {
+ if (strcmp(MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS, name) == 0 ||
+ strcmp(MESSENGER_OPTION_SAVED_OPTIONS, name) == 0)
+ {
+ result = (void*)value;
+ }
+ else
+ {
+ LogError("Failed to clone messenger option (option with name '%s' is not suppported)", name);
+ result = NULL;
+ }
+ }
+
+ return result;
+}
+
+static void telemetry_messenger_destroy_option(const char* name, const void* value)
+{
+ if (name == NULL)
+ {
+ LogError("Failed to destroy messenger option (name is NULL)");
+ }
+ else if (value == NULL)
+ {
+ LogError("Failed to destroy messenger option (value is NULL)");
+ }
+ else
+ {
+ // Nothing to be done for the supported options.
+ }
+}
+
+
+// Public API:
+
+int telemetry_messenger_subscribe_for_messages(TELEMETRY_MESSENGER_HANDLE messenger_handle, ON_TELEMETRY_MESSENGER_MESSAGE_RECEIVED on_message_received_callback, void* context)
+{
+ int result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_016: [If `messenger_handle` is NULL, telemetry_messenger_subscribe_for_messages() shall fail and return __FAILURE__]
+ if (messenger_handle == NULL)
+ {
+ result = __FAILURE__;
+ LogError("telemetry_messenger_subscribe_for_messages failed (messenger_handle is NULL)");
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_017: [If `instance->receive_messages` is already true, telemetry_messenger_subscribe_for_messages() shall fail and return __FAILURE__]
+ if (instance->receive_messages)
+ {
+ result = __FAILURE__;
+ LogError("telemetry_messenger_subscribe_for_messages failed (messenger already subscribed)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_018: [If `on_message_received_callback` is NULL, telemetry_messenger_subscribe_for_messages() shall fail and return __FAILURE__]
+ else if (on_message_received_callback == NULL)
+ {
+ result = __FAILURE__;
+ LogError("telemetry_messenger_subscribe_for_messages failed (on_message_received_callback is NULL)");
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_019: [`on_message_received_callback` shall be saved on `instance->on_message_received_callback`]
+ instance->on_message_received_callback = on_message_received_callback;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_020: [`context` shall be saved on `instance->on_message_received_context`]
+ instance->on_message_received_context = context;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_021: [telemetry_messenger_subscribe_for_messages() shall set `instance->receive_messages` to true]
+ instance->receive_messages = true;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_022: [If no failures occurr, telemetry_messenger_subscribe_for_messages() shall return 0]
+ result = RESULT_OK;
+ }
+ }
+
+ return result;
+}
+
+int telemetry_messenger_unsubscribe_for_messages(TELEMETRY_MESSENGER_HANDLE messenger_handle)
+{
+ int result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_023: [If `messenger_handle` is NULL, telemetry_messenger_unsubscribe_for_messages() shall fail and return __FAILURE__]
+ if (messenger_handle == NULL)
+ {
+ result = __FAILURE__;
+ LogError("telemetry_messenger_unsubscribe_for_messages failed (messenger_handle is NULL)");
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_024: [If `instance->receive_messages` is already false, telemetry_messenger_unsubscribe_for_messages() shall fail and return __FAILURE__]
+ if (instance->receive_messages == false)
+ {
+ result = __FAILURE__;
+ LogError("telemetry_messenger_unsubscribe_for_messages failed (messenger is not subscribed)");
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_025: [telemetry_messenger_unsubscribe_for_messages() shall set `instance->receive_messages` to false]
+ instance->receive_messages = false;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_026: [telemetry_messenger_unsubscribe_for_messages() shall set `instance->on_message_received_callback` to NULL]
+ instance->on_message_received_callback = NULL;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_027: [telemetry_messenger_unsubscribe_for_messages() shall set `instance->on_message_received_context` to NULL]
+ instance->on_message_received_context = NULL;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_028: [If no failures occurr, telemetry_messenger_unsubscribe_for_messages() shall return 0]
+ result = RESULT_OK;
+ }
+ }
+
+ return result;
+}
+
+int telemetry_messenger_send_message_disposition(TELEMETRY_MESSENGER_HANDLE messenger_handle, TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* disposition_info, TELEMETRY_MESSENGER_DISPOSITION_RESULT disposition_result)
+{
+ int result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_179: [If `messenger_handle` or `disposition_info` are NULL, telemetry_messenger_send_message_disposition() shall fail and return __FAILURE__]
+ if (messenger_handle == NULL || disposition_info == NULL)
+ {
+ LogError("Failed sending message disposition (either messenger_handle (%p) or disposition_info (%p) are NULL)", messenger_handle, disposition_info);
+ result = __FAILURE__;
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_180: [If `disposition_info->source` is NULL, telemetry_messenger_send_message_disposition() shall fail and return __FAILURE__]
+ else if (disposition_info->source == NULL)
+ {
+ LogError("Failed sending message disposition (disposition_info->source is NULL)");
+ result = __FAILURE__;
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* messenger = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_189: [If `messenger_handle->message_receiver` is NULL, telemetry_messenger_send_message_disposition() shall fail and return __FAILURE__]
+ if (messenger->message_receiver == NULL)
+ {
+ LogError("Failed sending message disposition (message_receiver is not created; check if it is subscribed)");
+ result = __FAILURE__;
+ }
+ else
+ {
+ AMQP_VALUE uamqp_disposition_result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_181: [An AMQP_VALUE disposition result shall be created corresponding to the `disposition_result` provided]
+ if ((uamqp_disposition_result = create_uamqp_disposition_result_from(disposition_result)) == NULL)
+ {
+ LogError("Failed sending message disposition (disposition result %d is not supported)", disposition_result);
+ result = __FAILURE__;
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_182: [`messagereceiver_send_message_disposition()` shall be invoked passing `disposition_info->source`, `disposition_info->message_id` and the corresponding AMQP_VALUE disposition result]
+ if (messagereceiver_send_message_disposition(messenger->message_receiver, disposition_info->source, disposition_info->message_id, uamqp_disposition_result) != RESULT_OK)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_183: [If `messagereceiver_send_message_disposition()` fails, telemetry_messenger_send_message_disposition() shall fail and return __FAILURE__]
+ LogError("Failed sending message disposition (messagereceiver_send_message_disposition failed)");
+ result = __FAILURE__;
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_185: [If no failures occurr, telemetry_messenger_send_message_disposition() shall return 0]
+ result = RESULT_OK;
+ }
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_184: [telemetry_messenger_send_message_disposition() shall destroy the AMQP_VALUE disposition result]
+ amqpvalue_destroy(uamqp_disposition_result);
+ }
+ }
+ }
+
+ return result;
+}
+
+int telemetry_messenger_send_async(TELEMETRY_MESSENGER_HANDLE messenger_handle, IOTHUB_MESSAGE_LIST* message, ON_TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE on_messenger_event_send_complete_callback, void* context)
+{
+ int result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_134: [If `messenger_handle` is NULL, telemetry_messenger_send_async() shall fail and return a non-zero value]
+ if (messenger_handle == NULL)
+ {
+ LogError("Failed sending event (messenger_handle is NULL)");
+ result = __FAILURE__;
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_135: [If `message` is NULL, telemetry_messenger_send_async() shall fail and return a non-zero value]
+ else if (message == NULL)
+ {
+ LogError("Failed sending event (message is NULL)");
+ result = __FAILURE__;
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_136: [If `on_event_send_complete_callback` is NULL, telemetry_messenger_send_async() shall fail and return a non-zero value]
+ else if (on_messenger_event_send_complete_callback == NULL)
+ {
+ LogError("Failed sending event (on_event_send_complete_callback is NULL)");
+ result = __FAILURE__;
+ }
+ else
+ {
+ MESSENGER_SEND_EVENT_TASK *task;
+ TELEMETRY_MESSENGER_INSTANCE *instance = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_137: [telemetry_messenger_send_async() shall allocate memory for a MESSENGER_SEND_EVENT_TASK structure (aka `task`)]
+ if ((task = (MESSENGER_SEND_EVENT_TASK*)malloc(sizeof(MESSENGER_SEND_EVENT_TASK))) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_138: [If malloc() fails, telemetry_messenger_send_async() shall fail and return a non-zero value]
+ LogError("Failed sending event (failed to create struct for task; malloc failed)");
+ result = __FAILURE__;
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_100: [`task` shall be added to `instance->waiting_to_send` using singlylinkedlist_add()]
+ else if (singlylinkedlist_add(instance->waiting_to_send, task) == NULL)
+ {
+ LogError("Failed sending event (singlylinkedlist_add failed)");
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_139: [If singlylinkedlist_add() fails, telemetry_messenger_send_async() shall fail and return a non-zero value]
+ result = __FAILURE__;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_142: [If any failure occurs, telemetry_messenger_send_async() shall free any memory it has allocated]
+ free(task);
+ }
+ else
+ {
+ memset(task, 0, sizeof(MESSENGER_SEND_EVENT_TASK));
+ task->message = message;
+ task->on_event_send_complete_callback = on_messenger_event_send_complete_callback;
+ task->context = context;
+ task->send_time = INDEFINITE_TIME;
+ task->messenger = instance;
+ task->is_timed_out = false;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_143: [If no failures occur, telemetry_messenger_send_async() shall return zero]
+ result = RESULT_OK;
+ }
+ }
+
+ return result;
+}
+
+int telemetry_messenger_get_send_status(TELEMETRY_MESSENGER_HANDLE messenger_handle, TELEMETRY_MESSENGER_SEND_STATUS* send_status)
+{
+ int result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_144: [If `messenger_handle` is NULL, telemetry_messenger_get_send_status() shall fail and return a non-zero value]
+ if (messenger_handle == NULL)
+ {
+ LogError("telemetry_messenger_get_send_status failed (messenger_handle is NULL)");
+ result = __FAILURE__;
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_145: [If `send_status` is NULL, telemetry_messenger_get_send_status() shall fail and return a non-zero value]
+ else if (send_status == NULL)
+ {
+ LogError("telemetry_messenger_get_send_status failed (send_status is NULL)");
+ result = __FAILURE__;
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+ LIST_ITEM_HANDLE wts_list_head = singlylinkedlist_get_head_item(instance->waiting_to_send);
+ LIST_ITEM_HANDLE ip_list_head = singlylinkedlist_get_head_item(instance->in_progress_list);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_147: [If `instance->in_progress_list` and `instance->wait_to_send_list` are empty, send_status shall be set to TELEMETRY_MESSENGER_SEND_STATUS_IDLE]
+ if (wts_list_head == NULL && ip_list_head == NULL)
+ {
+ *send_status = TELEMETRY_MESSENGER_SEND_STATUS_IDLE;
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_148: [Otherwise, send_status shall be set to TELEMETRY_MESSENGER_SEND_STATUS_BUSY]
+ else
+ {
+ *send_status = TELEMETRY_MESSENGER_SEND_STATUS_BUSY;
+ }
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_149: [If no failures occur, telemetry_messenger_get_send_status() shall return 0]
+ result = RESULT_OK;
+ }
+
+ return result;
+}
+
+int telemetry_messenger_start(TELEMETRY_MESSENGER_HANDLE messenger_handle, SESSION_HANDLE session_handle)
+{
+ int result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_029: [If `messenger_handle` is NULL, telemetry_messenger_start() shall fail and return __FAILURE__]
+ if (messenger_handle == NULL)
+ {
+ result = __FAILURE__;
+ LogError("telemetry_messenger_start failed (messenger_handle is NULL)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_030: [If `session_handle` is NULL, telemetry_messenger_start() shall fail and return __FAILURE__]
+ else if (session_handle == NULL)
+ {
+ result = __FAILURE__;
+ LogError("telemetry_messenger_start failed (session_handle is NULL)");
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_031: [If `instance->state` is not TELEMETRY_MESSENGER_STATE_STOPPED, telemetry_messenger_start() shall fail and return __FAILURE__]
+ if (instance->state != TELEMETRY_MESSENGER_STATE_STOPPED)
+ {
+ result = __FAILURE__;
+ LogError("telemetry_messenger_start failed (current state is %d; expected TELEMETRY_MESSENGER_STATE_STOPPED)", instance->state);
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_032: [`session_handle` shall be saved on `instance->session_handle`]
+ instance->session_handle = session_handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_115: [If no failures occurr, `instance->state` shall be set to TELEMETRY_MESSENGER_STATE_STARTING, and `instance->on_state_changed_callback` invoked if provided]
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_STARTING);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_056: [If no failures occurr, telemetry_messenger_start() shall return 0]
+ result = RESULT_OK;
+ }
+ }
+
+ return result;
+}
+
+int telemetry_messenger_stop(TELEMETRY_MESSENGER_HANDLE messenger_handle)
+{
+ int result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_057: [If `messenger_handle` is NULL, telemetry_messenger_stop() shall fail and return a non-zero value]
+ if (messenger_handle == NULL)
+ {
+ result = __FAILURE__;
+ LogError("telemetry_messenger_stop failed (messenger_handle is NULL)");
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_058: [If `instance->state` is TELEMETRY_MESSENGER_STATE_STOPPED, telemetry_messenger_stop() shall fail and return a non-zero value]
+ if (instance->state == TELEMETRY_MESSENGER_STATE_STOPPED)
+ {
+ result = __FAILURE__;
+ LogError("telemetry_messenger_stop failed (messenger is already stopped)");
+ }
+ else
+ {
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_STOPPING);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_152: [telemetry_messenger_stop() shall close and destroy `instance->message_sender` and `instance->message_receiver`]
+ destroy_event_sender(instance);
+ destroy_message_receiver(instance);
+
+ remove_timed_out_events(instance);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_162: [telemetry_messenger_stop() shall move all items from `instance->in_progress_list` to the beginning of `instance->wait_to_send_list`]
+ if (move_events_to_wait_to_send_list(instance) != RESULT_OK)
+ {
+ LogError("Messenger failed to move events in progress back to wait_to_send list");
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_163: [If not all items from `instance->in_progress_list` can be moved back to `instance->wait_to_send_list`, `instance->state` shall be set to TELEMETRY_MESSENGER_STATE_ERROR, and `instance->on_state_changed_callback` invoked]
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ result = __FAILURE__;
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_164: [If all items get successfuly moved back to `instance->wait_to_send_list`, `instance->state` shall be set to TELEMETRY_MESSENGER_STATE_STOPPED, and `instance->on_state_changed_callback` invoked]
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_STOPPED);
+ result = RESULT_OK;
+ }
+ }
+ }
+
+ return result;
+}
+
+// @brief
+// Sets the messenger module state based on the state changes from messagesender and messagereceiver
+static void process_state_changes(TELEMETRY_MESSENGER_INSTANCE* instance)
+{
+ // Note: messagesender and messagereceiver are still not created or already destroyed
+ // when state is TELEMETRY_MESSENGER_STATE_STOPPED, so no checking is needed there.
+
+ if (instance->state == TELEMETRY_MESSENGER_STATE_STARTED)
+ {
+ if (instance->message_sender_current_state != MESSAGE_SENDER_STATE_OPEN)
+ {
+ LogError("messagesender reported unexpected state %d while messenger was started", instance->message_sender_current_state);
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ }
+ else if (instance->message_receiver != NULL && instance->message_receiver_current_state != MESSAGE_RECEIVER_STATE_OPEN)
+ {
+ if (instance->message_receiver_current_state == MESSAGE_RECEIVER_STATE_OPENING)
+ {
+ int is_timed_out;
+ if (is_timeout_reached(instance->last_message_receiver_state_change_time, MAX_MESSAGE_RECEIVER_STATE_CHANGE_TIMEOUT_SECS, &is_timed_out) != RESULT_OK)
+ {
+ LogError("messenger got an error (failed to verify messagereceiver start timeout)");
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ }
+ else if (is_timed_out == 1)
+ {
+ LogError("messenger got an error (messagereceiver failed to start within expected timeout (%d secs))", MAX_MESSAGE_RECEIVER_STATE_CHANGE_TIMEOUT_SECS);
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ }
+ }
+ else if (instance->message_receiver_current_state == MESSAGE_RECEIVER_STATE_ERROR ||
+ instance->message_receiver_current_state == MESSAGE_RECEIVER_STATE_IDLE)
+ {
+ LogError("messagereceiver reported unexpected state %d while messenger is starting", instance->message_receiver_current_state);
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ }
+ }
+ }
+ else
+ {
+ if (instance->state == TELEMETRY_MESSENGER_STATE_STARTING)
+ {
+ if (instance->message_sender_current_state == MESSAGE_SENDER_STATE_OPEN)
+ {
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_STARTED);
+ }
+ else if (instance->message_sender_current_state == MESSAGE_SENDER_STATE_OPENING)
+ {
+ int is_timed_out;
+ if (is_timeout_reached(instance->last_message_sender_state_change_time, MAX_MESSAGE_SENDER_STATE_CHANGE_TIMEOUT_SECS, &is_timed_out) != RESULT_OK)
+ {
+ LogError("messenger failed to start (failed to verify messagesender start timeout)");
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ }
+ else if (is_timed_out == 1)
+ {
+ LogError("messenger failed to start (messagesender failed to start within expected timeout (%d secs))", MAX_MESSAGE_SENDER_STATE_CHANGE_TIMEOUT_SECS);
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ }
+ }
+ // For this module, the only valid scenario where messagesender state is IDLE is if
+ // the messagesender hasn't been created yet or already destroyed.
+ else if ((instance->message_sender_current_state == MESSAGE_SENDER_STATE_ERROR) ||
+ (instance->message_sender_current_state == MESSAGE_SENDER_STATE_CLOSING) ||
+ (instance->message_sender_current_state == MESSAGE_SENDER_STATE_IDLE && instance->message_sender != NULL))
+ {
+ LogError("messagesender reported unexpected state %d while messenger is starting", instance->message_sender_current_state);
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ }
+ }
+ // message sender and receiver are stopped/destroyed synchronously, so no need for state control.
+ }
+}
+
+void telemetry_messenger_do_work(TELEMETRY_MESSENGER_HANDLE messenger_handle)
+{
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_065: [If `messenger_handle` is NULL, telemetry_messenger_do_work() shall fail and return]
+ if (messenger_handle == NULL)
+ {
+ LogError("telemetry_messenger_do_work failed (messenger_handle is NULL)");
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+
+ process_state_changes(instance);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_151: [If `instance->state` is TELEMETRY_MESSENGER_STATE_STARTING, telemetry_messenger_do_work() shall create and open `instance->message_sender`]
+ if (instance->state == TELEMETRY_MESSENGER_STATE_STARTING)
+ {
+ if (instance->message_sender == NULL)
+ {
+ if (create_event_sender(instance) != RESULT_OK)
+ {
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ }
+ }
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_066: [If `instance->state` is not TELEMETRY_MESSENGER_STATE_STARTED, telemetry_messenger_do_work() shall return]
+ else if (instance->state == TELEMETRY_MESSENGER_STATE_STARTED)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_067: [If `instance->receive_messages` is true and `instance->message_receiver` is NULL, a message_receiver shall be created]
+ if (instance->receive_messages == true &&
+ instance->message_receiver == NULL &&
+ create_message_receiver(instance) != RESULT_OK)
+ {
+ LogError("telemetry_messenger_do_work warning (failed creating the message receiver [%s])", STRING_c_str(instance->device_id));
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_092: [If `instance->receive_messages` is false and `instance->message_receiver` is not NULL, it shall be destroyed]
+ else if (instance->receive_messages == false && instance->message_receiver != NULL)
+ {
+ destroy_message_receiver(instance);
+ }
+
+ if (process_event_send_timeouts(instance) != RESULT_OK)
+ {
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ }
+ else if (send_pending_events(instance) != RESULT_OK && instance->event_send_retry_limit > 0)
+ {
+ instance->event_send_error_count++;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_161: [If telemetry_messenger_do_work() fail sending events for `instance->event_send_retry_limit` times in a row, it shall invoke `instance->on_state_changed_callback`, if provided, with error code TELEMETRY_MESSENGER_STATE_ERROR]
+ if (instance->event_send_error_count >= instance->event_send_retry_limit)
+ {
+ LogError("telemetry_messenger_do_work failed (failed sending events; reached max number of consecutive attempts)");
+ update_messenger_state(instance, TELEMETRY_MESSENGER_STATE_ERROR);
+ }
+ }
+ else
+ {
+ instance->event_send_error_count = 0;
+ }
+ }
+ }
+}
+
+void telemetry_messenger_destroy(TELEMETRY_MESSENGER_HANDLE messenger_handle)
+{
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_109: [If `messenger_handle` is NULL, telemetry_messenger_destroy() shall fail and return]
+ if (messenger_handle == NULL)
+ {
+ LogError("telemetry_messenger_destroy failed (messenger_handle is NULL)");
+ }
+ else
+ {
+ LIST_ITEM_HANDLE list_node;
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_110: [If the `instance->state` is not TELEMETRY_MESSENGER_STATE_STOPPED, telemetry_messenger_destroy() shall invoke telemetry_messenger_stop()]
+ if (instance->state != TELEMETRY_MESSENGER_STATE_STOPPED)
+ {
+ (void)telemetry_messenger_stop(messenger_handle);
+ }
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_111: [All elements of `instance->in_progress_list` and `instance->wait_to_send_list` shall be removed, invoking `task->on_event_send_complete_callback` for each with EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED]
+
+ // Note: yes telemetry_messenger_stop() tried to move all events from in_progress_list to wait_to_send_list,
+ // but we need to iterate through in case any events failed to be moved.
+ while ((list_node = singlylinkedlist_get_head_item(instance->in_progress_list)) != NULL)
+ {
+ MESSENGER_SEND_EVENT_TASK* task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_node);
+
+ (void)singlylinkedlist_remove(instance->in_progress_list, list_node);
+
+ if (task != NULL)
+ {
+ task->on_event_send_complete_callback(task->message, TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED, (void*)task->context);
+ free(task);
+ }
+ }
+
+ while ((list_node = singlylinkedlist_get_head_item(instance->waiting_to_send)) != NULL)
+ {
+ MESSENGER_SEND_EVENT_TASK* task = (MESSENGER_SEND_EVENT_TASK*)singlylinkedlist_item_get_value(list_node);
+
+ (void)singlylinkedlist_remove(instance->waiting_to_send, list_node);
+
+ if (task != NULL)
+ {
+ task->on_event_send_complete_callback(task->message, TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED, (void*)task->context);
+ free(task);
+ }
+ }
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_150: [`instance->in_progress_list` and `instance->wait_to_send_list` shall be destroyed using singlylinkedlist_destroy()]
+ singlylinkedlist_destroy(instance->waiting_to_send);
+ singlylinkedlist_destroy(instance->in_progress_list);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_112: [`instance->iothub_host_fqdn` shall be destroyed using STRING_delete()]
+ STRING_delete(instance->iothub_host_fqdn);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_113: [`instance->device_id` shall be destroyed using STRING_delete()]
+ STRING_delete(instance->device_id);
+
+ STRING_delete(instance->product_info);
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_114: [telemetry_messenger_destroy() shall destroy `instance` with free()]
+ (void)free(instance);
+ }
+}
+
+TELEMETRY_MESSENGER_HANDLE telemetry_messenger_create(const TELEMETRY_MESSENGER_CONFIG* messenger_config, const char* product_info)
+{
+ TELEMETRY_MESSENGER_HANDLE handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_001: [If parameter `messenger_config` is NULL, telemetry_messenger_create() shall return NULL]
+ if (messenger_config == NULL)
+ {
+ handle = NULL;
+ LogError("telemetry_messenger_create failed (messenger_config is NULL)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_002: [If `messenger_config->device_id` is NULL, telemetry_messenger_create() shall return NULL]
+ else if (messenger_config->device_id == NULL)
+ {
+ handle = NULL;
+ LogError("telemetry_messenger_create failed (messenger_config->device_id is NULL)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_003: [If `messenger_config->iothub_host_fqdn` is NULL, telemetry_messenger_create() shall return NULL]
+ else if (messenger_config->iothub_host_fqdn == NULL)
+ {
+ handle = NULL;
+ LogError("telemetry_messenger_create failed (messenger_config->iothub_host_fqdn is NULL)");
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_006: [telemetry_messenger_create() shall allocate memory for the messenger instance structure (aka `instance`)]
+ if ((instance = (TELEMETRY_MESSENGER_INSTANCE*)malloc(sizeof(TELEMETRY_MESSENGER_INSTANCE))) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_007: [If malloc() fails, telemetry_messenger_create() shall fail and return NULL]
+ handle = NULL;
+ LogError("telemetry_messenger_create failed (messenger_config->wait_to_send_list is NULL)");
+ }
+ else
+ {
+ memset(instance, 0, sizeof(TELEMETRY_MESSENGER_INSTANCE));
+ instance->state = TELEMETRY_MESSENGER_STATE_STOPPED;
+ instance->message_sender_current_state = MESSAGE_SENDER_STATE_IDLE;
+ instance->message_sender_previous_state = MESSAGE_SENDER_STATE_IDLE;
+ instance->message_receiver_current_state = MESSAGE_RECEIVER_STATE_IDLE;
+ instance->message_receiver_previous_state = MESSAGE_RECEIVER_STATE_IDLE;
+ instance->event_send_retry_limit = DEFAULT_EVENT_SEND_RETRY_LIMIT;
+ instance->event_send_timeout_secs = DEFAULT_EVENT_SEND_TIMEOUT_SECS;
+ instance->last_message_sender_state_change_time = INDEFINITE_TIME;
+ instance->last_message_receiver_state_change_time = INDEFINITE_TIME;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_008: [telemetry_messenger_create() shall save a copy of `messenger_config->device_id` into `instance->device_id`]
+ if ((instance->device_id = STRING_construct(messenger_config->device_id)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_009: [If STRING_construct() fails, telemetry_messenger_create() shall fail and return NULL]
+ handle = NULL;
+ LogError("telemetry_messenger_create failed (device_id could not be copied; STRING_construct failed)");
+ }
+ else if ((instance->product_info = STRING_construct(product_info)) == NULL)
+ {
+ handle = NULL;
+ LogError("telemetry_messenger_create failed (product_info could not be copied; STRING_construct failed)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_010: [telemetry_messenger_create() shall save a copy of `messenger_config->iothub_host_fqdn` into `instance->iothub_host_fqdn`]
+ else if ((instance->iothub_host_fqdn = STRING_construct(messenger_config->iothub_host_fqdn)) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_011: [If STRING_construct() fails, telemetry_messenger_create() shall fail and return NULL]
+ handle = NULL;
+ LogError("telemetry_messenger_create failed (iothub_host_fqdn could not be copied; STRING_construct failed)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_165: [`instance->wait_to_send_list` shall be set using singlylinkedlist_create()]
+ else if ((instance->waiting_to_send = singlylinkedlist_create()) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_166: [If singlylinkedlist_create() fails, telemetry_messenger_create() shall fail and return NULL]
+ handle = NULL;
+ LogError("telemetry_messenger_create failed (singlylinkedlist_create failed to create wait_to_send_list)");
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_132: [`instance->in_progress_list` shall be set using singlylinkedlist_create()]
+ else if ((instance->in_progress_list = singlylinkedlist_create()) == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_133: [If singlylinkedlist_create() fails, telemetry_messenger_create() shall fail and return NULL]
+ handle = NULL;
+ LogError("telemetry_messenger_create failed (singlylinkedlist_create failed to create in_progress_list)");
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_013: [`messenger_config->on_state_changed_callback` shall be saved into `instance->on_state_changed_callback`]
+ instance->on_state_changed_callback = messenger_config->on_state_changed_callback;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_014: [`messenger_config->on_state_changed_context` shall be saved into `instance->on_state_changed_context`]
+ instance->on_state_changed_context = messenger_config->on_state_changed_context;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_015: [If no failures occurr, telemetry_messenger_create() shall return a handle to `instance`]
+ handle = (TELEMETRY_MESSENGER_HANDLE)instance;
+ }
+ }
+
+ if (handle == NULL)
+ {
+ telemetry_messenger_destroy((TELEMETRY_MESSENGER_HANDLE)instance);
+ }
+ }
+
+ return handle;
+}
+
+int telemetry_messenger_set_option(TELEMETRY_MESSENGER_HANDLE messenger_handle, const char* name, void* value)
+{
+ int result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_167: [If `messenger_handle` or `name` or `value` is NULL, telemetry_messenger_set_option shall fail and return a non-zero value]
+ if (messenger_handle == NULL || name == NULL || value == NULL)
+ {
+ LogError("telemetry_messenger_set_option failed (one of the followin are NULL: messenger_handle=%p, name=%p, value=%p)",
+ messenger_handle, name, value);
+ result = __FAILURE__;
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_168: [If name matches MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS, `value` shall be saved on `instance->event_send_timeout_secs`]
+ if (strcmp(MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS, name) == 0)
+ {
+ instance->event_send_timeout_secs = *((size_t*)value);
+ result = RESULT_OK;
+ }
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_169: [If name matches MESSENGER_OPTION_SAVED_OPTIONS, `value` shall be applied using OptionHandler_FeedOptions]
+ else if (strcmp(MESSENGER_OPTION_SAVED_OPTIONS, name) == 0)
+ {
+ if (OptionHandler_FeedOptions((OPTIONHANDLER_HANDLE)value, messenger_handle) != OPTIONHANDLER_OK)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_170: [If OptionHandler_FeedOptions fails, telemetry_messenger_set_option shall fail and return a non-zero value]
+ LogError("telemetry_messenger_set_option failed (OptionHandler_FeedOptions failed)");
+ result = __FAILURE__;
+ }
+ else
+ {
+ result = RESULT_OK;
+ }
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_171: [If name does not match any supported option, authentication_set_option shall fail and return a non-zero value]
+ LogError("telemetry_messenger_set_option failed (option with name '%s' is not suppported)", name);
+ result = __FAILURE__;
+ }
+ }
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_172: [If no errors occur, telemetry_messenger_set_option shall return 0]
+ return result;
+}
+
+OPTIONHANDLER_HANDLE telemetry_messenger_retrieve_options(TELEMETRY_MESSENGER_HANDLE messenger_handle)
+{
+ OPTIONHANDLER_HANDLE result;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_173: [If `messenger_handle` is NULL, telemetry_messenger_retrieve_options shall fail and return NULL]
+ if (messenger_handle == NULL)
+ {
+ LogError("Failed to retrieve options from messenger instance (messenger_handle is NULL)");
+ result = NULL;
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_174: [An OPTIONHANDLER_HANDLE instance shall be created using OptionHandler_Create]
+ OPTIONHANDLER_HANDLE options = OptionHandler_Create(telemetry_messenger_clone_option, telemetry_messenger_destroy_option, (pfSetOption)telemetry_messenger_set_option);
+
+ if (options == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_175: [If an OPTIONHANDLER_HANDLE instance fails to be created, telemetry_messenger_retrieve_options shall fail and return NULL]
+ LogError("Failed to retrieve options from messenger instance (OptionHandler_Create failed)");
+ result = NULL;
+ }
+ else
+ {
+ TELEMETRY_MESSENGER_INSTANCE* instance = (TELEMETRY_MESSENGER_INSTANCE*)messenger_handle;
+
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_176: [Each option of `instance` shall be added to the OPTIONHANDLER_HANDLE instance using OptionHandler_AddOption]
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_177: [If OptionHandler_AddOption fails, telemetry_messenger_retrieve_options shall fail and return NULL]
+ if (OptionHandler_AddOption(options, MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS, (void*)&instance->event_send_timeout_secs) != OPTIONHANDLER_OK)
+ {
+ LogError("Failed to retrieve options from messenger instance (OptionHandler_Create failed for option '%s')", MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS);
+ result = NULL;
+ }
+ else
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_179: [If no failures occur, telemetry_messenger_retrieve_options shall return the OPTIONHANDLER_HANDLE instance]
+ result = options;
+ }
+
+ if (result == NULL)
+ {
+ // Codes_SRS_IOTHUBTRANSPORT_AMQP_MESSENGER_09_178: [If telemetry_messenger_retrieve_options fails, any allocated memory shall be freed]
+ OptionHandler_Destroy(options);
+ }
+ }
+ }
+
+ return result;
+}
--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/iothubtransport_amqp_telemetry_messenger.h Fri Jun 16 16:12:09 2017 -0700
@@ -0,0 +1,91 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+#ifndef IOTHUBTRANSPORT_AMQP_TELEMETRY_MESSENGER
+#define IOTHUBTRANSPORT_AMQP_TELEMETRY_MESSENGER
+
+#include "azure_c_shared_utility/umock_c_prod.h"
+#include "azure_c_shared_utility/optionhandler.h"
+#include "azure_uamqp_c/session.h"
+#include "iothub_client_private.h"
+
+#ifdef __cplusplus
+extern "C"
+{
+#endif
+
+
+static const char* MESSENGER_OPTION_EVENT_SEND_TIMEOUT_SECS = "telemetry_event_send_timeout_secs";
+static const char* MESSENGER_OPTION_SAVED_OPTIONS = "saved_telemetry_messenger_options";
+
+typedef struct TELEMETRY_MESSENGER_INSTANCE* TELEMETRY_MESSENGER_HANDLE;
+
+typedef enum TELEMETRY_MESSENGER_SEND_STATUS_TAG
+{
+ TELEMETRY_MESSENGER_SEND_STATUS_IDLE,
+ TELEMETRY_MESSENGER_SEND_STATUS_BUSY
+} TELEMETRY_MESSENGER_SEND_STATUS;
+
+typedef enum TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_TAG
+{
+ TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_OK,
+ TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_CANNOT_PARSE,
+ TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_FAIL_SENDING,
+ TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_ERROR_TIMEOUT,
+ TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT_MESSENGER_DESTROYED
+} TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT;
+
+typedef enum TELEMETRY_MESSENGER_DISPOSITION_RESULT_TAG
+{
+ TELEMETRY_MESSENGER_DISPOSITION_RESULT_NONE,
+ TELEMETRY_MESSENGER_DISPOSITION_RESULT_ACCEPTED,
+ TELEMETRY_MESSENGER_DISPOSITION_RESULT_REJECTED,
+ TELEMETRY_MESSENGER_DISPOSITION_RESULT_RELEASED
+} TELEMETRY_MESSENGER_DISPOSITION_RESULT;
+
+typedef enum TELEMETRY_MESSENGER_STATE_TAG
+{
+ TELEMETRY_MESSENGER_STATE_STARTING,
+ TELEMETRY_MESSENGER_STATE_STARTED,
+ TELEMETRY_MESSENGER_STATE_STOPPING,
+ TELEMETRY_MESSENGER_STATE_STOPPED,
+ TELEMETRY_MESSENGER_STATE_ERROR
+} TELEMETRY_MESSENGER_STATE;
+
+typedef struct TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO_TAG
+{
+ delivery_number message_id;
+ char* source;
+} TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO;
+
+typedef void(*ON_TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE)(IOTHUB_MESSAGE_LIST* iothub_message_list, TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE_RESULT messenger_event_send_complete_result, void* context);
+typedef void(*ON_TELEMETRY_MESSENGER_STATE_CHANGED_CALLBACK)(void* context, TELEMETRY_MESSENGER_STATE previous_state, TELEMETRY_MESSENGER_STATE new_state);
+typedef TELEMETRY_MESSENGER_DISPOSITION_RESULT(*ON_TELEMETRY_MESSENGER_MESSAGE_RECEIVED)(IOTHUB_MESSAGE_HANDLE message, TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO* disposition_info, void* context);
+
+typedef struct TELEMETRY_MESSENGER_CONFIG_TAG
+{
+ const char* device_id;
+ char* iothub_host_fqdn;
+ ON_TELEMETRY_MESSENGER_STATE_CHANGED_CALLBACK on_state_changed_callback;
+ void* on_state_changed_context;
+} TELEMETRY_MESSENGER_CONFIG;
+
+MOCKABLE_FUNCTION(, TELEMETRY_MESSENGER_HANDLE, telemetry_messenger_create, const TELEMETRY_MESSENGER_CONFIG*, messenger_config, const char*, product_info);
+MOCKABLE_FUNCTION(, int, telemetry_messenger_send_async, TELEMETRY_MESSENGER_HANDLE, messenger_handle, IOTHUB_MESSAGE_LIST*, message, ON_TELEMETRY_MESSENGER_EVENT_SEND_COMPLETE, on_messenger_event_send_complete_callback, void*, context);
+MOCKABLE_FUNCTION(, int, telemetry_messenger_subscribe_for_messages, TELEMETRY_MESSENGER_HANDLE, messenger_handle, ON_TELEMETRY_MESSENGER_MESSAGE_RECEIVED, on_message_received_callback, void*, context);
+MOCKABLE_FUNCTION(, int, telemetry_messenger_unsubscribe_for_messages, TELEMETRY_MESSENGER_HANDLE, messenger_handle);
+MOCKABLE_FUNCTION(, int, telemetry_messenger_send_message_disposition, TELEMETRY_MESSENGER_HANDLE, messenger_handle, TELEMETRY_MESSENGER_MESSAGE_DISPOSITION_INFO*, disposition_info, TELEMETRY_MESSENGER_DISPOSITION_RESULT, disposition_result);
+MOCKABLE_FUNCTION(, int, telemetry_messenger_get_send_status, TELEMETRY_MESSENGER_HANDLE, messenger_handle, TELEMETRY_MESSENGER_SEND_STATUS*, send_status);
+MOCKABLE_FUNCTION(, int, telemetry_messenger_start, TELEMETRY_MESSENGER_HANDLE, messenger_handle, SESSION_HANDLE, session_handle);
+MOCKABLE_FUNCTION(, int, telemetry_messenger_stop, TELEMETRY_MESSENGER_HANDLE, messenger_handle);
+MOCKABLE_FUNCTION(, void, telemetry_messenger_do_work, TELEMETRY_MESSENGER_HANDLE, messenger_handle);
+MOCKABLE_FUNCTION(, void, telemetry_messenger_destroy, TELEMETRY_MESSENGER_HANDLE, messenger_handle);
+MOCKABLE_FUNCTION(, int, telemetry_messenger_set_option, TELEMETRY_MESSENGER_HANDLE, messenger_handle, const char*, name, void*, value);
+MOCKABLE_FUNCTION(, OPTIONHANDLER_HANDLE, telemetry_messenger_retrieve_options, TELEMETRY_MESSENGER_HANDLE, messenger_handle);
+
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*IOTHUBTRANSPORT_AMQP_TELEMETRY_MESSENGER*/
