Azure IoT / azure_uamqp_c

A small memory footprint AMQP implimentation

Dependents:   iothub_client_sample_amqp remote_monitoring simplesample_amqp

message_sender.c@16:22a72cf8e416, 2017-01-13 (annotated)

Committer:
AzureIoTClient
Date:
Fri Jan 13 18:40:42 2017 -0800
Revision:
16:22a72cf8e416
Parent:
13:9abd748f4e78
Child:
17:923575db8b2d
1.1.4

Who changed what in which revision?

UserRevisionLine numberNew contents of line
Azure.IoT Build 0:6ae2f7bca550 1 // Copyright (c) Microsoft. All rights reserved.
Azure.IoT Build 0:6ae2f7bca550 2 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
Azure.IoT Build 0:6ae2f7bca550 3
Azure.IoT Build 0:6ae2f7bca550 4 #include <stdlib.h>
Azure.IoT Build 0:6ae2f7bca550 5 #ifdef _CRTDBG_MAP_ALLOC
Azure.IoT Build 0:6ae2f7bca550 6 #include <crtdbg.h>
Azure.IoT Build 0:6ae2f7bca550 7 #endif
AzureIoTClient 6:641a9672db08 8 #include <stdbool.h>
Azure.IoT Build 0:6ae2f7bca550 9 #include <string.h>
Azure.IoT Build 5:ae49385aff34 10 #include "azure_c_shared_utility/xlogging.h"
Azure.IoT Build 0:6ae2f7bca550 11 #include "azure_uamqp_c/message_sender.h"
Azure.IoT Build 0:6ae2f7bca550 12 #include "azure_uamqp_c/amqpalloc.h"
Azure.IoT Build 0:6ae2f7bca550 13 #include "azure_uamqp_c/amqpvalue_to_string.h"
Azure.IoT Build 0:6ae2f7bca550 14
Azure.IoT Build 0:6ae2f7bca550 15 typedef enum MESSAGE_SEND_STATE_TAG
Azure.IoT Build 0:6ae2f7bca550 16 {
AzureIoTClient 6:641a9672db08 17 MESSAGE_SEND_STATE_NOT_SENT,
AzureIoTClient 6:641a9672db08 18 MESSAGE_SEND_STATE_PENDING
Azure.IoT Build 0:6ae2f7bca550 19 } MESSAGE_SEND_STATE;
Azure.IoT Build 0:6ae2f7bca550 20
Azure.IoT Build 0:6ae2f7bca550 21 typedef enum SEND_ONE_MESSAGE_RESULT_TAG
Azure.IoT Build 0:6ae2f7bca550 22 {
AzureIoTClient 6:641a9672db08 23 SEND_ONE_MESSAGE_OK,
AzureIoTClient 6:641a9672db08 24 SEND_ONE_MESSAGE_ERROR,
AzureIoTClient 6:641a9672db08 25 SEND_ONE_MESSAGE_BUSY
Azure.IoT Build 0:6ae2f7bca550 26 } SEND_ONE_MESSAGE_RESULT;
Azure.IoT Build 0:6ae2f7bca550 27
Azure.IoT Build 0:6ae2f7bca550 28 typedef struct MESSAGE_WITH_CALLBACK_TAG
Azure.IoT Build 0:6ae2f7bca550 29 {
AzureIoTClient 6:641a9672db08 30 MESSAGE_HANDLE message;
AzureIoTClient 6:641a9672db08 31 ON_MESSAGE_SEND_COMPLETE on_message_send_complete;
AzureIoTClient 6:641a9672db08 32 void* context;
AzureIoTClient 6:641a9672db08 33 MESSAGE_SENDER_HANDLE message_sender;
AzureIoTClient 6:641a9672db08 34 MESSAGE_SEND_STATE message_send_state;
Azure.IoT Build 0:6ae2f7bca550 35 } MESSAGE_WITH_CALLBACK;
Azure.IoT Build 0:6ae2f7bca550 36
Azure.IoT Build 0:6ae2f7bca550 37 typedef struct MESSAGE_SENDER_INSTANCE_TAG
Azure.IoT Build 0:6ae2f7bca550 38 {
AzureIoTClient 6:641a9672db08 39 LINK_HANDLE link;
AzureIoTClient 6:641a9672db08 40 size_t message_count;
AzureIoTClient 6:641a9672db08 41 MESSAGE_WITH_CALLBACK** messages;
AzureIoTClient 6:641a9672db08 42 MESSAGE_SENDER_STATE message_sender_state;
AzureIoTClient 6:641a9672db08 43 ON_MESSAGE_SENDER_STATE_CHANGED on_message_sender_state_changed;
AzureIoTClient 6:641a9672db08 44 void* on_message_sender_state_changed_context;
AzureIoTClient 6:641a9672db08 45 unsigned int is_trace_on : 1;
Azure.IoT Build 0:6ae2f7bca550 46 } MESSAGE_SENDER_INSTANCE;
Azure.IoT Build 0:6ae2f7bca550 47
Azure.IoT Build 0:6ae2f7bca550 48 static void remove_pending_message_by_index(MESSAGE_SENDER_INSTANCE* message_sender_instance, size_t index)
Azure.IoT Build 0:6ae2f7bca550 49 {
AzureIoTClient 6:641a9672db08 50 MESSAGE_WITH_CALLBACK** new_messages;
Azure.IoT Build 0:6ae2f7bca550 51
AzureIoTClient 6:641a9672db08 52 if (message_sender_instance->messages[index]->message != NULL)
AzureIoTClient 6:641a9672db08 53 {
AzureIoTClient 6:641a9672db08 54 message_destroy(message_sender_instance->messages[index]->message);
AzureIoTClient 6:641a9672db08 55 message_sender_instance->messages[index]->message = NULL;
AzureIoTClient 6:641a9672db08 56 }
Azure.IoT Build 0:6ae2f7bca550 57
AzureIoTClient 6:641a9672db08 58 amqpalloc_free(message_sender_instance->messages[index]);
Azure.IoT Build 0:6ae2f7bca550 59
AzureIoTClient 6:641a9672db08 60 if (message_sender_instance->message_count - index > 1)
AzureIoTClient 6:641a9672db08 61 {
AzureIoTClient 6:641a9672db08 62 (void)memmove(&message_sender_instance->messages[index], &message_sender_instance->messages[index + 1], sizeof(MESSAGE_WITH_CALLBACK*) * (message_sender_instance->message_count - index - 1));
AzureIoTClient 6:641a9672db08 63 }
Azure.IoT Build 0:6ae2f7bca550 64
AzureIoTClient 6:641a9672db08 65 message_sender_instance->message_count--;
Azure.IoT Build 0:6ae2f7bca550 66
AzureIoTClient 6:641a9672db08 67 if (message_sender_instance->message_count > 0)
AzureIoTClient 6:641a9672db08 68 {
AzureIoTClient 6:641a9672db08 69 new_messages = (MESSAGE_WITH_CALLBACK**)amqpalloc_realloc(message_sender_instance->messages, sizeof(MESSAGE_WITH_CALLBACK*) * (message_sender_instance->message_count));
AzureIoTClient 6:641a9672db08 70 if (new_messages != NULL)
AzureIoTClient 6:641a9672db08 71 {
AzureIoTClient 6:641a9672db08 72 message_sender_instance->messages = new_messages;
AzureIoTClient 6:641a9672db08 73 }
AzureIoTClient 6:641a9672db08 74 }
AzureIoTClient 6:641a9672db08 75 else
AzureIoTClient 6:641a9672db08 76 {
AzureIoTClient 6:641a9672db08 77 amqpalloc_free(message_sender_instance->messages);
AzureIoTClient 6:641a9672db08 78 message_sender_instance->messages = NULL;
AzureIoTClient 6:641a9672db08 79 }
Azure.IoT Build 0:6ae2f7bca550 80 }
Azure.IoT Build 0:6ae2f7bca550 81
Azure.IoT Build 0:6ae2f7bca550 82 static void remove_pending_message(MESSAGE_SENDER_INSTANCE* message_sender_instance, MESSAGE_WITH_CALLBACK* message_with_callback)
Azure.IoT Build 0:6ae2f7bca550 83 {
AzureIoTClient 6:641a9672db08 84 size_t i;
Azure.IoT Build 0:6ae2f7bca550 85
AzureIoTClient 6:641a9672db08 86 for (i = 0; i < message_sender_instance->message_count; i++)
AzureIoTClient 6:641a9672db08 87 {
AzureIoTClient 6:641a9672db08 88 if (message_sender_instance->messages[i] == message_with_callback)
AzureIoTClient 6:641a9672db08 89 {
AzureIoTClient 6:641a9672db08 90 remove_pending_message_by_index(message_sender_instance, i);
AzureIoTClient 6:641a9672db08 91 break;
AzureIoTClient 6:641a9672db08 92 }
AzureIoTClient 6:641a9672db08 93 }
Azure.IoT Build 0:6ae2f7bca550 94 }
Azure.IoT Build 0:6ae2f7bca550 95
AzureIoTClient 7:9e9ab3b0efef 96 static void on_delivery_settled(void* context, delivery_number delivery_no, AMQP_VALUE delivery_state)
Azure.IoT Build 0:6ae2f7bca550 97 {
AzureIoTClient 6:641a9672db08 98 MESSAGE_WITH_CALLBACK* message_with_callback = (MESSAGE_WITH_CALLBACK*)context;
AzureIoTClient 6:641a9672db08 99 MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_with_callback->message_sender;
AzureIoTClient 6:641a9672db08 100 (void)delivery_no;
Azure.IoT Build 0:6ae2f7bca550 101
AzureIoTClient 6:641a9672db08 102 if (message_with_callback->on_message_send_complete != NULL)
AzureIoTClient 6:641a9672db08 103 {
AzureIoTClient 7:9e9ab3b0efef 104 AMQP_VALUE descriptor = amqpvalue_get_inplace_descriptor(delivery_state);
AzureIoTClient 16:22a72cf8e416 105 if ((descriptor == NULL) && (delivery_state != NULL))
AzureIoTClient 7:9e9ab3b0efef 106 {
AzureIoTClient 7:9e9ab3b0efef 107 LogError("Error getting descriptor for delivery state");
AzureIoTClient 7:9e9ab3b0efef 108 }
AzureIoTClient 7:9e9ab3b0efef 109 else
AzureIoTClient 7:9e9ab3b0efef 110 {
AzureIoTClient 7:9e9ab3b0efef 111 MESSAGE_SEND_RESULT message_send_result;
AzureIoTClient 7:9e9ab3b0efef 112
AzureIoTClient 7:9e9ab3b0efef 113 if ((delivery_state == NULL) ||
AzureIoTClient 7:9e9ab3b0efef 114 (is_accepted_type_by_descriptor(descriptor)))
AzureIoTClient 7:9e9ab3b0efef 115 {
AzureIoTClient 7:9e9ab3b0efef 116 message_send_result = MESSAGE_SEND_OK;
AzureIoTClient 7:9e9ab3b0efef 117 }
AzureIoTClient 7:9e9ab3b0efef 118 else
AzureIoTClient 7:9e9ab3b0efef 119 {
AzureIoTClient 7:9e9ab3b0efef 120 message_send_result = MESSAGE_SEND_ERROR;
AzureIoTClient 7:9e9ab3b0efef 121 }
AzureIoTClient 7:9e9ab3b0efef 122
AzureIoTClient 7:9e9ab3b0efef 123 message_with_callback->on_message_send_complete(message_with_callback->context, message_send_result);
AzureIoTClient 7:9e9ab3b0efef 124 }
AzureIoTClient 6:641a9672db08 125 }
Azure.IoT Build 0:6ae2f7bca550 126
AzureIoTClient 6:641a9672db08 127 remove_pending_message(message_sender_instance, message_with_callback);
Azure.IoT Build 0:6ae2f7bca550 128 }
Azure.IoT Build 0:6ae2f7bca550 129
Azure.IoT Build 0:6ae2f7bca550 130 static int encode_bytes(void* context, const unsigned char* bytes, size_t length)
Azure.IoT Build 0:6ae2f7bca550 131 {
AzureIoTClient 6:641a9672db08 132 PAYLOAD* payload = (PAYLOAD*)context;
AzureIoTClient 6:641a9672db08 133 (void)memcpy((unsigned char*)payload->bytes + payload->length, bytes, length);
AzureIoTClient 6:641a9672db08 134 payload->length += length;
AzureIoTClient 6:641a9672db08 135 return 0;
Azure.IoT Build 0:6ae2f7bca550 136 }
Azure.IoT Build 0:6ae2f7bca550 137
Azure.IoT Build 0:6ae2f7bca550 138 static void log_message_chunk(MESSAGE_SENDER_INSTANCE* message_sender_instance, const char* name, AMQP_VALUE value)
Azure.IoT Build 0:6ae2f7bca550 139 {
AzureIoTClient 9:c22db038556c 140 #ifdef NO_LOGGING
AzureIoTClient 9:c22db038556c 141 UNUSED(message_sender_instance);
AzureIoTClient 9:c22db038556c 142 UNUSED(name);
AzureIoTClient 9:c22db038556c 143 UNUSED(value);
AzureIoTClient 9:c22db038556c 144 #else
AzureIoTClient 6:641a9672db08 145 if (xlogging_get_log_function() != NULL && message_sender_instance->is_trace_on == 1)
AzureIoTClient 6:641a9672db08 146 {
AzureIoTClient 6:641a9672db08 147 char* value_as_string = NULL;
AzureIoTClient 16:22a72cf8e416 148 LOG(AZ_LOG_TRACE, 0, "%s", name);
AzureIoTClient 16:22a72cf8e416 149 LOG(AZ_LOG_TRACE, 0, "%s", (value_as_string = amqpvalue_to_string(value)));
AzureIoTClient 6:641a9672db08 150 if (value_as_string != NULL)
AzureIoTClient 6:641a9672db08 151 {
AzureIoTClient 6:641a9672db08 152 amqpalloc_free(value_as_string);
AzureIoTClient 6:641a9672db08 153 }
AzureIoTClient 6:641a9672db08 154 }
AzureIoTClient 9:c22db038556c 155 #endif
Azure.IoT Build 0:6ae2f7bca550 156 }
Azure.IoT Build 0:6ae2f7bca550 157
Azure.IoT Build 0:6ae2f7bca550 158 static SEND_ONE_MESSAGE_RESULT send_one_message(MESSAGE_SENDER_INSTANCE* message_sender_instance, MESSAGE_WITH_CALLBACK* message_with_callback, MESSAGE_HANDLE message)
Azure.IoT Build 0:6ae2f7bca550 159 {
AzureIoTClient 6:641a9672db08 160 SEND_ONE_MESSAGE_RESULT result;
Azure.IoT Build 0:6ae2f7bca550 161
AzureIoTClient 6:641a9672db08 162 size_t encoded_size;
AzureIoTClient 6:641a9672db08 163 size_t total_encoded_size = 0;
AzureIoTClient 6:641a9672db08 164 MESSAGE_BODY_TYPE message_body_type;
Azure.IoT Build 0:6ae2f7bca550 165 message_format message_format;
Azure.IoT Build 0:6ae2f7bca550 166
AzureIoTClient 6:641a9672db08 167 if ((message_get_body_type(message, &message_body_type) != 0) ||
Azure.IoT Build 0:6ae2f7bca550 168 (message_get_message_format(message, &message_format) != 0))
AzureIoTClient 6:641a9672db08 169 {
AzureIoTClient 6:641a9672db08 170 result = SEND_ONE_MESSAGE_ERROR;
AzureIoTClient 6:641a9672db08 171 }
AzureIoTClient 6:641a9672db08 172 else
AzureIoTClient 6:641a9672db08 173 {
AzureIoTClient 6:641a9672db08 174 // header
AzureIoTClient 6:641a9672db08 175 HEADER_HANDLE header;
AzureIoTClient 6:641a9672db08 176 AMQP_VALUE header_amqp_value;
AzureIoTClient 6:641a9672db08 177 PROPERTIES_HANDLE properties;
AzureIoTClient 6:641a9672db08 178 AMQP_VALUE properties_amqp_value;
AzureIoTClient 6:641a9672db08 179 AMQP_VALUE application_properties;
AzureIoTClient 6:641a9672db08 180 AMQP_VALUE application_properties_value;
AzureIoTClient 6:641a9672db08 181 AMQP_VALUE body_amqp_value = NULL;
AzureIoTClient 6:641a9672db08 182 size_t body_data_count = 0;
AzureIoTClient 13:9abd748f4e78 183 AMQP_VALUE msg_annotations = NULL;
Azure.IoT Build 0:6ae2f7bca550 184
AzureIoTClient 6:641a9672db08 185 message_get_header(message, &header);
AzureIoTClient 6:641a9672db08 186 header_amqp_value = amqpvalue_create_header(header);
AzureIoTClient 6:641a9672db08 187 if (header != NULL)
AzureIoTClient 6:641a9672db08 188 {
AzureIoTClient 6:641a9672db08 189 amqpvalue_get_encoded_size(header_amqp_value, &encoded_size);
AzureIoTClient 6:641a9672db08 190 total_encoded_size += encoded_size;
AzureIoTClient 6:641a9672db08 191 }
Azure.IoT Build 0:6ae2f7bca550 192
AzureIoTClient 13:9abd748f4e78 193 // message annotations
AzureIoTClient 13:9abd748f4e78 194 if ((message_get_message_annotations(message, &msg_annotations) == 0) &&
AzureIoTClient 13:9abd748f4e78 195 (msg_annotations != NULL))
AzureIoTClient 13:9abd748f4e78 196 {
AzureIoTClient 13:9abd748f4e78 197 amqpvalue_get_encoded_size(msg_annotations, &encoded_size);
AzureIoTClient 13:9abd748f4e78 198 total_encoded_size += encoded_size;
AzureIoTClient 13:9abd748f4e78 199 }
AzureIoTClient 13:9abd748f4e78 200
AzureIoTClient 6:641a9672db08 201 // properties
AzureIoTClient 6:641a9672db08 202 message_get_properties(message, &properties);
AzureIoTClient 6:641a9672db08 203 properties_amqp_value = amqpvalue_create_properties(properties);
AzureIoTClient 6:641a9672db08 204 if (properties != NULL)
AzureIoTClient 6:641a9672db08 205 {
AzureIoTClient 6:641a9672db08 206 amqpvalue_get_encoded_size(properties_amqp_value, &encoded_size);
AzureIoTClient 6:641a9672db08 207 total_encoded_size += encoded_size;
AzureIoTClient 6:641a9672db08 208 }
Azure.IoT Build 0:6ae2f7bca550 209
AzureIoTClient 6:641a9672db08 210 // application properties
AzureIoTClient 6:641a9672db08 211 message_get_application_properties(message, &application_properties);
AzureIoTClient 6:641a9672db08 212 application_properties_value = amqpvalue_create_application_properties(application_properties);
AzureIoTClient 6:641a9672db08 213 if (application_properties != NULL)
AzureIoTClient 6:641a9672db08 214 {
AzureIoTClient 6:641a9672db08 215 amqpvalue_get_encoded_size(application_properties_value, &encoded_size);
AzureIoTClient 6:641a9672db08 216 total_encoded_size += encoded_size;
AzureIoTClient 6:641a9672db08 217 }
Azure.IoT Build 0:6ae2f7bca550 218
AzureIoTClient 6:641a9672db08 219 result = SEND_ONE_MESSAGE_OK;
Azure.IoT Build 0:6ae2f7bca550 220
AzureIoTClient 6:641a9672db08 221 // body - amqp data
AzureIoTClient 6:641a9672db08 222 switch (message_body_type)
AzureIoTClient 6:641a9672db08 223 {
AzureIoTClient 6:641a9672db08 224 default:
AzureIoTClient 6:641a9672db08 225 result = SEND_ONE_MESSAGE_ERROR;
AzureIoTClient 6:641a9672db08 226 break;
Azure.IoT Build 0:6ae2f7bca550 227
AzureIoTClient 6:641a9672db08 228 case MESSAGE_BODY_TYPE_VALUE:
AzureIoTClient 6:641a9672db08 229 {
AzureIoTClient 6:641a9672db08 230 AMQP_VALUE message_body_amqp_value;
AzureIoTClient 6:641a9672db08 231 if (message_get_inplace_body_amqp_value(message, &message_body_amqp_value) != 0)
AzureIoTClient 6:641a9672db08 232 {
AzureIoTClient 6:641a9672db08 233 result = SEND_ONE_MESSAGE_ERROR;
AzureIoTClient 6:641a9672db08 234 }
AzureIoTClient 6:641a9672db08 235 else
AzureIoTClient 6:641a9672db08 236 {
AzureIoTClient 6:641a9672db08 237 body_amqp_value = amqpvalue_create_amqp_value(message_body_amqp_value);
AzureIoTClient 6:641a9672db08 238 if ((body_amqp_value == NULL) ||
AzureIoTClient 6:641a9672db08 239 (amqpvalue_get_encoded_size(body_amqp_value, &encoded_size) != 0))
AzureIoTClient 6:641a9672db08 240 {
AzureIoTClient 6:641a9672db08 241 result = SEND_ONE_MESSAGE_ERROR;
AzureIoTClient 6:641a9672db08 242 }
AzureIoTClient 6:641a9672db08 243 else
AzureIoTClient 6:641a9672db08 244 {
AzureIoTClient 6:641a9672db08 245 total_encoded_size += encoded_size;
AzureIoTClient 6:641a9672db08 246 }
AzureIoTClient 6:641a9672db08 247 }
Azure.IoT Build 0:6ae2f7bca550 248
AzureIoTClient 6:641a9672db08 249 break;
AzureIoTClient 6:641a9672db08 250 }
Azure.IoT Build 0:6ae2f7bca550 251
AzureIoTClient 6:641a9672db08 252 case MESSAGE_BODY_TYPE_DATA:
AzureIoTClient 6:641a9672db08 253 {
AzureIoTClient 6:641a9672db08 254 BINARY_DATA binary_data;
Azure.IoT Build 0:6ae2f7bca550 255 size_t i;
Azure.IoT Build 0:6ae2f7bca550 256
Azure.IoT Build 0:6ae2f7bca550 257 if (message_get_body_amqp_data_count(message, &body_data_count) != 0)
Azure.IoT Build 0:6ae2f7bca550 258 {
Azure.IoT Build 0:6ae2f7bca550 259 result = SEND_ONE_MESSAGE_ERROR;
Azure.IoT Build 0:6ae2f7bca550 260 }
Azure.IoT Build 0:6ae2f7bca550 261 else
Azure.IoT Build 0:6ae2f7bca550 262 {
Azure.IoT Build 0:6ae2f7bca550 263 for (i = 0; i < body_data_count; i++)
Azure.IoT Build 0:6ae2f7bca550 264 {
Azure.IoT Build 0:6ae2f7bca550 265 if (message_get_body_amqp_data(message, i, &binary_data) != 0)
Azure.IoT Build 0:6ae2f7bca550 266 {
Azure.IoT Build 0:6ae2f7bca550 267 result = SEND_ONE_MESSAGE_ERROR;
Azure.IoT Build 0:6ae2f7bca550 268 }
Azure.IoT Build 0:6ae2f7bca550 269 else
Azure.IoT Build 0:6ae2f7bca550 270 {
AzureIoTClient 6:641a9672db08 271 amqp_binary binary_value;
AzureIoTClient 6:641a9672db08 272 binary_value.bytes = binary_data.bytes;
AzureIoTClient 6:641a9672db08 273 binary_value.length = (uint32_t)binary_data.length;
Azure.IoT Build 0:6ae2f7bca550 274 AMQP_VALUE body_amqp_data = amqpvalue_create_data(binary_value);
Azure.IoT Build 0:6ae2f7bca550 275 if (body_amqp_data == NULL)
Azure.IoT Build 0:6ae2f7bca550 276 {
Azure.IoT Build 0:6ae2f7bca550 277 result = SEND_ONE_MESSAGE_ERROR;
Azure.IoT Build 0:6ae2f7bca550 278 }
Azure.IoT Build 0:6ae2f7bca550 279 else
Azure.IoT Build 0:6ae2f7bca550 280 {
Azure.IoT Build 0:6ae2f7bca550 281 if (amqpvalue_get_encoded_size(body_amqp_data, &encoded_size) != 0)
Azure.IoT Build 0:6ae2f7bca550 282 {
Azure.IoT Build 0:6ae2f7bca550 283 result = SEND_ONE_MESSAGE_ERROR;
Azure.IoT Build 0:6ae2f7bca550 284 }
Azure.IoT Build 0:6ae2f7bca550 285 else
Azure.IoT Build 0:6ae2f7bca550 286 {
Azure.IoT Build 0:6ae2f7bca550 287 total_encoded_size += encoded_size;
Azure.IoT Build 0:6ae2f7bca550 288 }
Azure.IoT Build 0:6ae2f7bca550 289
Azure.IoT Build 0:6ae2f7bca550 290 amqpvalue_destroy(body_amqp_data);
Azure.IoT Build 0:6ae2f7bca550 291 }
Azure.IoT Build 0:6ae2f7bca550 292 }
Azure.IoT Build 0:6ae2f7bca550 293 }
Azure.IoT Build 0:6ae2f7bca550 294 }
AzureIoTClient 6:641a9672db08 295 break;
AzureIoTClient 6:641a9672db08 296 }
AzureIoTClient 6:641a9672db08 297 }
Azure.IoT Build 0:6ae2f7bca550 298
AzureIoTClient 6:641a9672db08 299 if (result == 0)
AzureIoTClient 6:641a9672db08 300 {
AzureIoTClient 6:641a9672db08 301 void* data_bytes = amqpalloc_malloc(total_encoded_size);
AzureIoTClient 6:641a9672db08 302 PAYLOAD payload;
AzureIoTClient 6:641a9672db08 303 payload.bytes = data_bytes;
AzureIoTClient 6:641a9672db08 304 payload.length = 0;
AzureIoTClient 6:641a9672db08 305 result = SEND_ONE_MESSAGE_OK;
Azure.IoT Build 0:6ae2f7bca550 306
AzureIoTClient 6:641a9672db08 307 if (header != NULL)
AzureIoTClient 6:641a9672db08 308 {
AzureIoTClient 6:641a9672db08 309 if (amqpvalue_encode(header_amqp_value, encode_bytes, &payload) != 0)
AzureIoTClient 6:641a9672db08 310 {
AzureIoTClient 6:641a9672db08 311 result = SEND_ONE_MESSAGE_ERROR;
AzureIoTClient 6:641a9672db08 312 }
Azure.IoT Build 0:6ae2f7bca550 313
AzureIoTClient 6:641a9672db08 314 log_message_chunk(message_sender_instance, "Header:", header_amqp_value);
AzureIoTClient 6:641a9672db08 315 }
Azure.IoT Build 0:6ae2f7bca550 316
AzureIoTClient 13:9abd748f4e78 317 if ((result == SEND_ONE_MESSAGE_OK) && (msg_annotations != NULL))
AzureIoTClient 13:9abd748f4e78 318 {
AzureIoTClient 13:9abd748f4e78 319 if (amqpvalue_encode(msg_annotations, encode_bytes, &payload) != 0)
AzureIoTClient 13:9abd748f4e78 320 {
AzureIoTClient 13:9abd748f4e78 321 result = SEND_ONE_MESSAGE_ERROR;
AzureIoTClient 13:9abd748f4e78 322 }
AzureIoTClient 13:9abd748f4e78 323
AzureIoTClient 13:9abd748f4e78 324 log_message_chunk(message_sender_instance, "Message Annotations:", msg_annotations);
AzureIoTClient 13:9abd748f4e78 325 }
AzureIoTClient 13:9abd748f4e78 326
AzureIoTClient 6:641a9672db08 327 if ((result == SEND_ONE_MESSAGE_OK) && (properties != NULL))
AzureIoTClient 6:641a9672db08 328 {
AzureIoTClient 6:641a9672db08 329 if (amqpvalue_encode(properties_amqp_value, encode_bytes, &payload) != 0)
AzureIoTClient 6:641a9672db08 330 {
AzureIoTClient 6:641a9672db08 331 result = SEND_ONE_MESSAGE_ERROR;
AzureIoTClient 6:641a9672db08 332 }
Azure.IoT Build 0:6ae2f7bca550 333
AzureIoTClient 6:641a9672db08 334 log_message_chunk(message_sender_instance, "Properties:", properties_amqp_value);
AzureIoTClient 6:641a9672db08 335 }
Azure.IoT Build 0:6ae2f7bca550 336
AzureIoTClient 6:641a9672db08 337 if ((result == SEND_ONE_MESSAGE_OK) && (application_properties != NULL))
AzureIoTClient 6:641a9672db08 338 {
AzureIoTClient 6:641a9672db08 339 if (amqpvalue_encode(application_properties_value, encode_bytes, &payload) != 0)
AzureIoTClient 6:641a9672db08 340 {
AzureIoTClient 6:641a9672db08 341 result = SEND_ONE_MESSAGE_ERROR;
AzureIoTClient 6:641a9672db08 342 }
Azure.IoT Build 0:6ae2f7bca550 343
AzureIoTClient 6:641a9672db08 344 log_message_chunk(message_sender_instance, "Application properties:", application_properties_value);
AzureIoTClient 6:641a9672db08 345 }
Azure.IoT Build 0:6ae2f7bca550 346
AzureIoTClient 6:641a9672db08 347 if (result == SEND_ONE_MESSAGE_OK)
AzureIoTClient 6:641a9672db08 348 {
AzureIoTClient 6:641a9672db08 349 switch (message_body_type)
AzureIoTClient 6:641a9672db08 350 {
AzureIoTClient 6:641a9672db08 351 case MESSAGE_BODY_TYPE_VALUE:
AzureIoTClient 6:641a9672db08 352 {
AzureIoTClient 6:641a9672db08 353 if (amqpvalue_encode(body_amqp_value, encode_bytes, &payload) != 0)
AzureIoTClient 6:641a9672db08 354 {
AzureIoTClient 6:641a9672db08 355 result = SEND_ONE_MESSAGE_ERROR;
AzureIoTClient 6:641a9672db08 356 }
Azure.IoT Build 0:6ae2f7bca550 357
AzureIoTClient 6:641a9672db08 358 log_message_chunk(message_sender_instance, "Body - amqp value:", body_amqp_value);
AzureIoTClient 6:641a9672db08 359 break;
AzureIoTClient 6:641a9672db08 360 }
AzureIoTClient 6:641a9672db08 361 case MESSAGE_BODY_TYPE_DATA:
AzureIoTClient 6:641a9672db08 362 {
Azure.IoT Build 0:6ae2f7bca550 363 BINARY_DATA binary_data;
Azure.IoT Build 0:6ae2f7bca550 364 size_t i;
Azure.IoT Build 0:6ae2f7bca550 365
Azure.IoT Build 0:6ae2f7bca550 366 for (i = 0; i < body_data_count; i++)
Azure.IoT Build 0:6ae2f7bca550 367 {
Azure.IoT Build 0:6ae2f7bca550 368 if (message_get_body_amqp_data(message, i, &binary_data) != 0)
Azure.IoT Build 0:6ae2f7bca550 369 {
Azure.IoT Build 0:6ae2f7bca550 370 result = SEND_ONE_MESSAGE_ERROR;
Azure.IoT Build 0:6ae2f7bca550 371 }
Azure.IoT Build 0:6ae2f7bca550 372 else
Azure.IoT Build 0:6ae2f7bca550 373 {
AzureIoTClient 6:641a9672db08 374 amqp_binary binary_value;
AzureIoTClient 6:641a9672db08 375 binary_value.bytes = binary_data.bytes;
AzureIoTClient 6:641a9672db08 376 binary_value.length = (uint32_t)binary_data.length;
Azure.IoT Build 0:6ae2f7bca550 377 AMQP_VALUE body_amqp_data = amqpvalue_create_data(binary_value);
Azure.IoT Build 0:6ae2f7bca550 378 if (body_amqp_data == NULL)
Azure.IoT Build 0:6ae2f7bca550 379 {
Azure.IoT Build 0:6ae2f7bca550 380 result = SEND_ONE_MESSAGE_ERROR;
Azure.IoT Build 0:6ae2f7bca550 381 }
Azure.IoT Build 0:6ae2f7bca550 382 else
Azure.IoT Build 0:6ae2f7bca550 383 {
Azure.IoT Build 0:6ae2f7bca550 384 if (amqpvalue_encode(body_amqp_data, encode_bytes, &payload) != 0)
Azure.IoT Build 0:6ae2f7bca550 385 {
Azure.IoT Build 0:6ae2f7bca550 386 result = SEND_ONE_MESSAGE_ERROR;
Azure.IoT Build 0:6ae2f7bca550 387 break;
Azure.IoT Build 0:6ae2f7bca550 388 }
Azure.IoT Build 0:6ae2f7bca550 389
Azure.IoT Build 0:6ae2f7bca550 390 amqpvalue_destroy(body_amqp_data);
Azure.IoT Build 0:6ae2f7bca550 391 }
Azure.IoT Build 0:6ae2f7bca550 392 }
Azure.IoT Build 0:6ae2f7bca550 393 }
AzureIoTClient 6:641a9672db08 394 break;
AzureIoTClient 6:641a9672db08 395 }
AzureIoTClient 6:641a9672db08 396 }
AzureIoTClient 6:641a9672db08 397 }
Azure.IoT Build 0:6ae2f7bca550 398
AzureIoTClient 6:641a9672db08 399 if (result == SEND_ONE_MESSAGE_OK)
AzureIoTClient 6:641a9672db08 400 {
AzureIoTClient 6:641a9672db08 401 message_with_callback->message_send_state = MESSAGE_SEND_STATE_PENDING;
AzureIoTClient 6:641a9672db08 402 switch (link_transfer(message_sender_instance->link, message_format, &payload, 1, on_delivery_settled, message_with_callback))
AzureIoTClient 6:641a9672db08 403 {
AzureIoTClient 6:641a9672db08 404 default:
AzureIoTClient 6:641a9672db08 405 case LINK_TRANSFER_ERROR:
AzureIoTClient 6:641a9672db08 406 if (message_with_callback->on_message_send_complete != NULL)
AzureIoTClient 6:641a9672db08 407 {
AzureIoTClient 6:641a9672db08 408 message_with_callback->on_message_send_complete(message_with_callback->context, MESSAGE_SEND_ERROR);
AzureIoTClient 6:641a9672db08 409 }
Azure.IoT Build 0:6ae2f7bca550 410
AzureIoTClient 6:641a9672db08 411 result = SEND_ONE_MESSAGE_ERROR;
AzureIoTClient 6:641a9672db08 412 break;
Azure.IoT Build 0:6ae2f7bca550 413
AzureIoTClient 6:641a9672db08 414 case LINK_TRANSFER_BUSY:
AzureIoTClient 6:641a9672db08 415 message_with_callback->message_send_state = MESSAGE_SEND_STATE_NOT_SENT;
AzureIoTClient 6:641a9672db08 416 result = SEND_ONE_MESSAGE_BUSY;
AzureIoTClient 6:641a9672db08 417 break;
Azure.IoT Build 0:6ae2f7bca550 418
AzureIoTClient 6:641a9672db08 419 case LINK_TRANSFER_OK:
AzureIoTClient 6:641a9672db08 420 result = SEND_ONE_MESSAGE_OK;
AzureIoTClient 6:641a9672db08 421 break;
AzureIoTClient 6:641a9672db08 422 }
AzureIoTClient 6:641a9672db08 423 }
Azure.IoT Build 0:6ae2f7bca550 424
AzureIoTClient 6:641a9672db08 425 amqpalloc_free(data_bytes);
Azure.IoT Build 0:6ae2f7bca550 426
AzureIoTClient 6:641a9672db08 427 if (body_amqp_value != NULL)
AzureIoTClient 6:641a9672db08 428 {
AzureIoTClient 6:641a9672db08 429 amqpvalue_destroy(body_amqp_value);
AzureIoTClient 6:641a9672db08 430 }
Azure.IoT Build 0:6ae2f7bca550 431
AzureIoTClient 6:641a9672db08 432 amqpvalue_destroy(application_properties);
AzureIoTClient 6:641a9672db08 433 amqpvalue_destroy(application_properties_value);
AzureIoTClient 6:641a9672db08 434 amqpvalue_destroy(properties_amqp_value);
AzureIoTClient 6:641a9672db08 435 properties_destroy(properties);
AzureIoTClient 6:641a9672db08 436 }
AzureIoTClient 6:641a9672db08 437 }
Azure.IoT Build 0:6ae2f7bca550 438
AzureIoTClient 6:641a9672db08 439 return result;
Azure.IoT Build 0:6ae2f7bca550 440 }
Azure.IoT Build 0:6ae2f7bca550 441
Azure.IoT Build 0:6ae2f7bca550 442 static void send_all_pending_messages(MESSAGE_SENDER_INSTANCE* message_sender_instance)
Azure.IoT Build 0:6ae2f7bca550 443 {
AzureIoTClient 6:641a9672db08 444 size_t i;
Azure.IoT Build 0:6ae2f7bca550 445
AzureIoTClient 6:641a9672db08 446 for (i = 0; i < message_sender_instance->message_count; i++)
AzureIoTClient 6:641a9672db08 447 {
AzureIoTClient 6:641a9672db08 448 if (message_sender_instance->messages[i]->message_send_state == MESSAGE_SEND_STATE_NOT_SENT)
AzureIoTClient 6:641a9672db08 449 {
AzureIoTClient 6:641a9672db08 450 switch (send_one_message(message_sender_instance, message_sender_instance->messages[i], message_sender_instance->messages[i]->message))
AzureIoTClient 6:641a9672db08 451 {
AzureIoTClient 6:641a9672db08 452 default:
AzureIoTClient 6:641a9672db08 453 case SEND_ONE_MESSAGE_ERROR:
AzureIoTClient 6:641a9672db08 454 {
AzureIoTClient 6:641a9672db08 455 ON_MESSAGE_SEND_COMPLETE on_message_send_complete = message_sender_instance->messages[i]->on_message_send_complete;
AzureIoTClient 6:641a9672db08 456 void* context = message_sender_instance->messages[i]->context;
AzureIoTClient 6:641a9672db08 457 remove_pending_message_by_index(message_sender_instance, i);
Azure.IoT Build 0:6ae2f7bca550 458
AzureIoTClient 6:641a9672db08 459 on_message_send_complete(context, MESSAGE_SEND_ERROR);
AzureIoTClient 6:641a9672db08 460 i = message_sender_instance->message_count;
AzureIoTClient 6:641a9672db08 461 break;
AzureIoTClient 6:641a9672db08 462 }
AzureIoTClient 6:641a9672db08 463 case SEND_ONE_MESSAGE_BUSY:
AzureIoTClient 6:641a9672db08 464 i = message_sender_instance->message_count + 1;
AzureIoTClient 6:641a9672db08 465 break;
Azure.IoT Build 0:6ae2f7bca550 466
AzureIoTClient 6:641a9672db08 467 case SEND_ONE_MESSAGE_OK:
AzureIoTClient 6:641a9672db08 468 break;
AzureIoTClient 6:641a9672db08 469 }
Azure.IoT Build 0:6ae2f7bca550 470
AzureIoTClient 6:641a9672db08 471 i--;
AzureIoTClient 6:641a9672db08 472 }
AzureIoTClient 6:641a9672db08 473 }
Azure.IoT Build 0:6ae2f7bca550 474 }
Azure.IoT Build 0:6ae2f7bca550 475
Azure.IoT Build 0:6ae2f7bca550 476 static void set_message_sender_state(MESSAGE_SENDER_INSTANCE* message_sender_instance, MESSAGE_SENDER_STATE new_state)
Azure.IoT Build 0:6ae2f7bca550 477 {
AzureIoTClient 6:641a9672db08 478 MESSAGE_SENDER_STATE previous_state = message_sender_instance->message_sender_state;
AzureIoTClient 6:641a9672db08 479 message_sender_instance->message_sender_state = new_state;
AzureIoTClient 6:641a9672db08 480 if (message_sender_instance->on_message_sender_state_changed != NULL)
AzureIoTClient 6:641a9672db08 481 {
AzureIoTClient 6:641a9672db08 482 message_sender_instance->on_message_sender_state_changed(message_sender_instance->on_message_sender_state_changed_context, new_state, previous_state);
AzureIoTClient 6:641a9672db08 483 }
Azure.IoT Build 0:6ae2f7bca550 484 }
Azure.IoT Build 0:6ae2f7bca550 485
AzureIoTClient 2:64b4feb67cd3 486 static void indicate_all_messages_as_error(MESSAGE_SENDER_INSTANCE* message_sender_instance)
AzureIoTClient 2:64b4feb67cd3 487 {
AzureIoTClient 2:64b4feb67cd3 488 size_t i;
AzureIoTClient 2:64b4feb67cd3 489
AzureIoTClient 2:64b4feb67cd3 490 for (i = 0; i < message_sender_instance->message_count; i++)
AzureIoTClient 2:64b4feb67cd3 491 {
AzureIoTClient 2:64b4feb67cd3 492 if (message_sender_instance->messages[i]->on_message_send_complete != NULL)
AzureIoTClient 2:64b4feb67cd3 493 {
AzureIoTClient 2:64b4feb67cd3 494 message_sender_instance->messages[i]->on_message_send_complete(message_sender_instance->messages[i]->context, MESSAGE_SEND_ERROR);
AzureIoTClient 2:64b4feb67cd3 495 }
AzureIoTClient 2:64b4feb67cd3 496
AzureIoTClient 2:64b4feb67cd3 497 message_destroy(message_sender_instance->messages[i]->message);
AzureIoTClient 2:64b4feb67cd3 498 amqpalloc_free(message_sender_instance->messages[i]);
AzureIoTClient 2:64b4feb67cd3 499 }
AzureIoTClient 2:64b4feb67cd3 500
AzureIoTClient 2:64b4feb67cd3 501 if (message_sender_instance->messages != NULL)
AzureIoTClient 2:64b4feb67cd3 502 {
AzureIoTClient 2:64b4feb67cd3 503 message_sender_instance->message_count = 0;
AzureIoTClient 2:64b4feb67cd3 504
AzureIoTClient 2:64b4feb67cd3 505 amqpalloc_free(message_sender_instance->messages);
AzureIoTClient 2:64b4feb67cd3 506 message_sender_instance->messages = NULL;
AzureIoTClient 2:64b4feb67cd3 507 }
AzureIoTClient 2:64b4feb67cd3 508 }
AzureIoTClient 2:64b4feb67cd3 509
Azure.IoT Build 0:6ae2f7bca550 510 static void on_link_state_changed(void* context, LINK_STATE new_link_state, LINK_STATE previous_link_state)
Azure.IoT Build 0:6ae2f7bca550 511 {
AzureIoTClient 6:641a9672db08 512 MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)context;
AzureIoTClient 6:641a9672db08 513 (void)previous_link_state;
Azure.IoT Build 0:6ae2f7bca550 514
AzureIoTClient 6:641a9672db08 515 switch (new_link_state)
AzureIoTClient 6:641a9672db08 516 {
AzureIoTClient 6:641a9672db08 517 case LINK_STATE_ATTACHED:
AzureIoTClient 6:641a9672db08 518 if (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_OPENING)
AzureIoTClient 6:641a9672db08 519 {
AzureIoTClient 6:641a9672db08 520 set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_OPEN);
AzureIoTClient 6:641a9672db08 521 }
AzureIoTClient 6:641a9672db08 522 break;
AzureIoTClient 6:641a9672db08 523 case LINK_STATE_DETACHED:
Azure.IoT Build 0:6ae2f7bca550 524 if ((message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_OPEN) ||
Azure.IoT Build 0:6ae2f7bca550 525 (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_CLOSING))
Azure.IoT Build 0:6ae2f7bca550 526 {
Azure.IoT Build 0:6ae2f7bca550 527 /* User initiated transition, we should be good */
AzureIoTClient 2:64b4feb67cd3 528 indicate_all_messages_as_error(message_sender_instance);
Azure.IoT Build 0:6ae2f7bca550 529 set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_IDLE);
Azure.IoT Build 0:6ae2f7bca550 530 }
Azure.IoT Build 0:6ae2f7bca550 531 else if (message_sender_instance->message_sender_state != MESSAGE_SENDER_STATE_IDLE)
Azure.IoT Build 0:6ae2f7bca550 532 {
Azure.IoT Build 0:6ae2f7bca550 533 /* Any other transition must be an error */
Azure.IoT Build 0:6ae2f7bca550 534 set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_ERROR);
Azure.IoT Build 0:6ae2f7bca550 535 }
Azure.IoT Build 0:6ae2f7bca550 536 break;
AzureIoTClient 6:641a9672db08 537 case LINK_STATE_ERROR:
AzureIoTClient 6:641a9672db08 538 if (message_sender_instance->message_sender_state != MESSAGE_SENDER_STATE_ERROR)
AzureIoTClient 6:641a9672db08 539 {
AzureIoTClient 2:64b4feb67cd3 540 indicate_all_messages_as_error(message_sender_instance);
AzureIoTClient 6:641a9672db08 541 set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_ERROR);
AzureIoTClient 6:641a9672db08 542 }
AzureIoTClient 6:641a9672db08 543 break;
AzureIoTClient 6:641a9672db08 544 }
Azure.IoT Build 0:6ae2f7bca550 545 }
Azure.IoT Build 0:6ae2f7bca550 546
Azure.IoT Build 0:6ae2f7bca550 547 static void on_link_flow_on(void* context)
Azure.IoT Build 0:6ae2f7bca550 548 {
AzureIoTClient 6:641a9672db08 549 MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)context;
AzureIoTClient 6:641a9672db08 550 send_all_pending_messages(message_sender_instance);
Azure.IoT Build 0:6ae2f7bca550 551 }
Azure.IoT Build 0:6ae2f7bca550 552
Azure.IoT Build 5:ae49385aff34 553 MESSAGE_SENDER_HANDLE messagesender_create(LINK_HANDLE link, ON_MESSAGE_SENDER_STATE_CHANGED on_message_sender_state_changed, void* context)
Azure.IoT Build 0:6ae2f7bca550 554 {
AzureIoTClient 6:641a9672db08 555 MESSAGE_SENDER_INSTANCE* result = amqpalloc_malloc(sizeof(MESSAGE_SENDER_INSTANCE));
AzureIoTClient 6:641a9672db08 556 if (result != NULL)
AzureIoTClient 6:641a9672db08 557 {
AzureIoTClient 6:641a9672db08 558 result->messages = NULL;
AzureIoTClient 6:641a9672db08 559 result->message_count = 0;
AzureIoTClient 6:641a9672db08 560 result->link = link;
AzureIoTClient 6:641a9672db08 561 result->on_message_sender_state_changed = on_message_sender_state_changed;
AzureIoTClient 6:641a9672db08 562 result->on_message_sender_state_changed_context = context;
AzureIoTClient 6:641a9672db08 563 result->message_sender_state = MESSAGE_SENDER_STATE_IDLE;
AzureIoTClient 6:641a9672db08 564 result->is_trace_on = 0;
AzureIoTClient 6:641a9672db08 565 }
Azure.IoT Build 0:6ae2f7bca550 566
AzureIoTClient 6:641a9672db08 567 return result;
Azure.IoT Build 0:6ae2f7bca550 568 }
Azure.IoT Build 0:6ae2f7bca550 569
Azure.IoT Build 0:6ae2f7bca550 570 void messagesender_destroy(MESSAGE_SENDER_HANDLE message_sender)
Azure.IoT Build 0:6ae2f7bca550 571 {
AzureIoTClient 6:641a9672db08 572 if (message_sender != NULL)
AzureIoTClient 6:641a9672db08 573 {
AzureIoTClient 6:641a9672db08 574 MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_sender;
Azure.IoT Build 0:6ae2f7bca550 575
AzureIoTClient 6:641a9672db08 576 messagesender_close(message_sender_instance);
Azure.IoT Build 0:6ae2f7bca550 577
AzureIoTClient 2:64b4feb67cd3 578 indicate_all_messages_as_error(message_sender_instance);
Azure.IoT Build 0:6ae2f7bca550 579
AzureIoTClient 6:641a9672db08 580 amqpalloc_free(message_sender);
AzureIoTClient 6:641a9672db08 581 }
Azure.IoT Build 0:6ae2f7bca550 582 }
Azure.IoT Build 0:6ae2f7bca550 583
Azure.IoT Build 0:6ae2f7bca550 584 int messagesender_open(MESSAGE_SENDER_HANDLE message_sender)
Azure.IoT Build 0:6ae2f7bca550 585 {
AzureIoTClient 6:641a9672db08 586 int result;
Azure.IoT Build 0:6ae2f7bca550 587
AzureIoTClient 6:641a9672db08 588 if (message_sender == NULL)
AzureIoTClient 6:641a9672db08 589 {
AzureIoTClient 6:641a9672db08 590 result = __LINE__;
AzureIoTClient 6:641a9672db08 591 }
AzureIoTClient 6:641a9672db08 592 else
AzureIoTClient 6:641a9672db08 593 {
AzureIoTClient 6:641a9672db08 594 MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_sender;
Azure.IoT Build 0:6ae2f7bca550 595
AzureIoTClient 6:641a9672db08 596 if (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_IDLE)
AzureIoTClient 6:641a9672db08 597 {
AzureIoTClient 6:641a9672db08 598 set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_OPENING);
AzureIoTClient 6:641a9672db08 599 if (link_attach(message_sender_instance->link, NULL, on_link_state_changed, on_link_flow_on, message_sender_instance) != 0)
AzureIoTClient 6:641a9672db08 600 {
AzureIoTClient 6:641a9672db08 601 result = __LINE__;
AzureIoTClient 6:641a9672db08 602 set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_ERROR);
AzureIoTClient 6:641a9672db08 603 }
AzureIoTClient 6:641a9672db08 604 else
AzureIoTClient 6:641a9672db08 605 {
AzureIoTClient 6:641a9672db08 606 result = 0;
AzureIoTClient 6:641a9672db08 607 }
AzureIoTClient 6:641a9672db08 608 }
AzureIoTClient 6:641a9672db08 609 else
AzureIoTClient 6:641a9672db08 610 {
AzureIoTClient 6:641a9672db08 611 result = 0;
AzureIoTClient 6:641a9672db08 612 }
AzureIoTClient 6:641a9672db08 613 }
Azure.IoT Build 0:6ae2f7bca550 614
AzureIoTClient 6:641a9672db08 615 return result;
Azure.IoT Build 0:6ae2f7bca550 616 }
Azure.IoT Build 0:6ae2f7bca550 617
Azure.IoT Build 0:6ae2f7bca550 618 int messagesender_close(MESSAGE_SENDER_HANDLE message_sender)
Azure.IoT Build 0:6ae2f7bca550 619 {
AzureIoTClient 6:641a9672db08 620 int result;
Azure.IoT Build 0:6ae2f7bca550 621
AzureIoTClient 6:641a9672db08 622 if (message_sender == NULL)
AzureIoTClient 6:641a9672db08 623 {
AzureIoTClient 6:641a9672db08 624 result = __LINE__;
AzureIoTClient 6:641a9672db08 625 }
AzureIoTClient 6:641a9672db08 626 else
AzureIoTClient 6:641a9672db08 627 {
AzureIoTClient 6:641a9672db08 628 MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_sender;
Azure.IoT Build 0:6ae2f7bca550 629
AzureIoTClient 6:641a9672db08 630 if ((message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_OPENING) ||
AzureIoTClient 6:641a9672db08 631 (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_OPEN))
AzureIoTClient 6:641a9672db08 632 {
AzureIoTClient 6:641a9672db08 633 set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_CLOSING);
AzureIoTClient 12:b30dacf113f2 634 if (link_detach(message_sender_instance->link, true) != 0)
AzureIoTClient 6:641a9672db08 635 {
AzureIoTClient 6:641a9672db08 636 result = __LINE__;
AzureIoTClient 6:641a9672db08 637 set_message_sender_state(message_sender_instance, MESSAGE_SENDER_STATE_ERROR);
AzureIoTClient 6:641a9672db08 638 }
AzureIoTClient 6:641a9672db08 639 else
AzureIoTClient 6:641a9672db08 640 {
AzureIoTClient 6:641a9672db08 641 result = 0;
AzureIoTClient 6:641a9672db08 642 }
AzureIoTClient 6:641a9672db08 643 }
AzureIoTClient 6:641a9672db08 644 else
AzureIoTClient 6:641a9672db08 645 {
AzureIoTClient 6:641a9672db08 646 result = 0;
AzureIoTClient 6:641a9672db08 647 }
AzureIoTClient 6:641a9672db08 648 }
Azure.IoT Build 0:6ae2f7bca550 649
AzureIoTClient 6:641a9672db08 650 return result;
Azure.IoT Build 0:6ae2f7bca550 651 }
Azure.IoT Build 0:6ae2f7bca550 652
Azure.IoT Build 0:6ae2f7bca550 653 int messagesender_send(MESSAGE_SENDER_HANDLE message_sender, MESSAGE_HANDLE message, ON_MESSAGE_SEND_COMPLETE on_message_send_complete, void* callback_context)
Azure.IoT Build 0:6ae2f7bca550 654 {
AzureIoTClient 6:641a9672db08 655 int result;
Azure.IoT Build 0:6ae2f7bca550 656
AzureIoTClient 6:641a9672db08 657 if ((message_sender == NULL) ||
AzureIoTClient 6:641a9672db08 658 (message == NULL))
AzureIoTClient 6:641a9672db08 659 {
AzureIoTClient 6:641a9672db08 660 result = __LINE__;
AzureIoTClient 6:641a9672db08 661 }
AzureIoTClient 6:641a9672db08 662 else
AzureIoTClient 6:641a9672db08 663 {
AzureIoTClient 6:641a9672db08 664 MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_sender;
AzureIoTClient 6:641a9672db08 665 if (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_ERROR)
AzureIoTClient 6:641a9672db08 666 {
AzureIoTClient 6:641a9672db08 667 result = __LINE__;
AzureIoTClient 6:641a9672db08 668 }
AzureIoTClient 6:641a9672db08 669 else
AzureIoTClient 6:641a9672db08 670 {
AzureIoTClient 6:641a9672db08 671 MESSAGE_WITH_CALLBACK* message_with_callback = (MESSAGE_WITH_CALLBACK*)amqpalloc_malloc(sizeof(MESSAGE_WITH_CALLBACK));
AzureIoTClient 6:641a9672db08 672 if (message_with_callback == NULL)
AzureIoTClient 6:641a9672db08 673 {
AzureIoTClient 6:641a9672db08 674 result = __LINE__;
AzureIoTClient 6:641a9672db08 675 }
AzureIoTClient 6:641a9672db08 676 else
AzureIoTClient 6:641a9672db08 677 {
AzureIoTClient 6:641a9672db08 678 MESSAGE_WITH_CALLBACK** new_messages = (MESSAGE_WITH_CALLBACK**)amqpalloc_realloc(message_sender_instance->messages, sizeof(MESSAGE_WITH_CALLBACK*) * (message_sender_instance->message_count + 1));
AzureIoTClient 6:641a9672db08 679 if (new_messages == NULL)
AzureIoTClient 6:641a9672db08 680 {
AzureIoTClient 6:641a9672db08 681 amqpalloc_free(message_with_callback);
AzureIoTClient 6:641a9672db08 682 result = __LINE__;
AzureIoTClient 6:641a9672db08 683 }
AzureIoTClient 6:641a9672db08 684 else
AzureIoTClient 6:641a9672db08 685 {
AzureIoTClient 6:641a9672db08 686 result = 0;
Azure.IoT Build 0:6ae2f7bca550 687
AzureIoTClient 6:641a9672db08 688 message_sender_instance->messages = new_messages;
AzureIoTClient 6:641a9672db08 689 if (message_sender_instance->message_sender_state != MESSAGE_SENDER_STATE_OPEN)
AzureIoTClient 6:641a9672db08 690 {
AzureIoTClient 6:641a9672db08 691 message_with_callback->message = message_clone(message);
AzureIoTClient 6:641a9672db08 692 if (message_with_callback->message == NULL)
AzureIoTClient 6:641a9672db08 693 {
AzureIoTClient 6:641a9672db08 694 amqpalloc_free(message_with_callback);
AzureIoTClient 6:641a9672db08 695 result = __LINE__;
AzureIoTClient 6:641a9672db08 696 }
AzureIoTClient 6:641a9672db08 697
AzureIoTClient 6:641a9672db08 698 message_with_callback->message_send_state = MESSAGE_SEND_STATE_NOT_SENT;
AzureIoTClient 6:641a9672db08 699 }
AzureIoTClient 6:641a9672db08 700 else
AzureIoTClient 6:641a9672db08 701 {
AzureIoTClient 6:641a9672db08 702 message_with_callback->message = NULL;
AzureIoTClient 6:641a9672db08 703 message_with_callback->message_send_state = MESSAGE_SEND_STATE_PENDING;
AzureIoTClient 6:641a9672db08 704 }
Azure.IoT Build 0:6ae2f7bca550 705
AzureIoTClient 6:641a9672db08 706 if (result == 0)
AzureIoTClient 6:641a9672db08 707 {
AzureIoTClient 6:641a9672db08 708 message_with_callback->on_message_send_complete = on_message_send_complete;
AzureIoTClient 6:641a9672db08 709 message_with_callback->context = callback_context;
AzureIoTClient 6:641a9672db08 710 message_with_callback->message_sender = message_sender_instance;
AzureIoTClient 6:641a9672db08 711
AzureIoTClient 6:641a9672db08 712 message_sender_instance->messages[message_sender_instance->message_count] = message_with_callback;
AzureIoTClient 6:641a9672db08 713 message_sender_instance->message_count++;
Azure.IoT Build 0:6ae2f7bca550 714
AzureIoTClient 6:641a9672db08 715 if (message_sender_instance->message_sender_state == MESSAGE_SENDER_STATE_OPEN)
AzureIoTClient 6:641a9672db08 716 {
AzureIoTClient 6:641a9672db08 717 switch (send_one_message(message_sender_instance, message_with_callback, message))
AzureIoTClient 6:641a9672db08 718 {
AzureIoTClient 6:641a9672db08 719 default:
AzureIoTClient 6:641a9672db08 720 case SEND_ONE_MESSAGE_ERROR:
AzureIoTClient 6:641a9672db08 721 remove_pending_message_by_index(message_sender_instance, message_sender_instance->message_count - 1);
AzureIoTClient 6:641a9672db08 722 result = __LINE__;
AzureIoTClient 6:641a9672db08 723 break;
Azure.IoT Build 0:6ae2f7bca550 724
AzureIoTClient 6:641a9672db08 725 case SEND_ONE_MESSAGE_BUSY:
AzureIoTClient 6:641a9672db08 726 message_with_callback->message = message_clone(message);
AzureIoTClient 6:641a9672db08 727 if (message_with_callback->message == NULL)
AzureIoTClient 6:641a9672db08 728 {
AzureIoTClient 6:641a9672db08 729 amqpalloc_free(message_with_callback);
AzureIoTClient 6:641a9672db08 730 result = __LINE__;
AzureIoTClient 6:641a9672db08 731 }
AzureIoTClient 6:641a9672db08 732 else
AzureIoTClient 6:641a9672db08 733 {
AzureIoTClient 6:641a9672db08 734 message_with_callback->message_send_state = MESSAGE_SEND_STATE_NOT_SENT;
AzureIoTClient 6:641a9672db08 735 result = 0;
AzureIoTClient 6:641a9672db08 736 }
AzureIoTClient 6:641a9672db08 737 break;
Azure.IoT Build 0:6ae2f7bca550 738
AzureIoTClient 6:641a9672db08 739 case SEND_ONE_MESSAGE_OK:
AzureIoTClient 6:641a9672db08 740 result = 0;
AzureIoTClient 6:641a9672db08 741 break;
AzureIoTClient 6:641a9672db08 742 }
AzureIoTClient 6:641a9672db08 743 }
AzureIoTClient 6:641a9672db08 744 }
AzureIoTClient 6:641a9672db08 745 }
AzureIoTClient 6:641a9672db08 746 }
AzureIoTClient 6:641a9672db08 747 }
AzureIoTClient 6:641a9672db08 748 }
AzureIoTClient 6:641a9672db08 749 return result;
AzureIoTClient 6:641a9672db08 750 }
Azure.IoT Build 0:6ae2f7bca550 751
AzureIoTClient 6:641a9672db08 752 void messagesender_set_trace(MESSAGE_SENDER_HANDLE message_sender, bool traceOn)
AzureIoTClient 6:641a9672db08 753 {
AzureIoTClient 6:641a9672db08 754 MESSAGE_SENDER_INSTANCE* message_sender_instance = (MESSAGE_SENDER_INSTANCE*)message_sender;
AzureIoTClient 6:641a9672db08 755 if (message_sender_instance != NULL)
AzureIoTClient 6:641a9672db08 756 {
AzureIoTClient 6:641a9672db08 757 message_sender_instance->is_trace_on = traceOn ? 1 : 0;
AzureIoTClient 6:641a9672db08 758 }
Azure.IoT Build 0:6ae2f7bca550 759 }