Mark Radbourne / Mbed 2 deprecated iothub_client_sample_amqp

Demo using MBED TLS

Dependencies:   EthernetInterface NTPClient iothub_amqp_transport iothub_client mbed-rtos mbed

Fork of iothub_client_sample_amqp by Azure IoT

azure_uamqp_c/session.c@58:f50b97b08851, 2017-01-05 (annotated)

Committer:
markrad
Date:
Thu Jan 05 00:20:03 2017 +0000
Revision:
58:f50b97b08851
Sample using MBED TLS

Who changed what in which revision?

UserRevisionLine numberNew contents of line
markrad 58:f50b97b08851 1 // Copyright (c) Microsoft. All rights reserved.
markrad 58:f50b97b08851 2 // Licensed under the MIT license. See LICENSE file in the project root for full license information.
markrad 58:f50b97b08851 3
markrad 58:f50b97b08851 4 #include <stdlib.h>
markrad 58:f50b97b08851 5 #ifdef _CRTDBG_MAP_ALLOC
markrad 58:f50b97b08851 6 #include <crtdbg.h>
markrad 58:f50b97b08851 7 #endif
markrad 58:f50b97b08851 8 #include <string.h>
markrad 58:f50b97b08851 9 #include "azure_uamqp_c/session.h"
markrad 58:f50b97b08851 10 #include "azure_uamqp_c/connection.h"
markrad 58:f50b97b08851 11 #include "azure_uamqp_c/amqpalloc.h"
markrad 58:f50b97b08851 12 #include "azure_c_shared_utility/xlogging.h"
markrad 58:f50b97b08851 13
markrad 58:f50b97b08851 14 typedef struct LINK_ENDPOINT_INSTANCE_TAG
markrad 58:f50b97b08851 15 {
markrad 58:f50b97b08851 16 char* name;
markrad 58:f50b97b08851 17 handle input_handle;
markrad 58:f50b97b08851 18 handle output_handle;
markrad 58:f50b97b08851 19 ON_ENDPOINT_FRAME_RECEIVED frame_received_callback;
markrad 58:f50b97b08851 20 ON_SESSION_STATE_CHANGED on_session_state_changed;
markrad 58:f50b97b08851 21 ON_SESSION_FLOW_ON on_session_flow_on;
markrad 58:f50b97b08851 22 void* callback_context;
markrad 58:f50b97b08851 23 SESSION_HANDLE session;
markrad 58:f50b97b08851 24 } LINK_ENDPOINT_INSTANCE;
markrad 58:f50b97b08851 25
markrad 58:f50b97b08851 26 typedef struct SESSION_INSTANCE_TAG
markrad 58:f50b97b08851 27 {
markrad 58:f50b97b08851 28 ON_ENDPOINT_FRAME_RECEIVED frame_received_callback;
markrad 58:f50b97b08851 29 void* frame_received_callback_context;
markrad 58:f50b97b08851 30 SESSION_STATE session_state;
markrad 58:f50b97b08851 31 SESSION_STATE previous_session_state;
markrad 58:f50b97b08851 32 CONNECTION_HANDLE connection;
markrad 58:f50b97b08851 33 ENDPOINT_HANDLE endpoint;
markrad 58:f50b97b08851 34 LINK_ENDPOINT_INSTANCE** link_endpoints;
markrad 58:f50b97b08851 35 uint32_t link_endpoint_count;
markrad 58:f50b97b08851 36
markrad 58:f50b97b08851 37 ON_LINK_ATTACHED on_link_attached;
markrad 58:f50b97b08851 38 void* on_link_attached_callback_context;
markrad 58:f50b97b08851 39
markrad 58:f50b97b08851 40 /* Codes_SRS_SESSION_01_016: [next-outgoing-id The next-outgoing-id is the transfer-id to assign to the next transfer frame.] */
markrad 58:f50b97b08851 41 transfer_number next_outgoing_id;
markrad 58:f50b97b08851 42 transfer_number next_incoming_id;
markrad 58:f50b97b08851 43 uint32_t desired_incoming_window;
markrad 58:f50b97b08851 44 uint32_t incoming_window;
markrad 58:f50b97b08851 45 uint32_t outgoing_window;
markrad 58:f50b97b08851 46 handle handle_max;
markrad 58:f50b97b08851 47 uint32_t remote_incoming_window;
markrad 58:f50b97b08851 48 uint32_t remote_outgoing_window;
markrad 58:f50b97b08851 49 int is_underlying_connection_open : 1;
markrad 58:f50b97b08851 50 } SESSION_INSTANCE;
markrad 58:f50b97b08851 51
markrad 58:f50b97b08851 52 #define UNDERLYING_CONNECTION_NOT_OPEN 0
markrad 58:f50b97b08851 53 #define UNDERLYING_CONNECTION_OPEN -1
markrad 58:f50b97b08851 54
markrad 58:f50b97b08851 55 static void session_set_state(SESSION_INSTANCE* session_instance, SESSION_STATE session_state)
markrad 58:f50b97b08851 56 {
markrad 58:f50b97b08851 57 uint64_t i;
markrad 58:f50b97b08851 58
markrad 58:f50b97b08851 59 session_instance->previous_session_state = session_instance->session_state;
markrad 58:f50b97b08851 60 session_instance->session_state = session_state;
markrad 58:f50b97b08851 61
markrad 58:f50b97b08851 62 for (i = 0; i < session_instance->link_endpoint_count; i++)
markrad 58:f50b97b08851 63 {
markrad 58:f50b97b08851 64 if (session_instance->link_endpoints[i]->on_session_state_changed != NULL)
markrad 58:f50b97b08851 65 {
markrad 58:f50b97b08851 66 session_instance->link_endpoints[i]->on_session_state_changed(session_instance->link_endpoints[i]->callback_context, session_state, session_instance->previous_session_state);
markrad 58:f50b97b08851 67 }
markrad 58:f50b97b08851 68 }
markrad 58:f50b97b08851 69 }
markrad 58:f50b97b08851 70
markrad 58:f50b97b08851 71 static int send_end_frame(SESSION_INSTANCE* session_instance, ERROR_HANDLE error_handle)
markrad 58:f50b97b08851 72 {
markrad 58:f50b97b08851 73 int result;
markrad 58:f50b97b08851 74 END_HANDLE end_performative;
markrad 58:f50b97b08851 75
markrad 58:f50b97b08851 76 end_performative = end_create();
markrad 58:f50b97b08851 77 if (end_performative == NULL)
markrad 58:f50b97b08851 78 {
markrad 58:f50b97b08851 79 result = __LINE__;
markrad 58:f50b97b08851 80 }
markrad 58:f50b97b08851 81 else
markrad 58:f50b97b08851 82 {
markrad 58:f50b97b08851 83 if ((error_handle != NULL) &&
markrad 58:f50b97b08851 84 (end_set_error(end_performative, error_handle) != 0))
markrad 58:f50b97b08851 85 {
markrad 58:f50b97b08851 86 result = __LINE__;
markrad 58:f50b97b08851 87 }
markrad 58:f50b97b08851 88 else
markrad 58:f50b97b08851 89 {
markrad 58:f50b97b08851 90 AMQP_VALUE end_performative_value = amqpvalue_create_end(end_performative);
markrad 58:f50b97b08851 91 if (end_performative_value == NULL)
markrad 58:f50b97b08851 92 {
markrad 58:f50b97b08851 93 result = __LINE__;
markrad 58:f50b97b08851 94 }
markrad 58:f50b97b08851 95 else
markrad 58:f50b97b08851 96 {
markrad 58:f50b97b08851 97 if (connection_encode_frame(session_instance->endpoint, end_performative_value, NULL, 0, NULL, NULL) != 0)
markrad 58:f50b97b08851 98 {
markrad 58:f50b97b08851 99 result = __LINE__;
markrad 58:f50b97b08851 100 }
markrad 58:f50b97b08851 101 else
markrad 58:f50b97b08851 102 {
markrad 58:f50b97b08851 103 result = 0;
markrad 58:f50b97b08851 104 }
markrad 58:f50b97b08851 105
markrad 58:f50b97b08851 106 amqpvalue_destroy(end_performative_value);
markrad 58:f50b97b08851 107 }
markrad 58:f50b97b08851 108 }
markrad 58:f50b97b08851 109
markrad 58:f50b97b08851 110 end_destroy(end_performative);
markrad 58:f50b97b08851 111 }
markrad 58:f50b97b08851 112
markrad 58:f50b97b08851 113 return result;
markrad 58:f50b97b08851 114 }
markrad 58:f50b97b08851 115
markrad 58:f50b97b08851 116 static void end_session_with_error(SESSION_INSTANCE* session_instance, const char* condition_value, const char* description)
markrad 58:f50b97b08851 117 {
markrad 58:f50b97b08851 118 ERROR_HANDLE error_handle = error_create(condition_value);
markrad 58:f50b97b08851 119 if (error_handle == NULL)
markrad 58:f50b97b08851 120 {
markrad 58:f50b97b08851 121 /* fatal error */
markrad 58:f50b97b08851 122 session_set_state(session_instance, SESSION_STATE_DISCARDING);
markrad 58:f50b97b08851 123 (void)connection_close(session_instance->connection, "amqp:internal-error", "Cannot allocate error handle to end session");
markrad 58:f50b97b08851 124 }
markrad 58:f50b97b08851 125 else
markrad 58:f50b97b08851 126 {
markrad 58:f50b97b08851 127 if ((error_set_description(error_handle, description) != 0) ||
markrad 58:f50b97b08851 128 (send_end_frame(session_instance, error_handle) != 0))
markrad 58:f50b97b08851 129 {
markrad 58:f50b97b08851 130 /* fatal error */
markrad 58:f50b97b08851 131 session_set_state(session_instance, SESSION_STATE_DISCARDING);
markrad 58:f50b97b08851 132 (void)connection_close(session_instance->connection, "amqp:internal-error", "Cannot allocate error handle to end session");
markrad 58:f50b97b08851 133 }
markrad 58:f50b97b08851 134 else
markrad 58:f50b97b08851 135 {
markrad 58:f50b97b08851 136 session_set_state(session_instance, SESSION_STATE_DISCARDING);
markrad 58:f50b97b08851 137 }
markrad 58:f50b97b08851 138
markrad 58:f50b97b08851 139 error_destroy(error_handle);
markrad 58:f50b97b08851 140 }
markrad 58:f50b97b08851 141 }
markrad 58:f50b97b08851 142
markrad 58:f50b97b08851 143 static int send_begin(SESSION_INSTANCE* session_instance)
markrad 58:f50b97b08851 144 {
markrad 58:f50b97b08851 145 int result;
markrad 58:f50b97b08851 146 BEGIN_HANDLE begin = begin_create(session_instance->next_outgoing_id, session_instance->incoming_window, session_instance->outgoing_window);
markrad 58:f50b97b08851 147
markrad 58:f50b97b08851 148 if (begin == NULL)
markrad 58:f50b97b08851 149 {
markrad 58:f50b97b08851 150 result = __LINE__;
markrad 58:f50b97b08851 151 }
markrad 58:f50b97b08851 152 else
markrad 58:f50b97b08851 153 {
markrad 58:f50b97b08851 154 uint16_t remote_channel;
markrad 58:f50b97b08851 155 if (begin_set_handle_max(begin, session_instance->handle_max) != 0)
markrad 58:f50b97b08851 156 {
markrad 58:f50b97b08851 157 result = __LINE__;
markrad 58:f50b97b08851 158 }
markrad 58:f50b97b08851 159 else if ((session_instance->session_state == SESSION_STATE_BEGIN_RCVD) &&
markrad 58:f50b97b08851 160 ((connection_endpoint_get_incoming_channel(session_instance->endpoint, &remote_channel) != 0) ||
markrad 58:f50b97b08851 161 (begin_set_remote_channel(begin, remote_channel) != 0)))
markrad 58:f50b97b08851 162 {
markrad 58:f50b97b08851 163 result = __LINE__;
markrad 58:f50b97b08851 164 }
markrad 58:f50b97b08851 165 else
markrad 58:f50b97b08851 166 {
markrad 58:f50b97b08851 167 AMQP_VALUE begin_performative_value = amqpvalue_create_begin(begin);
markrad 58:f50b97b08851 168 if (begin_performative_value == NULL)
markrad 58:f50b97b08851 169 {
markrad 58:f50b97b08851 170 result = __LINE__;
markrad 58:f50b97b08851 171 }
markrad 58:f50b97b08851 172 else
markrad 58:f50b97b08851 173 {
markrad 58:f50b97b08851 174 if (connection_encode_frame(session_instance->endpoint, begin_performative_value, NULL, 0, NULL, NULL) != 0)
markrad 58:f50b97b08851 175 {
markrad 58:f50b97b08851 176 result = __LINE__;
markrad 58:f50b97b08851 177 }
markrad 58:f50b97b08851 178 else
markrad 58:f50b97b08851 179 {
markrad 58:f50b97b08851 180 result = 0;
markrad 58:f50b97b08851 181 }
markrad 58:f50b97b08851 182
markrad 58:f50b97b08851 183 amqpvalue_destroy(begin_performative_value);
markrad 58:f50b97b08851 184 }
markrad 58:f50b97b08851 185 }
markrad 58:f50b97b08851 186
markrad 58:f50b97b08851 187 begin_destroy(begin);
markrad 58:f50b97b08851 188 }
markrad 58:f50b97b08851 189
markrad 58:f50b97b08851 190 return result;
markrad 58:f50b97b08851 191 }
markrad 58:f50b97b08851 192
markrad 58:f50b97b08851 193 static int send_flow(SESSION_INSTANCE* session)
markrad 58:f50b97b08851 194 {
markrad 58:f50b97b08851 195 int result;
markrad 58:f50b97b08851 196 if (session == NULL)
markrad 58:f50b97b08851 197 {
markrad 58:f50b97b08851 198 result = __LINE__;
markrad 58:f50b97b08851 199 }
markrad 58:f50b97b08851 200 else
markrad 58:f50b97b08851 201 {
markrad 58:f50b97b08851 202 FLOW_HANDLE flow = flow_create(session->incoming_window, session->next_outgoing_id, session->outgoing_window);
markrad 58:f50b97b08851 203
markrad 58:f50b97b08851 204 if (flow == NULL)
markrad 58:f50b97b08851 205 {
markrad 58:f50b97b08851 206 result = __LINE__;
markrad 58:f50b97b08851 207 }
markrad 58:f50b97b08851 208 else
markrad 58:f50b97b08851 209 {
markrad 58:f50b97b08851 210 if (flow_set_next_incoming_id(flow, session->next_incoming_id) != 0)
markrad 58:f50b97b08851 211 {
markrad 58:f50b97b08851 212 result = __LINE__;
markrad 58:f50b97b08851 213 }
markrad 58:f50b97b08851 214 else
markrad 58:f50b97b08851 215 {
markrad 58:f50b97b08851 216 AMQP_VALUE flow_performative_value = amqpvalue_create_flow(flow);
markrad 58:f50b97b08851 217 if (flow_performative_value == NULL)
markrad 58:f50b97b08851 218 {
markrad 58:f50b97b08851 219 result = __LINE__;
markrad 58:f50b97b08851 220 }
markrad 58:f50b97b08851 221 else
markrad 58:f50b97b08851 222 {
markrad 58:f50b97b08851 223 if (connection_encode_frame(session->endpoint, flow_performative_value, NULL, 0, NULL, NULL) != 0)
markrad 58:f50b97b08851 224 {
markrad 58:f50b97b08851 225 result = __LINE__;
markrad 58:f50b97b08851 226 }
markrad 58:f50b97b08851 227 else
markrad 58:f50b97b08851 228 {
markrad 58:f50b97b08851 229 result = 0;
markrad 58:f50b97b08851 230 }
markrad 58:f50b97b08851 231
markrad 58:f50b97b08851 232 amqpvalue_destroy(flow_performative_value);
markrad 58:f50b97b08851 233 }
markrad 58:f50b97b08851 234 }
markrad 58:f50b97b08851 235
markrad 58:f50b97b08851 236 flow_destroy(flow);
markrad 58:f50b97b08851 237 }
markrad 58:f50b97b08851 238 }
markrad 58:f50b97b08851 239
markrad 58:f50b97b08851 240 return result;
markrad 58:f50b97b08851 241 }
markrad 58:f50b97b08851 242
markrad 58:f50b97b08851 243 static LINK_ENDPOINT_INSTANCE* find_link_endpoint_by_name(SESSION_INSTANCE* session, const char* name)
markrad 58:f50b97b08851 244 {
markrad 58:f50b97b08851 245 uint32_t i;
markrad 58:f50b97b08851 246 LINK_ENDPOINT_INSTANCE* result;
markrad 58:f50b97b08851 247
markrad 58:f50b97b08851 248 for (i = 0; i < session->link_endpoint_count; i++)
markrad 58:f50b97b08851 249 {
markrad 58:f50b97b08851 250 if (strcmp(session->link_endpoints[i]->name, name) == 0)
markrad 58:f50b97b08851 251 {
markrad 58:f50b97b08851 252 break;
markrad 58:f50b97b08851 253 }
markrad 58:f50b97b08851 254 }
markrad 58:f50b97b08851 255
markrad 58:f50b97b08851 256 if (i == session->link_endpoint_count)
markrad 58:f50b97b08851 257 {
markrad 58:f50b97b08851 258 result = NULL;
markrad 58:f50b97b08851 259 }
markrad 58:f50b97b08851 260 else
markrad 58:f50b97b08851 261 {
markrad 58:f50b97b08851 262 result = session->link_endpoints[i];
markrad 58:f50b97b08851 263 }
markrad 58:f50b97b08851 264
markrad 58:f50b97b08851 265 return result;
markrad 58:f50b97b08851 266 }
markrad 58:f50b97b08851 267
markrad 58:f50b97b08851 268 static LINK_ENDPOINT_INSTANCE* find_link_endpoint_by_input_handle(SESSION_INSTANCE* session, handle input_handle)
markrad 58:f50b97b08851 269 {
markrad 58:f50b97b08851 270 uint32_t i;
markrad 58:f50b97b08851 271 LINK_ENDPOINT_INSTANCE* result;
markrad 58:f50b97b08851 272
markrad 58:f50b97b08851 273 for (i = 0; i < session->link_endpoint_count; i++)
markrad 58:f50b97b08851 274 {
markrad 58:f50b97b08851 275 if (session->link_endpoints[i]->input_handle == input_handle)
markrad 58:f50b97b08851 276 {
markrad 58:f50b97b08851 277 break;
markrad 58:f50b97b08851 278 }
markrad 58:f50b97b08851 279 }
markrad 58:f50b97b08851 280
markrad 58:f50b97b08851 281 if (i == session->link_endpoint_count)
markrad 58:f50b97b08851 282 {
markrad 58:f50b97b08851 283 result = NULL;
markrad 58:f50b97b08851 284 }
markrad 58:f50b97b08851 285 else
markrad 58:f50b97b08851 286 {
markrad 58:f50b97b08851 287 result = session->link_endpoints[i];
markrad 58:f50b97b08851 288 }
markrad 58:f50b97b08851 289
markrad 58:f50b97b08851 290 return result;
markrad 58:f50b97b08851 291 }
markrad 58:f50b97b08851 292
markrad 58:f50b97b08851 293 static LINK_ENDPOINT_INSTANCE* find_link_endpoint_by_output_handle(SESSION_INSTANCE* session, handle output_handle)
markrad 58:f50b97b08851 294 {
markrad 58:f50b97b08851 295 uint32_t i;
markrad 58:f50b97b08851 296 LINK_ENDPOINT_INSTANCE* result;
markrad 58:f50b97b08851 297
markrad 58:f50b97b08851 298 for (i = 0; i < session->link_endpoint_count; i++)
markrad 58:f50b97b08851 299 {
markrad 58:f50b97b08851 300 if (session->link_endpoints[i]->output_handle == output_handle)
markrad 58:f50b97b08851 301 {
markrad 58:f50b97b08851 302 break;
markrad 58:f50b97b08851 303 }
markrad 58:f50b97b08851 304 }
markrad 58:f50b97b08851 305
markrad 58:f50b97b08851 306 if (i == session->link_endpoint_count)
markrad 58:f50b97b08851 307 {
markrad 58:f50b97b08851 308 result = NULL;
markrad 58:f50b97b08851 309 }
markrad 58:f50b97b08851 310 else
markrad 58:f50b97b08851 311 {
markrad 58:f50b97b08851 312 result = session->link_endpoints[i];
markrad 58:f50b97b08851 313 }
markrad 58:f50b97b08851 314
markrad 58:f50b97b08851 315 return result;
markrad 58:f50b97b08851 316 }
markrad 58:f50b97b08851 317
markrad 58:f50b97b08851 318 static void on_connection_state_changed(void* context, CONNECTION_STATE new_connection_state, CONNECTION_STATE previous_connection_state)
markrad 58:f50b97b08851 319 {
markrad 58:f50b97b08851 320 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)context;
markrad 58:f50b97b08851 321
markrad 58:f50b97b08851 322 /* Codes_SRS_SESSION_01_060: [If the previous connection state is not OPENED and the new connection state is OPENED, the BEGIN frame shall be sent out and the state shall be switched to BEGIN_SENT.] */
markrad 58:f50b97b08851 323 if ((new_connection_state == CONNECTION_STATE_OPENED) && (previous_connection_state != CONNECTION_STATE_OPENED) && (session_instance->session_state == SESSION_STATE_UNMAPPED))
markrad 58:f50b97b08851 324 {
markrad 58:f50b97b08851 325 if (send_begin(session_instance) == 0)
markrad 58:f50b97b08851 326 {
markrad 58:f50b97b08851 327 session_set_state(session_instance, SESSION_STATE_BEGIN_SENT);
markrad 58:f50b97b08851 328 }
markrad 58:f50b97b08851 329 }
markrad 58:f50b97b08851 330 /* Codes_SRS_SESSION_01_061: [If the previous connection state is OPENED and the new connection state is not OPENED anymore, the state shall be switched to DISCARDING.] */
markrad 58:f50b97b08851 331 else if ((new_connection_state == CONNECTION_STATE_CLOSE_RCVD) || (new_connection_state == CONNECTION_STATE_END))
markrad 58:f50b97b08851 332 {
markrad 58:f50b97b08851 333 session_set_state(session_instance, SESSION_STATE_DISCARDING);
markrad 58:f50b97b08851 334 }
markrad 58:f50b97b08851 335 /* Codes_SRS_SESSION_09_001: [If the new connection state is ERROR, the state shall be switched to ERROR.] */
markrad 58:f50b97b08851 336 else if (new_connection_state == CONNECTION_STATE_ERROR)
markrad 58:f50b97b08851 337 {
markrad 58:f50b97b08851 338 session_set_state(session_instance, SESSION_STATE_ERROR);
markrad 58:f50b97b08851 339 }
markrad 58:f50b97b08851 340 }
markrad 58:f50b97b08851 341
markrad 58:f50b97b08851 342 static void on_frame_received(void* context, AMQP_VALUE performative, uint32_t payload_size, const unsigned char* payload_bytes)
markrad 58:f50b97b08851 343 {
markrad 58:f50b97b08851 344 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)context;
markrad 58:f50b97b08851 345 AMQP_VALUE descriptor = amqpvalue_get_inplace_descriptor(performative);
markrad 58:f50b97b08851 346
markrad 58:f50b97b08851 347 if (is_begin_type_by_descriptor(descriptor))
markrad 58:f50b97b08851 348 {
markrad 58:f50b97b08851 349 BEGIN_HANDLE begin_handle;
markrad 58:f50b97b08851 350
markrad 58:f50b97b08851 351 if (amqpvalue_get_begin(performative, &begin_handle) != 0)
markrad 58:f50b97b08851 352 {
markrad 58:f50b97b08851 353 connection_close(session_instance->connection, "amqp:decode-error", "Cannot decode BEGIN frame");
markrad 58:f50b97b08851 354 }
markrad 58:f50b97b08851 355 else
markrad 58:f50b97b08851 356 {
markrad 58:f50b97b08851 357 if ((begin_get_incoming_window(begin_handle, &session_instance->remote_incoming_window) != 0) ||
markrad 58:f50b97b08851 358 (begin_get_next_outgoing_id(begin_handle, &session_instance->next_incoming_id) != 0))
markrad 58:f50b97b08851 359 {
markrad 58:f50b97b08851 360 /* error */
markrad 58:f50b97b08851 361 begin_destroy(begin_handle);
markrad 58:f50b97b08851 362 session_set_state(session_instance, SESSION_STATE_DISCARDING);
markrad 58:f50b97b08851 363 connection_close(session_instance->connection, "amqp:decode-error", "Cannot get incoming windows and next outgoing id");
markrad 58:f50b97b08851 364 }
markrad 58:f50b97b08851 365 else
markrad 58:f50b97b08851 366 {
markrad 58:f50b97b08851 367 begin_destroy(begin_handle);
markrad 58:f50b97b08851 368
markrad 58:f50b97b08851 369 if (session_instance->session_state == SESSION_STATE_BEGIN_SENT)
markrad 58:f50b97b08851 370 {
markrad 58:f50b97b08851 371 session_set_state(session_instance, SESSION_STATE_MAPPED);
markrad 58:f50b97b08851 372 }
markrad 58:f50b97b08851 373 else if(session_instance->session_state == SESSION_STATE_UNMAPPED)
markrad 58:f50b97b08851 374 {
markrad 58:f50b97b08851 375 session_set_state(session_instance, SESSION_STATE_BEGIN_RCVD);
markrad 58:f50b97b08851 376 if (send_begin(session_instance) != 0)
markrad 58:f50b97b08851 377 {
markrad 58:f50b97b08851 378 connection_close(session_instance->connection, "amqp:internal-error", "Failed sending BEGIN frame");
markrad 58:f50b97b08851 379 session_set_state(session_instance, SESSION_STATE_DISCARDING);
markrad 58:f50b97b08851 380 }
markrad 58:f50b97b08851 381 else
markrad 58:f50b97b08851 382 {
markrad 58:f50b97b08851 383 session_set_state(session_instance, SESSION_STATE_MAPPED);
markrad 58:f50b97b08851 384 }
markrad 58:f50b97b08851 385 }
markrad 58:f50b97b08851 386 }
markrad 58:f50b97b08851 387 }
markrad 58:f50b97b08851 388 }
markrad 58:f50b97b08851 389 else if (is_attach_type_by_descriptor(descriptor))
markrad 58:f50b97b08851 390 {
markrad 58:f50b97b08851 391 const char* name = NULL;
markrad 58:f50b97b08851 392 ATTACH_HANDLE attach_handle;
markrad 58:f50b97b08851 393
markrad 58:f50b97b08851 394 if (amqpvalue_get_attach(performative, &attach_handle) != 0)
markrad 58:f50b97b08851 395 {
markrad 58:f50b97b08851 396 end_session_with_error(session_instance, "amqp:decode-error", "Cannot decode ATTACH frame");
markrad 58:f50b97b08851 397 }
markrad 58:f50b97b08851 398 else
markrad 58:f50b97b08851 399 {
markrad 58:f50b97b08851 400 role role;
markrad 58:f50b97b08851 401 AMQP_VALUE source;
markrad 58:f50b97b08851 402 AMQP_VALUE target;
markrad 58:f50b97b08851 403
markrad 58:f50b97b08851 404 if ((attach_get_name(attach_handle, &name) != 0) ||
markrad 58:f50b97b08851 405 (attach_get_role(attach_handle, &role) != 0) ||
markrad 58:f50b97b08851 406 (attach_get_source(attach_handle, &source) != 0) ||
markrad 58:f50b97b08851 407 (attach_get_target(attach_handle, &target) != 0))
markrad 58:f50b97b08851 408 {
markrad 58:f50b97b08851 409 end_session_with_error(session_instance, "amqp:decode-error", "Cannot get link name from ATTACH frame");
markrad 58:f50b97b08851 410 }
markrad 58:f50b97b08851 411 else
markrad 58:f50b97b08851 412 {
markrad 58:f50b97b08851 413 LINK_ENDPOINT_INSTANCE* link_endpoint = find_link_endpoint_by_name(session_instance, name);
markrad 58:f50b97b08851 414 if (link_endpoint == NULL)
markrad 58:f50b97b08851 415 {
markrad 58:f50b97b08851 416 /* new link attach */
markrad 58:f50b97b08851 417 if (session_instance->on_link_attached != NULL)
markrad 58:f50b97b08851 418 {
markrad 58:f50b97b08851 419 LINK_ENDPOINT_HANDLE new_link_endpoint = session_create_link_endpoint(session_instance, name);
markrad 58:f50b97b08851 420 if (new_link_endpoint == NULL)
markrad 58:f50b97b08851 421 {
markrad 58:f50b97b08851 422 end_session_with_error(session_instance, "amqp:internal-error", "Cannot create link endpoint");
markrad 58:f50b97b08851 423 }
markrad 58:f50b97b08851 424 else if (attach_get_handle(attach_handle, &new_link_endpoint->input_handle) != 0)
markrad 58:f50b97b08851 425 {
markrad 58:f50b97b08851 426 end_session_with_error(session_instance, "amqp:decode-error", "Cannot get input handle from ATTACH frame");
markrad 58:f50b97b08851 427 }
markrad 58:f50b97b08851 428 else
markrad 58:f50b97b08851 429 {
markrad 58:f50b97b08851 430 if (!session_instance->on_link_attached(session_instance->on_link_attached_callback_context, new_link_endpoint, name, role, source, target))
markrad 58:f50b97b08851 431 {
markrad 58:f50b97b08851 432 session_destroy_link_endpoint(new_link_endpoint);
markrad 58:f50b97b08851 433 new_link_endpoint = NULL;
markrad 58:f50b97b08851 434 }
markrad 58:f50b97b08851 435 else
markrad 58:f50b97b08851 436 {
markrad 58:f50b97b08851 437 if (new_link_endpoint->frame_received_callback != NULL)
markrad 58:f50b97b08851 438 {
markrad 58:f50b97b08851 439 new_link_endpoint->frame_received_callback(new_link_endpoint->callback_context, performative, payload_size, payload_bytes);
markrad 58:f50b97b08851 440 }
markrad 58:f50b97b08851 441 }
markrad 58:f50b97b08851 442 }
markrad 58:f50b97b08851 443 }
markrad 58:f50b97b08851 444 }
markrad 58:f50b97b08851 445 else
markrad 58:f50b97b08851 446 {
markrad 58:f50b97b08851 447 if (attach_get_handle(attach_handle, &link_endpoint->input_handle) != 0)
markrad 58:f50b97b08851 448 {
markrad 58:f50b97b08851 449 end_session_with_error(session_instance, "amqp:decode-error", "Cannot get input handle from ATTACH frame");
markrad 58:f50b97b08851 450 }
markrad 58:f50b97b08851 451 else
markrad 58:f50b97b08851 452 {
markrad 58:f50b97b08851 453 link_endpoint->frame_received_callback(link_endpoint->callback_context, performative, payload_size, payload_bytes);
markrad 58:f50b97b08851 454 }
markrad 58:f50b97b08851 455 }
markrad 58:f50b97b08851 456 }
markrad 58:f50b97b08851 457
markrad 58:f50b97b08851 458 attach_destroy(attach_handle);
markrad 58:f50b97b08851 459 }
markrad 58:f50b97b08851 460 }
markrad 58:f50b97b08851 461 else if (is_detach_type_by_descriptor(descriptor))
markrad 58:f50b97b08851 462 {
markrad 58:f50b97b08851 463 DETACH_HANDLE detach_handle;
markrad 58:f50b97b08851 464
markrad 58:f50b97b08851 465 if (amqpvalue_get_detach(performative, &detach_handle) != 0)
markrad 58:f50b97b08851 466 {
markrad 58:f50b97b08851 467 end_session_with_error(session_instance, "amqp:decode-error", "Cannot decode DETACH frame");
markrad 58:f50b97b08851 468 }
markrad 58:f50b97b08851 469 else
markrad 58:f50b97b08851 470 {
markrad 58:f50b97b08851 471 uint32_t remote_handle;
markrad 58:f50b97b08851 472 if (detach_get_handle(detach_handle, &remote_handle) != 0)
markrad 58:f50b97b08851 473 {
markrad 58:f50b97b08851 474 end_session_with_error(session_instance, "amqp:decode-error", "Cannot get handle from DETACH frame");
markrad 58:f50b97b08851 475
markrad 58:f50b97b08851 476 detach_destroy(detach_handle);
markrad 58:f50b97b08851 477 }
markrad 58:f50b97b08851 478 else
markrad 58:f50b97b08851 479 {
markrad 58:f50b97b08851 480 detach_destroy(detach_handle);
markrad 58:f50b97b08851 481
markrad 58:f50b97b08851 482 LINK_ENDPOINT_INSTANCE* link_endpoint = find_link_endpoint_by_input_handle(session_instance, remote_handle);
markrad 58:f50b97b08851 483 if (link_endpoint == NULL)
markrad 58:f50b97b08851 484 {
markrad 58:f50b97b08851 485 end_session_with_error(session_instance, "amqp:session:unattached-handle", "");
markrad 58:f50b97b08851 486 }
markrad 58:f50b97b08851 487 else
markrad 58:f50b97b08851 488 {
markrad 58:f50b97b08851 489 link_endpoint->frame_received_callback(link_endpoint->callback_context, performative, payload_size, payload_bytes);
markrad 58:f50b97b08851 490 }
markrad 58:f50b97b08851 491 }
markrad 58:f50b97b08851 492 }
markrad 58:f50b97b08851 493 }
markrad 58:f50b97b08851 494 else if (is_flow_type_by_descriptor(descriptor))
markrad 58:f50b97b08851 495 {
markrad 58:f50b97b08851 496 FLOW_HANDLE flow_handle;
markrad 58:f50b97b08851 497
markrad 58:f50b97b08851 498 if (amqpvalue_get_flow(performative, &flow_handle) != 0)
markrad 58:f50b97b08851 499 {
markrad 58:f50b97b08851 500 end_session_with_error(session_instance, "amqp:decode-error", "Cannot decode FLOW frame");
markrad 58:f50b97b08851 501 }
markrad 58:f50b97b08851 502 else
markrad 58:f50b97b08851 503 {
markrad 58:f50b97b08851 504 uint32_t remote_handle;
markrad 58:f50b97b08851 505 transfer_number flow_next_incoming_id;
markrad 58:f50b97b08851 506 uint32_t flow_incoming_window;
markrad 58:f50b97b08851 507
markrad 58:f50b97b08851 508 if (flow_get_next_incoming_id(flow_handle, &flow_next_incoming_id) != 0)
markrad 58:f50b97b08851 509 {
markrad 58:f50b97b08851 510 /*
markrad 58:f50b97b08851 511 If the next-incoming-id field of the flow frame is not set,
markrad 58:f50b97b08851 512 then remote-incomingwindow is computed as follows:
markrad 58:f50b97b08851 513 initial-outgoing-id(endpoint) + incoming-window(flow) - next-outgoing-id(endpoint)
markrad 58:f50b97b08851 514 */
markrad 58:f50b97b08851 515 flow_next_incoming_id = session_instance->next_outgoing_id;
markrad 58:f50b97b08851 516 }
markrad 58:f50b97b08851 517
markrad 58:f50b97b08851 518 if ((flow_get_next_outgoing_id(flow_handle, &session_instance->next_incoming_id) != 0) ||
markrad 58:f50b97b08851 519 (flow_get_incoming_window(flow_handle, &flow_incoming_window) != 0))
markrad 58:f50b97b08851 520 {
markrad 58:f50b97b08851 521 flow_destroy(flow_handle);
markrad 58:f50b97b08851 522
markrad 58:f50b97b08851 523 end_session_with_error(session_instance, "amqp:decode-error", "Cannot decode FLOW frame");
markrad 58:f50b97b08851 524 }
markrad 58:f50b97b08851 525 else
markrad 58:f50b97b08851 526 {
markrad 58:f50b97b08851 527 LINK_ENDPOINT_INSTANCE* link_endpoint_instance = NULL;
markrad 58:f50b97b08851 528
markrad 58:f50b97b08851 529 session_instance->remote_incoming_window = flow_next_incoming_id + flow_incoming_window - session_instance->next_outgoing_id;
markrad 58:f50b97b08851 530
markrad 58:f50b97b08851 531 if (flow_get_handle(flow_handle, &remote_handle) == 0)
markrad 58:f50b97b08851 532 {
markrad 58:f50b97b08851 533 link_endpoint_instance = find_link_endpoint_by_input_handle(session_instance, remote_handle);
markrad 58:f50b97b08851 534 }
markrad 58:f50b97b08851 535
markrad 58:f50b97b08851 536 flow_destroy(flow_handle);
markrad 58:f50b97b08851 537
markrad 58:f50b97b08851 538 if (link_endpoint_instance != NULL)
markrad 58:f50b97b08851 539 {
markrad 58:f50b97b08851 540 link_endpoint_instance->frame_received_callback(link_endpoint_instance->callback_context, performative, payload_size, payload_bytes);
markrad 58:f50b97b08851 541 }
markrad 58:f50b97b08851 542
markrad 58:f50b97b08851 543 size_t i = 0;
markrad 58:f50b97b08851 544 while ((session_instance->remote_incoming_window > 0) && (i < session_instance->link_endpoint_count))
markrad 58:f50b97b08851 545 {
markrad 58:f50b97b08851 546 /* notify the caller that it can send here */
markrad 58:f50b97b08851 547 if (session_instance->link_endpoints[i]->on_session_flow_on != NULL)
markrad 58:f50b97b08851 548 {
markrad 58:f50b97b08851 549 session_instance->link_endpoints[i]->on_session_flow_on(session_instance->link_endpoints[i]->callback_context);
markrad 58:f50b97b08851 550 }
markrad 58:f50b97b08851 551
markrad 58:f50b97b08851 552 i++;
markrad 58:f50b97b08851 553 }
markrad 58:f50b97b08851 554 }
markrad 58:f50b97b08851 555 }
markrad 58:f50b97b08851 556 }
markrad 58:f50b97b08851 557 else if (is_transfer_type_by_descriptor(descriptor))
markrad 58:f50b97b08851 558 {
markrad 58:f50b97b08851 559 TRANSFER_HANDLE transfer_handle;
markrad 58:f50b97b08851 560
markrad 58:f50b97b08851 561 if (amqpvalue_get_transfer(performative, &transfer_handle) != 0)
markrad 58:f50b97b08851 562 {
markrad 58:f50b97b08851 563 end_session_with_error(session_instance, "amqp:decode-error", "Cannot decode TRANSFER frame");
markrad 58:f50b97b08851 564 }
markrad 58:f50b97b08851 565 else
markrad 58:f50b97b08851 566 {
markrad 58:f50b97b08851 567 uint32_t remote_handle;
markrad 58:f50b97b08851 568 delivery_number delivery_id;
markrad 58:f50b97b08851 569
markrad 58:f50b97b08851 570 transfer_get_delivery_id(transfer_handle, &delivery_id);
markrad 58:f50b97b08851 571 if (transfer_get_handle(transfer_handle, &remote_handle) != 0)
markrad 58:f50b97b08851 572 {
markrad 58:f50b97b08851 573 transfer_destroy(transfer_handle);
markrad 58:f50b97b08851 574 end_session_with_error(session_instance, "amqp:decode-error", "Cannot get handle from TRANSFER frame");
markrad 58:f50b97b08851 575 }
markrad 58:f50b97b08851 576 else
markrad 58:f50b97b08851 577 {
markrad 58:f50b97b08851 578 transfer_destroy(transfer_handle);
markrad 58:f50b97b08851 579
markrad 58:f50b97b08851 580 session_instance->next_incoming_id++;
markrad 58:f50b97b08851 581 session_instance->remote_outgoing_window--;
markrad 58:f50b97b08851 582 session_instance->incoming_window--;
markrad 58:f50b97b08851 583
markrad 58:f50b97b08851 584 LINK_ENDPOINT_INSTANCE* link_endpoint = find_link_endpoint_by_input_handle(session_instance, remote_handle);
markrad 58:f50b97b08851 585 if (link_endpoint == NULL)
markrad 58:f50b97b08851 586 {
markrad 58:f50b97b08851 587 end_session_with_error(session_instance, "amqp:session:unattached-handle", "");
markrad 58:f50b97b08851 588 }
markrad 58:f50b97b08851 589 else
markrad 58:f50b97b08851 590 {
markrad 58:f50b97b08851 591 link_endpoint->frame_received_callback(link_endpoint->callback_context, performative, payload_size, payload_bytes);
markrad 58:f50b97b08851 592 }
markrad 58:f50b97b08851 593
markrad 58:f50b97b08851 594 if (session_instance->incoming_window == 0)
markrad 58:f50b97b08851 595 {
markrad 58:f50b97b08851 596 session_instance->incoming_window = session_instance->desired_incoming_window;
markrad 58:f50b97b08851 597 send_flow(session_instance);
markrad 58:f50b97b08851 598 }
markrad 58:f50b97b08851 599 }
markrad 58:f50b97b08851 600 }
markrad 58:f50b97b08851 601 }
markrad 58:f50b97b08851 602 else if (is_disposition_type_by_descriptor(descriptor))
markrad 58:f50b97b08851 603 {
markrad 58:f50b97b08851 604 uint32_t i;
markrad 58:f50b97b08851 605
markrad 58:f50b97b08851 606 for (i = 0; i < session_instance->link_endpoint_count; i++)
markrad 58:f50b97b08851 607 {
markrad 58:f50b97b08851 608 LINK_ENDPOINT_INSTANCE* link_endpoint = session_instance->link_endpoints[i];
markrad 58:f50b97b08851 609 link_endpoint->frame_received_callback(link_endpoint->callback_context, performative, payload_size, payload_bytes);
markrad 58:f50b97b08851 610 }
markrad 58:f50b97b08851 611 }
markrad 58:f50b97b08851 612 else if (is_end_type_by_descriptor(descriptor))
markrad 58:f50b97b08851 613 {
markrad 58:f50b97b08851 614 END_HANDLE end_handle;
markrad 58:f50b97b08851 615
markrad 58:f50b97b08851 616 if (amqpvalue_get_end(performative, &end_handle) != 0)
markrad 58:f50b97b08851 617 {
markrad 58:f50b97b08851 618 end_session_with_error(session_instance, "amqp:decode-error", "Cannot decode END frame");
markrad 58:f50b97b08851 619 }
markrad 58:f50b97b08851 620 else
markrad 58:f50b97b08851 621 {
markrad 58:f50b97b08851 622 if ((session_instance->session_state != SESSION_STATE_END_RCVD) &&
markrad 58:f50b97b08851 623 (session_instance->session_state != SESSION_STATE_DISCARDING))
markrad 58:f50b97b08851 624 {
markrad 58:f50b97b08851 625 session_set_state(session_instance, SESSION_STATE_END_RCVD);
markrad 58:f50b97b08851 626 if (send_end_frame(session_instance, NULL) != 0)
markrad 58:f50b97b08851 627 {
markrad 58:f50b97b08851 628 /* fatal error */
markrad 58:f50b97b08851 629 (void)connection_close(session_instance->connection, "amqp:internal-error", "Cannot send END frame.");
markrad 58:f50b97b08851 630 }
markrad 58:f50b97b08851 631
markrad 58:f50b97b08851 632 session_set_state(session_instance, SESSION_STATE_DISCARDING);
markrad 58:f50b97b08851 633 }
markrad 58:f50b97b08851 634 }
markrad 58:f50b97b08851 635 }
markrad 58:f50b97b08851 636 }
markrad 58:f50b97b08851 637
markrad 58:f50b97b08851 638 SESSION_HANDLE session_create(CONNECTION_HANDLE connection, ON_LINK_ATTACHED on_link_attached, void* callback_context)
markrad 58:f50b97b08851 639 {
markrad 58:f50b97b08851 640 SESSION_INSTANCE* result;
markrad 58:f50b97b08851 641
markrad 58:f50b97b08851 642 if (connection == NULL)
markrad 58:f50b97b08851 643 {
markrad 58:f50b97b08851 644 /* Codes_SRS_SESSION_01_031: [If connection is NULL, session_create shall fail and return NULL.] */
markrad 58:f50b97b08851 645 result = NULL;
markrad 58:f50b97b08851 646 }
markrad 58:f50b97b08851 647 else
markrad 58:f50b97b08851 648 {
markrad 58:f50b97b08851 649 /* Codes_SRS_SESSION_01_030: [session_create shall create a new session instance and return a non-NULL handle to it.] */
markrad 58:f50b97b08851 650 result = amqpalloc_malloc(sizeof(SESSION_INSTANCE));
markrad 58:f50b97b08851 651 /* Codes_SRS_SESSION_01_042: [If allocating memory for the session fails, session_create shall fail and return NULL.] */
markrad 58:f50b97b08851 652 if (result != NULL)
markrad 58:f50b97b08851 653 {
markrad 58:f50b97b08851 654 result->connection = connection;
markrad 58:f50b97b08851 655 result->link_endpoints = NULL;
markrad 58:f50b97b08851 656 result->link_endpoint_count = 0;
markrad 58:f50b97b08851 657 result->handle_max = 4294967295u;
markrad 58:f50b97b08851 658
markrad 58:f50b97b08851 659 /* Codes_SRS_SESSION_01_057: [The delivery ids shall be assigned starting at 0.] */
markrad 58:f50b97b08851 660 /* Codes_SRS_SESSION_01_017: [The nextoutgoing-id MAY be initialized to an arbitrary value ] */
markrad 58:f50b97b08851 661 result->next_outgoing_id = 0;
markrad 58:f50b97b08851 662
markrad 58:f50b97b08851 663 result->desired_incoming_window = 1;
markrad 58:f50b97b08851 664 result->incoming_window = 1;
markrad 58:f50b97b08851 665 result->outgoing_window = 1;
markrad 58:f50b97b08851 666 result->handle_max = 4294967295u;
markrad 58:f50b97b08851 667 result->remote_incoming_window = 0;
markrad 58:f50b97b08851 668 result->remote_outgoing_window = 0;
markrad 58:f50b97b08851 669 result->previous_session_state = SESSION_STATE_UNMAPPED;
markrad 58:f50b97b08851 670 result->is_underlying_connection_open = UNDERLYING_CONNECTION_NOT_OPEN;
markrad 58:f50b97b08851 671 result->session_state = SESSION_STATE_UNMAPPED;
markrad 58:f50b97b08851 672 result->on_link_attached = on_link_attached;
markrad 58:f50b97b08851 673 result->on_link_attached_callback_context = callback_context;
markrad 58:f50b97b08851 674
markrad 58:f50b97b08851 675 /* Codes_SRS_SESSION_01_032: [session_create shall create a new session endpoint by calling connection_create_endpoint.] */
markrad 58:f50b97b08851 676 result->endpoint = connection_create_endpoint(connection);
markrad 58:f50b97b08851 677 if (result->endpoint == NULL)
markrad 58:f50b97b08851 678 {
markrad 58:f50b97b08851 679 /* Codes_SRS_SESSION_01_033: [If connection_create_endpoint fails, session_create shall fail and return NULL.] */
markrad 58:f50b97b08851 680 amqpalloc_free(result);
markrad 58:f50b97b08851 681 result = NULL;
markrad 58:f50b97b08851 682 }
markrad 58:f50b97b08851 683 else
markrad 58:f50b97b08851 684 {
markrad 58:f50b97b08851 685 session_set_state(result, SESSION_STATE_UNMAPPED);
markrad 58:f50b97b08851 686 }
markrad 58:f50b97b08851 687 }
markrad 58:f50b97b08851 688 }
markrad 58:f50b97b08851 689
markrad 58:f50b97b08851 690 return result;
markrad 58:f50b97b08851 691 }
markrad 58:f50b97b08851 692
markrad 58:f50b97b08851 693 SESSION_HANDLE session_create_from_endpoint(CONNECTION_HANDLE connection, ENDPOINT_HANDLE endpoint, ON_LINK_ATTACHED on_link_attached, void* callback_context)
markrad 58:f50b97b08851 694 {
markrad 58:f50b97b08851 695 SESSION_INSTANCE* result;
markrad 58:f50b97b08851 696
markrad 58:f50b97b08851 697 if (endpoint == NULL)
markrad 58:f50b97b08851 698 {
markrad 58:f50b97b08851 699 result = NULL;
markrad 58:f50b97b08851 700 }
markrad 58:f50b97b08851 701 else
markrad 58:f50b97b08851 702 {
markrad 58:f50b97b08851 703 result = amqpalloc_malloc(sizeof(SESSION_INSTANCE));
markrad 58:f50b97b08851 704 if (result != NULL)
markrad 58:f50b97b08851 705 {
markrad 58:f50b97b08851 706 result->connection = connection;
markrad 58:f50b97b08851 707 result->link_endpoints = NULL;
markrad 58:f50b97b08851 708 result->link_endpoint_count = 0;
markrad 58:f50b97b08851 709 result->handle_max = 4294967295u;
markrad 58:f50b97b08851 710
markrad 58:f50b97b08851 711 result->next_outgoing_id = 0;
markrad 58:f50b97b08851 712
markrad 58:f50b97b08851 713 result->incoming_window = 1;
markrad 58:f50b97b08851 714 result->outgoing_window = 1;
markrad 58:f50b97b08851 715 result->handle_max = 4294967295u;
markrad 58:f50b97b08851 716 result->remote_incoming_window = 0;
markrad 58:f50b97b08851 717 result->remote_outgoing_window = 0;
markrad 58:f50b97b08851 718 result->previous_session_state = SESSION_STATE_UNMAPPED;
markrad 58:f50b97b08851 719 result->is_underlying_connection_open = UNDERLYING_CONNECTION_NOT_OPEN;
markrad 58:f50b97b08851 720 result->session_state = SESSION_STATE_UNMAPPED;
markrad 58:f50b97b08851 721 result->on_link_attached = on_link_attached;
markrad 58:f50b97b08851 722 result->on_link_attached_callback_context = callback_context;
markrad 58:f50b97b08851 723
markrad 58:f50b97b08851 724 result->endpoint = endpoint;
markrad 58:f50b97b08851 725 session_set_state(result, SESSION_STATE_UNMAPPED);
markrad 58:f50b97b08851 726 }
markrad 58:f50b97b08851 727 }
markrad 58:f50b97b08851 728
markrad 58:f50b97b08851 729 return result;
markrad 58:f50b97b08851 730 }
markrad 58:f50b97b08851 731
markrad 58:f50b97b08851 732 void session_destroy(SESSION_HANDLE session)
markrad 58:f50b97b08851 733 {
markrad 58:f50b97b08851 734 /* Codes_SRS_SESSION_01_036: [If session is NULL, session_destroy shall do nothing.] */
markrad 58:f50b97b08851 735 if (session != NULL)
markrad 58:f50b97b08851 736 {
markrad 58:f50b97b08851 737 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)session;
markrad 58:f50b97b08851 738
markrad 58:f50b97b08851 739 session_end(session, NULL, NULL);
markrad 58:f50b97b08851 740
markrad 58:f50b97b08851 741 /* Codes_SRS_SESSION_01_034: [session_destroy shall free all resources allocated by session_create.] */
markrad 58:f50b97b08851 742 /* Codes_SRS_SESSION_01_035: [The endpoint created in session_create shall be freed by calling connection_destroy_endpoint.] */
markrad 58:f50b97b08851 743 connection_destroy_endpoint(session_instance->endpoint);
markrad 58:f50b97b08851 744 if (session_instance->link_endpoints != NULL)
markrad 58:f50b97b08851 745 {
markrad 58:f50b97b08851 746 amqpalloc_free(session_instance->link_endpoints);
markrad 58:f50b97b08851 747 }
markrad 58:f50b97b08851 748
markrad 58:f50b97b08851 749 amqpalloc_free(session);
markrad 58:f50b97b08851 750 }
markrad 58:f50b97b08851 751 }
markrad 58:f50b97b08851 752
markrad 58:f50b97b08851 753 int session_begin(SESSION_HANDLE session)
markrad 58:f50b97b08851 754 {
markrad 58:f50b97b08851 755 int result;
markrad 58:f50b97b08851 756
markrad 58:f50b97b08851 757 if (session == NULL)
markrad 58:f50b97b08851 758 {
markrad 58:f50b97b08851 759 result = __LINE__;
markrad 58:f50b97b08851 760 }
markrad 58:f50b97b08851 761 else
markrad 58:f50b97b08851 762 {
markrad 58:f50b97b08851 763 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)session;
markrad 58:f50b97b08851 764
markrad 58:f50b97b08851 765 if (connection_start_endpoint(session_instance->endpoint, on_frame_received, on_connection_state_changed, session_instance) != 0)
markrad 58:f50b97b08851 766 {
markrad 58:f50b97b08851 767 result = __LINE__;
markrad 58:f50b97b08851 768 }
markrad 58:f50b97b08851 769 else
markrad 58:f50b97b08851 770 {
markrad 58:f50b97b08851 771 if (!session_instance->is_underlying_connection_open)
markrad 58:f50b97b08851 772 {
markrad 58:f50b97b08851 773 if (connection_open(session_instance->connection) != 0)
markrad 58:f50b97b08851 774 {
markrad 58:f50b97b08851 775 session_instance->is_underlying_connection_open = UNDERLYING_CONNECTION_NOT_OPEN;
markrad 58:f50b97b08851 776 result = __LINE__;
markrad 58:f50b97b08851 777 }
markrad 58:f50b97b08851 778 else
markrad 58:f50b97b08851 779 {
markrad 58:f50b97b08851 780 session_instance->is_underlying_connection_open = UNDERLYING_CONNECTION_OPEN;
markrad 58:f50b97b08851 781 result = 0;
markrad 58:f50b97b08851 782 }
markrad 58:f50b97b08851 783 }
markrad 58:f50b97b08851 784 else
markrad 58:f50b97b08851 785 {
markrad 58:f50b97b08851 786 result = 0;
markrad 58:f50b97b08851 787 }
markrad 58:f50b97b08851 788 }
markrad 58:f50b97b08851 789 }
markrad 58:f50b97b08851 790
markrad 58:f50b97b08851 791 return result;
markrad 58:f50b97b08851 792 }
markrad 58:f50b97b08851 793
markrad 58:f50b97b08851 794 int session_end(SESSION_HANDLE session, const char* condition_value, const char* description)
markrad 58:f50b97b08851 795 {
markrad 58:f50b97b08851 796 int result;
markrad 58:f50b97b08851 797
markrad 58:f50b97b08851 798 if (session == NULL)
markrad 58:f50b97b08851 799 {
markrad 58:f50b97b08851 800 result = __LINE__;
markrad 58:f50b97b08851 801 }
markrad 58:f50b97b08851 802 else
markrad 58:f50b97b08851 803 {
markrad 58:f50b97b08851 804 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)session;
markrad 58:f50b97b08851 805
markrad 58:f50b97b08851 806 if ((session_instance->session_state != SESSION_STATE_UNMAPPED) &&
markrad 58:f50b97b08851 807 (session_instance->session_state != SESSION_STATE_DISCARDING))
markrad 58:f50b97b08851 808 {
markrad 58:f50b97b08851 809 ERROR_HANDLE error_handle = NULL;
markrad 58:f50b97b08851 810 result = 0;
markrad 58:f50b97b08851 811
markrad 58:f50b97b08851 812 if (condition_value != NULL)
markrad 58:f50b97b08851 813 {
markrad 58:f50b97b08851 814 error_handle = error_create(condition_value);
markrad 58:f50b97b08851 815 if (error_handle == NULL)
markrad 58:f50b97b08851 816 {
markrad 58:f50b97b08851 817 result = __LINE__;
markrad 58:f50b97b08851 818 }
markrad 58:f50b97b08851 819 else
markrad 58:f50b97b08851 820 {
markrad 58:f50b97b08851 821 if (error_set_description(error_handle, description) != 0)
markrad 58:f50b97b08851 822 {
markrad 58:f50b97b08851 823 result = __LINE__;
markrad 58:f50b97b08851 824 }
markrad 58:f50b97b08851 825 }
markrad 58:f50b97b08851 826 }
markrad 58:f50b97b08851 827
markrad 58:f50b97b08851 828 if (result == 0)
markrad 58:f50b97b08851 829 {
markrad 58:f50b97b08851 830 if (send_end_frame(session_instance, error_handle) != 0)
markrad 58:f50b97b08851 831 {
markrad 58:f50b97b08851 832 result = __LINE__;
markrad 58:f50b97b08851 833 }
markrad 58:f50b97b08851 834 else
markrad 58:f50b97b08851 835 {
markrad 58:f50b97b08851 836 session_set_state(session_instance, SESSION_STATE_DISCARDING);
markrad 58:f50b97b08851 837 result = 0;
markrad 58:f50b97b08851 838 }
markrad 58:f50b97b08851 839 }
markrad 58:f50b97b08851 840
markrad 58:f50b97b08851 841 if (error_handle != NULL)
markrad 58:f50b97b08851 842 {
markrad 58:f50b97b08851 843 error_destroy(error_handle);
markrad 58:f50b97b08851 844 }
markrad 58:f50b97b08851 845 }
markrad 58:f50b97b08851 846 else
markrad 58:f50b97b08851 847 {
markrad 58:f50b97b08851 848 result = 0;
markrad 58:f50b97b08851 849 }
markrad 58:f50b97b08851 850 }
markrad 58:f50b97b08851 851
markrad 58:f50b97b08851 852 return result;
markrad 58:f50b97b08851 853 }
markrad 58:f50b97b08851 854
markrad 58:f50b97b08851 855 int session_set_incoming_window(SESSION_HANDLE session, uint32_t incoming_window)
markrad 58:f50b97b08851 856 {
markrad 58:f50b97b08851 857 int result;
markrad 58:f50b97b08851 858
markrad 58:f50b97b08851 859 if (session == NULL)
markrad 58:f50b97b08851 860 {
markrad 58:f50b97b08851 861 result = __LINE__;
markrad 58:f50b97b08851 862 }
markrad 58:f50b97b08851 863 else
markrad 58:f50b97b08851 864 {
markrad 58:f50b97b08851 865 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)session;
markrad 58:f50b97b08851 866
markrad 58:f50b97b08851 867 session_instance->desired_incoming_window = incoming_window;
markrad 58:f50b97b08851 868 session_instance->incoming_window = incoming_window;
markrad 58:f50b97b08851 869
markrad 58:f50b97b08851 870 result = 0;
markrad 58:f50b97b08851 871 }
markrad 58:f50b97b08851 872
markrad 58:f50b97b08851 873 return result;
markrad 58:f50b97b08851 874 }
markrad 58:f50b97b08851 875
markrad 58:f50b97b08851 876 int session_get_incoming_window(SESSION_HANDLE session, uint32_t* incoming_window)
markrad 58:f50b97b08851 877 {
markrad 58:f50b97b08851 878 int result;
markrad 58:f50b97b08851 879
markrad 58:f50b97b08851 880 if ((session == NULL) ||
markrad 58:f50b97b08851 881 (incoming_window == NULL))
markrad 58:f50b97b08851 882 {
markrad 58:f50b97b08851 883 result = __LINE__;
markrad 58:f50b97b08851 884 }
markrad 58:f50b97b08851 885 else
markrad 58:f50b97b08851 886 {
markrad 58:f50b97b08851 887 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)session;
markrad 58:f50b97b08851 888
markrad 58:f50b97b08851 889 *incoming_window = session_instance->incoming_window;
markrad 58:f50b97b08851 890
markrad 58:f50b97b08851 891 result = 0;
markrad 58:f50b97b08851 892 }
markrad 58:f50b97b08851 893
markrad 58:f50b97b08851 894 return result;
markrad 58:f50b97b08851 895 }
markrad 58:f50b97b08851 896
markrad 58:f50b97b08851 897 int session_set_outgoing_window(SESSION_HANDLE session, uint32_t outgoing_window)
markrad 58:f50b97b08851 898 {
markrad 58:f50b97b08851 899 int result;
markrad 58:f50b97b08851 900
markrad 58:f50b97b08851 901 if (session == NULL)
markrad 58:f50b97b08851 902 {
markrad 58:f50b97b08851 903 result = __LINE__;
markrad 58:f50b97b08851 904 }
markrad 58:f50b97b08851 905 else
markrad 58:f50b97b08851 906 {
markrad 58:f50b97b08851 907 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)session;
markrad 58:f50b97b08851 908
markrad 58:f50b97b08851 909 session_instance->outgoing_window = outgoing_window;
markrad 58:f50b97b08851 910
markrad 58:f50b97b08851 911 result = 0;
markrad 58:f50b97b08851 912 }
markrad 58:f50b97b08851 913
markrad 58:f50b97b08851 914 return result;
markrad 58:f50b97b08851 915 }
markrad 58:f50b97b08851 916
markrad 58:f50b97b08851 917 int session_get_outgoing_window(SESSION_HANDLE session, uint32_t* outgoing_window)
markrad 58:f50b97b08851 918 {
markrad 58:f50b97b08851 919 int result;
markrad 58:f50b97b08851 920
markrad 58:f50b97b08851 921 if ((session == NULL) ||
markrad 58:f50b97b08851 922 (outgoing_window == NULL))
markrad 58:f50b97b08851 923 {
markrad 58:f50b97b08851 924 result = __LINE__;
markrad 58:f50b97b08851 925 }
markrad 58:f50b97b08851 926 else
markrad 58:f50b97b08851 927 {
markrad 58:f50b97b08851 928 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)session;
markrad 58:f50b97b08851 929
markrad 58:f50b97b08851 930 *outgoing_window = session_instance->outgoing_window;
markrad 58:f50b97b08851 931
markrad 58:f50b97b08851 932 result = 0;
markrad 58:f50b97b08851 933 }
markrad 58:f50b97b08851 934
markrad 58:f50b97b08851 935 return result;
markrad 58:f50b97b08851 936 }
markrad 58:f50b97b08851 937
markrad 58:f50b97b08851 938 int session_set_handle_max(SESSION_HANDLE session, handle handle_max)
markrad 58:f50b97b08851 939 {
markrad 58:f50b97b08851 940 int result;
markrad 58:f50b97b08851 941
markrad 58:f50b97b08851 942 if (session == NULL)
markrad 58:f50b97b08851 943 {
markrad 58:f50b97b08851 944 result = __LINE__;
markrad 58:f50b97b08851 945 }
markrad 58:f50b97b08851 946 else
markrad 58:f50b97b08851 947 {
markrad 58:f50b97b08851 948 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)session;
markrad 58:f50b97b08851 949
markrad 58:f50b97b08851 950 session_instance->handle_max = handle_max;
markrad 58:f50b97b08851 951
markrad 58:f50b97b08851 952 result = 0;
markrad 58:f50b97b08851 953 }
markrad 58:f50b97b08851 954
markrad 58:f50b97b08851 955 return result;
markrad 58:f50b97b08851 956 }
markrad 58:f50b97b08851 957
markrad 58:f50b97b08851 958 int session_get_handle_max(SESSION_HANDLE session, handle* handle_max)
markrad 58:f50b97b08851 959 {
markrad 58:f50b97b08851 960 int result;
markrad 58:f50b97b08851 961
markrad 58:f50b97b08851 962 if ((session == NULL) ||
markrad 58:f50b97b08851 963 (handle_max == NULL))
markrad 58:f50b97b08851 964 {
markrad 58:f50b97b08851 965 result = __LINE__;
markrad 58:f50b97b08851 966 }
markrad 58:f50b97b08851 967 else
markrad 58:f50b97b08851 968 {
markrad 58:f50b97b08851 969 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)session;
markrad 58:f50b97b08851 970
markrad 58:f50b97b08851 971 *handle_max = session_instance->handle_max;
markrad 58:f50b97b08851 972
markrad 58:f50b97b08851 973 result = 0;
markrad 58:f50b97b08851 974 }
markrad 58:f50b97b08851 975
markrad 58:f50b97b08851 976 return result;
markrad 58:f50b97b08851 977 }
markrad 58:f50b97b08851 978
markrad 58:f50b97b08851 979 LINK_ENDPOINT_HANDLE session_create_link_endpoint(SESSION_HANDLE session, const char* name)
markrad 58:f50b97b08851 980 {
markrad 58:f50b97b08851 981 LINK_ENDPOINT_INSTANCE* result;
markrad 58:f50b97b08851 982
markrad 58:f50b97b08851 983 /* Codes_SRS_SESSION_01_044: [If session, name or frame_received_callback is NULL, session_create_link_endpoint shall fail and return NULL.] */
markrad 58:f50b97b08851 984 if ((session == NULL) ||
markrad 58:f50b97b08851 985 (name == NULL))
markrad 58:f50b97b08851 986 {
markrad 58:f50b97b08851 987 result = NULL;
markrad 58:f50b97b08851 988 }
markrad 58:f50b97b08851 989 else
markrad 58:f50b97b08851 990 {
markrad 58:f50b97b08851 991 /* Codes_SRS_SESSION_01_043: [session_create_link_endpoint shall create a link endpoint associated with a given session and return a non-NULL handle to it.] */
markrad 58:f50b97b08851 992 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)session;
markrad 58:f50b97b08851 993
markrad 58:f50b97b08851 994 result = (LINK_ENDPOINT_INSTANCE*)amqpalloc_malloc(sizeof(LINK_ENDPOINT_INSTANCE));
markrad 58:f50b97b08851 995 /* Codes_SRS_SESSION_01_045: [If allocating memory for the link endpoint fails, session_create_link_endpoint shall fail and return NULL.] */
markrad 58:f50b97b08851 996 if (result != NULL)
markrad 58:f50b97b08851 997 {
markrad 58:f50b97b08851 998 /* Codes_SRS_SESSION_01_046: [An unused handle shall be assigned to the link endpoint.] */
markrad 58:f50b97b08851 999 handle selected_handle = 0;
markrad 58:f50b97b08851 1000 size_t i;
markrad 58:f50b97b08851 1001
markrad 58:f50b97b08851 1002 for (i = 0; i < session_instance->link_endpoint_count; i++)
markrad 58:f50b97b08851 1003 {
markrad 58:f50b97b08851 1004 if (session_instance->link_endpoints[i]->output_handle > selected_handle)
markrad 58:f50b97b08851 1005 {
markrad 58:f50b97b08851 1006 break;
markrad 58:f50b97b08851 1007 }
markrad 58:f50b97b08851 1008
markrad 58:f50b97b08851 1009 selected_handle++;
markrad 58:f50b97b08851 1010 }
markrad 58:f50b97b08851 1011
markrad 58:f50b97b08851 1012 result->on_session_state_changed = NULL;
markrad 58:f50b97b08851 1013 result->on_session_flow_on = NULL;
markrad 58:f50b97b08851 1014 result->frame_received_callback = NULL;
markrad 58:f50b97b08851 1015 result->callback_context = NULL;
markrad 58:f50b97b08851 1016 result->output_handle = selected_handle;
markrad 58:f50b97b08851 1017 result->input_handle = 0xFFFFFFFF;
markrad 58:f50b97b08851 1018 result->name = amqpalloc_malloc(strlen(name) + 1);
markrad 58:f50b97b08851 1019 if (result->name == NULL)
markrad 58:f50b97b08851 1020 {
markrad 58:f50b97b08851 1021 /* Codes_SRS_SESSION_01_045: [If allocating memory for the link endpoint fails, session_create_link_endpoint shall fail and return NULL.] */
markrad 58:f50b97b08851 1022 amqpalloc_free(result);
markrad 58:f50b97b08851 1023 result = NULL;
markrad 58:f50b97b08851 1024 }
markrad 58:f50b97b08851 1025 else
markrad 58:f50b97b08851 1026 {
markrad 58:f50b97b08851 1027 LINK_ENDPOINT_INSTANCE** new_link_endpoints;
markrad 58:f50b97b08851 1028 strcpy(result->name, name);
markrad 58:f50b97b08851 1029 result->session = session;
markrad 58:f50b97b08851 1030
markrad 58:f50b97b08851 1031 new_link_endpoints = amqpalloc_realloc(session_instance->link_endpoints, sizeof(LINK_ENDPOINT_INSTANCE*) * (session_instance->link_endpoint_count + 1));
markrad 58:f50b97b08851 1032 if (new_link_endpoints == NULL)
markrad 58:f50b97b08851 1033 {
markrad 58:f50b97b08851 1034 /* Codes_SRS_SESSION_01_045: [If allocating memory for the link endpoint fails, session_create_link_endpoint shall fail and return NULL.] */
markrad 58:f50b97b08851 1035 amqpalloc_free(result->name);
markrad 58:f50b97b08851 1036 amqpalloc_free(result);
markrad 58:f50b97b08851 1037 result = NULL;
markrad 58:f50b97b08851 1038 }
markrad 58:f50b97b08851 1039 else
markrad 58:f50b97b08851 1040 {
markrad 58:f50b97b08851 1041 session_instance->link_endpoints = new_link_endpoints;
markrad 58:f50b97b08851 1042
markrad 58:f50b97b08851 1043 if (session_instance->link_endpoint_count - selected_handle > 0)
markrad 58:f50b97b08851 1044 {
markrad 58:f50b97b08851 1045 (void)memmove(&session_instance->link_endpoints[selected_handle + 1], &session_instance->link_endpoints[selected_handle], (session_instance->link_endpoint_count - selected_handle) * sizeof(LINK_ENDPOINT_INSTANCE*));
markrad 58:f50b97b08851 1046 }
markrad 58:f50b97b08851 1047
markrad 58:f50b97b08851 1048 session_instance->link_endpoints[selected_handle] = result;
markrad 58:f50b97b08851 1049 session_instance->link_endpoint_count++;
markrad 58:f50b97b08851 1050 }
markrad 58:f50b97b08851 1051 }
markrad 58:f50b97b08851 1052 }
markrad 58:f50b97b08851 1053 }
markrad 58:f50b97b08851 1054
markrad 58:f50b97b08851 1055 return result;
markrad 58:f50b97b08851 1056 }
markrad 58:f50b97b08851 1057
markrad 58:f50b97b08851 1058 void session_destroy_link_endpoint(LINK_ENDPOINT_HANDLE link_endpoint)
markrad 58:f50b97b08851 1059 {
markrad 58:f50b97b08851 1060 /* Codes_SRS_SESSION_01_050: [If link_endpoint is NULL, session_destroy_link_endpoint shall do nothing.] */
markrad 58:f50b97b08851 1061 if (link_endpoint != NULL)
markrad 58:f50b97b08851 1062 {
markrad 58:f50b97b08851 1063 LINK_ENDPOINT_INSTANCE* endpoint_instance = (LINK_ENDPOINT_INSTANCE*)link_endpoint;
markrad 58:f50b97b08851 1064 SESSION_INSTANCE* session_instance = endpoint_instance->session;
markrad 58:f50b97b08851 1065 uint64_t i;
markrad 58:f50b97b08851 1066
markrad 58:f50b97b08851 1067 /* Codes_SRS_SESSION_01_049: [session_destroy_link_endpoint shall free all resources associated with the endpoint.] */
markrad 58:f50b97b08851 1068 for (i = 0; i < session_instance->link_endpoint_count; i++)
markrad 58:f50b97b08851 1069 {
markrad 58:f50b97b08851 1070 if (session_instance->link_endpoints[i] == link_endpoint)
markrad 58:f50b97b08851 1071 {
markrad 58:f50b97b08851 1072 break;
markrad 58:f50b97b08851 1073 }
markrad 58:f50b97b08851 1074 }
markrad 58:f50b97b08851 1075
markrad 58:f50b97b08851 1076 if (i < session_instance->link_endpoint_count)
markrad 58:f50b97b08851 1077 {
markrad 58:f50b97b08851 1078 LINK_ENDPOINT_INSTANCE** new_endpoints;
markrad 58:f50b97b08851 1079
markrad 58:f50b97b08851 1080 if (i < (session_instance->link_endpoint_count - 1))
markrad 58:f50b97b08851 1081 {
markrad 58:f50b97b08851 1082 (void)memmove(&session_instance->link_endpoints[i], &session_instance->link_endpoints[i + 1], (session_instance->link_endpoint_count - (uint32_t)i - 1) * sizeof(LINK_ENDPOINT_INSTANCE*));
markrad 58:f50b97b08851 1083 }
markrad 58:f50b97b08851 1084
markrad 58:f50b97b08851 1085 session_instance->link_endpoint_count--;
markrad 58:f50b97b08851 1086
markrad 58:f50b97b08851 1087 if (session_instance->link_endpoint_count == 0)
markrad 58:f50b97b08851 1088 {
markrad 58:f50b97b08851 1089 amqpalloc_free(session_instance->link_endpoints);
markrad 58:f50b97b08851 1090 session_instance->link_endpoints = NULL;
markrad 58:f50b97b08851 1091 }
markrad 58:f50b97b08851 1092 else
markrad 58:f50b97b08851 1093 {
markrad 58:f50b97b08851 1094 new_endpoints = (LINK_ENDPOINT_INSTANCE**)amqpalloc_realloc(session_instance->link_endpoints, sizeof(LINK_ENDPOINT_INSTANCE*) * session_instance->link_endpoint_count);
markrad 58:f50b97b08851 1095 if (new_endpoints != NULL)
markrad 58:f50b97b08851 1096 {
markrad 58:f50b97b08851 1097 session_instance->link_endpoints = new_endpoints;
markrad 58:f50b97b08851 1098 }
markrad 58:f50b97b08851 1099 }
markrad 58:f50b97b08851 1100 }
markrad 58:f50b97b08851 1101
markrad 58:f50b97b08851 1102 if (endpoint_instance->name != NULL)
markrad 58:f50b97b08851 1103 {
markrad 58:f50b97b08851 1104 amqpalloc_free(endpoint_instance->name);
markrad 58:f50b97b08851 1105 }
markrad 58:f50b97b08851 1106
markrad 58:f50b97b08851 1107 amqpalloc_free(endpoint_instance);
markrad 58:f50b97b08851 1108 }
markrad 58:f50b97b08851 1109 }
markrad 58:f50b97b08851 1110
markrad 58:f50b97b08851 1111 int session_start_link_endpoint(LINK_ENDPOINT_HANDLE link_endpoint, ON_ENDPOINT_FRAME_RECEIVED frame_received_callback, ON_SESSION_STATE_CHANGED on_session_state_changed, ON_SESSION_FLOW_ON on_session_flow_on, void* context)
markrad 58:f50b97b08851 1112 {
markrad 58:f50b97b08851 1113 int result;
markrad 58:f50b97b08851 1114
markrad 58:f50b97b08851 1115 if ((link_endpoint == NULL) ||
markrad 58:f50b97b08851 1116 (frame_received_callback == NULL))
markrad 58:f50b97b08851 1117 {
markrad 58:f50b97b08851 1118 result = __LINE__;
markrad 58:f50b97b08851 1119 }
markrad 58:f50b97b08851 1120 else
markrad 58:f50b97b08851 1121 {
markrad 58:f50b97b08851 1122 link_endpoint->frame_received_callback = frame_received_callback;
markrad 58:f50b97b08851 1123 link_endpoint->on_session_state_changed = on_session_state_changed;
markrad 58:f50b97b08851 1124 link_endpoint->on_session_flow_on = on_session_flow_on;
markrad 58:f50b97b08851 1125 link_endpoint->callback_context = context;
markrad 58:f50b97b08851 1126
markrad 58:f50b97b08851 1127 if (link_endpoint->on_session_state_changed != NULL)
markrad 58:f50b97b08851 1128 {
markrad 58:f50b97b08851 1129 link_endpoint->on_session_state_changed(link_endpoint->callback_context, link_endpoint->session->session_state, link_endpoint->session->previous_session_state);
markrad 58:f50b97b08851 1130 }
markrad 58:f50b97b08851 1131
markrad 58:f50b97b08851 1132 result = 0;
markrad 58:f50b97b08851 1133 }
markrad 58:f50b97b08851 1134
markrad 58:f50b97b08851 1135 return result;
markrad 58:f50b97b08851 1136 }
markrad 58:f50b97b08851 1137
markrad 58:f50b97b08851 1138 static int encode_frame(LINK_ENDPOINT_HANDLE link_endpoint, const AMQP_VALUE performative, PAYLOAD* payloads, size_t payload_count)
markrad 58:f50b97b08851 1139 {
markrad 58:f50b97b08851 1140 int result;
markrad 58:f50b97b08851 1141
markrad 58:f50b97b08851 1142 if ((link_endpoint == NULL) ||
markrad 58:f50b97b08851 1143 (performative == NULL))
markrad 58:f50b97b08851 1144 {
markrad 58:f50b97b08851 1145 result = __LINE__;
markrad 58:f50b97b08851 1146 }
markrad 58:f50b97b08851 1147 else
markrad 58:f50b97b08851 1148 {
markrad 58:f50b97b08851 1149 LINK_ENDPOINT_INSTANCE* link_endpoint_instance = (LINK_ENDPOINT_INSTANCE*)link_endpoint;
markrad 58:f50b97b08851 1150 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)link_endpoint_instance->session;
markrad 58:f50b97b08851 1151
markrad 58:f50b97b08851 1152 if (connection_encode_frame(session_instance->endpoint, performative, payloads, payload_count, NULL, NULL) != 0)
markrad 58:f50b97b08851 1153 {
markrad 58:f50b97b08851 1154 result = __LINE__;
markrad 58:f50b97b08851 1155 }
markrad 58:f50b97b08851 1156 else
markrad 58:f50b97b08851 1157 {
markrad 58:f50b97b08851 1158 result = 0;
markrad 58:f50b97b08851 1159 }
markrad 58:f50b97b08851 1160 }
markrad 58:f50b97b08851 1161
markrad 58:f50b97b08851 1162 return result;
markrad 58:f50b97b08851 1163 }
markrad 58:f50b97b08851 1164
markrad 58:f50b97b08851 1165 int session_send_flow(LINK_ENDPOINT_HANDLE link_endpoint, FLOW_HANDLE flow)
markrad 58:f50b97b08851 1166 {
markrad 58:f50b97b08851 1167 int result;
markrad 58:f50b97b08851 1168
markrad 58:f50b97b08851 1169 if ((link_endpoint == NULL) ||
markrad 58:f50b97b08851 1170 (flow == NULL))
markrad 58:f50b97b08851 1171 {
markrad 58:f50b97b08851 1172 result = __LINE__;
markrad 58:f50b97b08851 1173 }
markrad 58:f50b97b08851 1174 else
markrad 58:f50b97b08851 1175 {
markrad 58:f50b97b08851 1176 LINK_ENDPOINT_INSTANCE* link_endpoint_instance = (LINK_ENDPOINT_INSTANCE*)link_endpoint;
markrad 58:f50b97b08851 1177 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)link_endpoint_instance->session;
markrad 58:f50b97b08851 1178
markrad 58:f50b97b08851 1179 result = 0;
markrad 58:f50b97b08851 1180
markrad 58:f50b97b08851 1181 if (session_instance->session_state == SESSION_STATE_BEGIN_RCVD)
markrad 58:f50b97b08851 1182 {
markrad 58:f50b97b08851 1183 if (flow_set_next_incoming_id(flow, session_instance->next_incoming_id) != 0)
markrad 58:f50b97b08851 1184 {
markrad 58:f50b97b08851 1185 result = __LINE__;
markrad 58:f50b97b08851 1186 }
markrad 58:f50b97b08851 1187 }
markrad 58:f50b97b08851 1188
markrad 58:f50b97b08851 1189 if (result == 0)
markrad 58:f50b97b08851 1190 {
markrad 58:f50b97b08851 1191 if ((flow_set_next_incoming_id(flow, session_instance->next_incoming_id) != 0) ||
markrad 58:f50b97b08851 1192 (flow_set_incoming_window(flow, session_instance->incoming_window) != 0) ||
markrad 58:f50b97b08851 1193 (flow_set_next_outgoing_id(flow, session_instance->next_outgoing_id) != 0) ||
markrad 58:f50b97b08851 1194 (flow_set_outgoing_window(flow, session_instance->outgoing_window) != 0) ||
markrad 58:f50b97b08851 1195 (flow_set_handle(flow, link_endpoint_instance->output_handle) != 0))
markrad 58:f50b97b08851 1196 {
markrad 58:f50b97b08851 1197 result = __LINE__;
markrad 58:f50b97b08851 1198 }
markrad 58:f50b97b08851 1199 else
markrad 58:f50b97b08851 1200 {
markrad 58:f50b97b08851 1201 AMQP_VALUE flow_performative_value = amqpvalue_create_flow(flow);
markrad 58:f50b97b08851 1202 if (flow_performative_value == NULL)
markrad 58:f50b97b08851 1203 {
markrad 58:f50b97b08851 1204 result = __LINE__;
markrad 58:f50b97b08851 1205 }
markrad 58:f50b97b08851 1206 else
markrad 58:f50b97b08851 1207 {
markrad 58:f50b97b08851 1208 if (encode_frame(link_endpoint, flow_performative_value, NULL, 0) != 0)
markrad 58:f50b97b08851 1209 {
markrad 58:f50b97b08851 1210 result = __LINE__;
markrad 58:f50b97b08851 1211 }
markrad 58:f50b97b08851 1212 else
markrad 58:f50b97b08851 1213 {
markrad 58:f50b97b08851 1214 result = 0;
markrad 58:f50b97b08851 1215 }
markrad 58:f50b97b08851 1216
markrad 58:f50b97b08851 1217 amqpvalue_destroy(flow_performative_value);
markrad 58:f50b97b08851 1218 }
markrad 58:f50b97b08851 1219 }
markrad 58:f50b97b08851 1220 }
markrad 58:f50b97b08851 1221 }
markrad 58:f50b97b08851 1222
markrad 58:f50b97b08851 1223 return result;
markrad 58:f50b97b08851 1224 }
markrad 58:f50b97b08851 1225
markrad 58:f50b97b08851 1226 int session_send_attach(LINK_ENDPOINT_HANDLE link_endpoint, ATTACH_HANDLE attach)
markrad 58:f50b97b08851 1227 {
markrad 58:f50b97b08851 1228 int result;
markrad 58:f50b97b08851 1229
markrad 58:f50b97b08851 1230 if ((link_endpoint == NULL) ||
markrad 58:f50b97b08851 1231 (attach == NULL))
markrad 58:f50b97b08851 1232 {
markrad 58:f50b97b08851 1233 result = __LINE__;
markrad 58:f50b97b08851 1234 }
markrad 58:f50b97b08851 1235 else
markrad 58:f50b97b08851 1236 {
markrad 58:f50b97b08851 1237 LINK_ENDPOINT_INSTANCE* link_endpoint_instance = (LINK_ENDPOINT_INSTANCE*)link_endpoint;
markrad 58:f50b97b08851 1238
markrad 58:f50b97b08851 1239 if (attach_set_handle(attach, link_endpoint_instance->output_handle) != 0)
markrad 58:f50b97b08851 1240 {
markrad 58:f50b97b08851 1241 result = __LINE__;
markrad 58:f50b97b08851 1242 }
markrad 58:f50b97b08851 1243 else
markrad 58:f50b97b08851 1244 {
markrad 58:f50b97b08851 1245 AMQP_VALUE attach_performative_value = amqpvalue_create_attach(attach);
markrad 58:f50b97b08851 1246 if (attach_performative_value == NULL)
markrad 58:f50b97b08851 1247 {
markrad 58:f50b97b08851 1248 result = __LINE__;
markrad 58:f50b97b08851 1249 }
markrad 58:f50b97b08851 1250 else
markrad 58:f50b97b08851 1251 {
markrad 58:f50b97b08851 1252 if (encode_frame(link_endpoint, attach_performative_value, NULL, 0) != 0)
markrad 58:f50b97b08851 1253 {
markrad 58:f50b97b08851 1254 result = __LINE__;
markrad 58:f50b97b08851 1255 }
markrad 58:f50b97b08851 1256 else
markrad 58:f50b97b08851 1257 {
markrad 58:f50b97b08851 1258 result = 0;
markrad 58:f50b97b08851 1259 }
markrad 58:f50b97b08851 1260
markrad 58:f50b97b08851 1261 amqpvalue_destroy(attach_performative_value);
markrad 58:f50b97b08851 1262 }
markrad 58:f50b97b08851 1263 }
markrad 58:f50b97b08851 1264 }
markrad 58:f50b97b08851 1265
markrad 58:f50b97b08851 1266 return result;
markrad 58:f50b97b08851 1267 }
markrad 58:f50b97b08851 1268
markrad 58:f50b97b08851 1269 int session_send_disposition(LINK_ENDPOINT_HANDLE link_endpoint, DISPOSITION_HANDLE disposition)
markrad 58:f50b97b08851 1270 {
markrad 58:f50b97b08851 1271 int result;
markrad 58:f50b97b08851 1272
markrad 58:f50b97b08851 1273 if ((link_endpoint == NULL) ||
markrad 58:f50b97b08851 1274 (disposition == NULL))
markrad 58:f50b97b08851 1275 {
markrad 58:f50b97b08851 1276 result = __LINE__;
markrad 58:f50b97b08851 1277 }
markrad 58:f50b97b08851 1278 else
markrad 58:f50b97b08851 1279 {
markrad 58:f50b97b08851 1280 AMQP_VALUE disposition_performative_value = amqpvalue_create_disposition(disposition);
markrad 58:f50b97b08851 1281 if (disposition_performative_value == NULL)
markrad 58:f50b97b08851 1282 {
markrad 58:f50b97b08851 1283 result = __LINE__;
markrad 58:f50b97b08851 1284 }
markrad 58:f50b97b08851 1285 else
markrad 58:f50b97b08851 1286 {
markrad 58:f50b97b08851 1287 if (encode_frame(link_endpoint, disposition_performative_value, NULL, 0) != 0)
markrad 58:f50b97b08851 1288 {
markrad 58:f50b97b08851 1289 result = __LINE__;
markrad 58:f50b97b08851 1290 }
markrad 58:f50b97b08851 1291 else
markrad 58:f50b97b08851 1292 {
markrad 58:f50b97b08851 1293 result = 0;
markrad 58:f50b97b08851 1294 }
markrad 58:f50b97b08851 1295
markrad 58:f50b97b08851 1296 amqpvalue_destroy(disposition_performative_value);
markrad 58:f50b97b08851 1297 }
markrad 58:f50b97b08851 1298 }
markrad 58:f50b97b08851 1299
markrad 58:f50b97b08851 1300 return result;
markrad 58:f50b97b08851 1301 }
markrad 58:f50b97b08851 1302
markrad 58:f50b97b08851 1303 int session_send_detach(LINK_ENDPOINT_HANDLE link_endpoint, DETACH_HANDLE detach)
markrad 58:f50b97b08851 1304 {
markrad 58:f50b97b08851 1305 int result;
markrad 58:f50b97b08851 1306
markrad 58:f50b97b08851 1307 if ((link_endpoint == NULL) ||
markrad 58:f50b97b08851 1308 (detach == NULL))
markrad 58:f50b97b08851 1309 {
markrad 58:f50b97b08851 1310 result = __LINE__;
markrad 58:f50b97b08851 1311 }
markrad 58:f50b97b08851 1312 else
markrad 58:f50b97b08851 1313 {
markrad 58:f50b97b08851 1314 LINK_ENDPOINT_INSTANCE* link_endpoint_instance = (LINK_ENDPOINT_INSTANCE*)link_endpoint;
markrad 58:f50b97b08851 1315
markrad 58:f50b97b08851 1316 if (detach_set_handle(detach, link_endpoint_instance->output_handle) != 0)
markrad 58:f50b97b08851 1317 {
markrad 58:f50b97b08851 1318 result = __LINE__;
markrad 58:f50b97b08851 1319 }
markrad 58:f50b97b08851 1320 else
markrad 58:f50b97b08851 1321 {
markrad 58:f50b97b08851 1322 AMQP_VALUE detach_performative_value = amqpvalue_create_detach(detach);
markrad 58:f50b97b08851 1323 if (detach_performative_value == NULL)
markrad 58:f50b97b08851 1324 {
markrad 58:f50b97b08851 1325 result = __LINE__;
markrad 58:f50b97b08851 1326 }
markrad 58:f50b97b08851 1327 else
markrad 58:f50b97b08851 1328 {
markrad 58:f50b97b08851 1329 if (encode_frame(link_endpoint, detach_performative_value, NULL, 0) != 0)
markrad 58:f50b97b08851 1330 {
markrad 58:f50b97b08851 1331 result = __LINE__;
markrad 58:f50b97b08851 1332 }
markrad 58:f50b97b08851 1333 else
markrad 58:f50b97b08851 1334 {
markrad 58:f50b97b08851 1335 result = 0;
markrad 58:f50b97b08851 1336 }
markrad 58:f50b97b08851 1337
markrad 58:f50b97b08851 1338 amqpvalue_destroy(detach_performative_value);
markrad 58:f50b97b08851 1339 }
markrad 58:f50b97b08851 1340 }
markrad 58:f50b97b08851 1341 }
markrad 58:f50b97b08851 1342
markrad 58:f50b97b08851 1343 return result;
markrad 58:f50b97b08851 1344 }
markrad 58:f50b97b08851 1345
markrad 58:f50b97b08851 1346 /* Codes_SRS_SESSION_01_051: [session_send_transfer shall send a transfer frame with the performative indicated in the transfer argument.] */
markrad 58:f50b97b08851 1347 SESSION_SEND_TRANSFER_RESULT session_send_transfer(LINK_ENDPOINT_HANDLE link_endpoint, TRANSFER_HANDLE transfer, PAYLOAD* payloads, size_t payload_count, delivery_number* delivery_id, ON_SEND_COMPLETE on_send_complete, void* callback_context)
markrad 58:f50b97b08851 1348 {
markrad 58:f50b97b08851 1349 SESSION_SEND_TRANSFER_RESULT result;
markrad 58:f50b97b08851 1350
markrad 58:f50b97b08851 1351 /* Codes_SRS_SESSION_01_054: [If link_endpoint or transfer is NULL, session_send_transfer shall fail and return a non-zero value.] */
markrad 58:f50b97b08851 1352 if ((link_endpoint == NULL) ||
markrad 58:f50b97b08851 1353 (transfer == NULL))
markrad 58:f50b97b08851 1354 {
markrad 58:f50b97b08851 1355 result = SESSION_SEND_TRANSFER_ERROR;
markrad 58:f50b97b08851 1356 }
markrad 58:f50b97b08851 1357 else
markrad 58:f50b97b08851 1358 {
markrad 58:f50b97b08851 1359 LINK_ENDPOINT_INSTANCE* link_endpoint_instance = (LINK_ENDPOINT_INSTANCE*)link_endpoint;
markrad 58:f50b97b08851 1360 SESSION_INSTANCE* session_instance = (SESSION_INSTANCE*)link_endpoint_instance->session;
markrad 58:f50b97b08851 1361
markrad 58:f50b97b08851 1362 /* Codes_SRS_SESSION_01_059: [When session_send_transfer is called while the session is not in the MAPPED state, session_send_transfer shall fail and return a non-zero value.] */
markrad 58:f50b97b08851 1363 if (session_instance->session_state != SESSION_STATE_MAPPED)
markrad 58:f50b97b08851 1364 {
markrad 58:f50b97b08851 1365 result = SESSION_SEND_TRANSFER_ERROR;
markrad 58:f50b97b08851 1366 }
markrad 58:f50b97b08851 1367 else
markrad 58:f50b97b08851 1368 {
markrad 58:f50b97b08851 1369 size_t payload_size = 0;
markrad 58:f50b97b08851 1370 size_t i;
markrad 58:f50b97b08851 1371
markrad 58:f50b97b08851 1372 for (i = 0; i < payload_count; i++)
markrad 58:f50b97b08851 1373 {
markrad 58:f50b97b08851 1374 if ((payloads[i].length > UINT32_MAX) ||
markrad 58:f50b97b08851 1375 (payload_size + payloads[i].length < payload_size))
markrad 58:f50b97b08851 1376 {
markrad 58:f50b97b08851 1377 break;
markrad 58:f50b97b08851 1378 }
markrad 58:f50b97b08851 1379
markrad 58:f50b97b08851 1380 payload_size += payloads[i].length;
markrad 58:f50b97b08851 1381 }
markrad 58:f50b97b08851 1382
markrad 58:f50b97b08851 1383 if ((i < payload_count) ||
markrad 58:f50b97b08851 1384 (payload_size > UINT32_MAX))
markrad 58:f50b97b08851 1385 {
markrad 58:f50b97b08851 1386 result = SESSION_SEND_TRANSFER_ERROR;
markrad 58:f50b97b08851 1387 }
markrad 58:f50b97b08851 1388 else
markrad 58:f50b97b08851 1389 {
markrad 58:f50b97b08851 1390 if (session_instance->remote_incoming_window == 0)
markrad 58:f50b97b08851 1391 {
markrad 58:f50b97b08851 1392 result = SESSION_SEND_TRANSFER_BUSY;
markrad 58:f50b97b08851 1393 }
markrad 58:f50b97b08851 1394 else
markrad 58:f50b97b08851 1395 {
markrad 58:f50b97b08851 1396 /* Codes_SRS_SESSION_01_012: [The session endpoint assigns each outgoing transfer frame an implicit transfer-id from a session scoped sequence.] */
markrad 58:f50b97b08851 1397 /* Codes_SRS_SESSION_01_027: [sending a transfer Upon sending a transfer, the sending endpoint will increment its next-outgoing-id] */
markrad 58:f50b97b08851 1398 *delivery_id = session_instance->next_outgoing_id;
markrad 58:f50b97b08851 1399 if ((transfer_set_handle(transfer, link_endpoint_instance->output_handle) != 0) ||
markrad 58:f50b97b08851 1400 (transfer_set_delivery_id(transfer, *delivery_id) != 0) ||
markrad 58:f50b97b08851 1401 (transfer_set_more(transfer, false) != 0))
markrad 58:f50b97b08851 1402 {
markrad 58:f50b97b08851 1403 /* Codes_SRS_SESSION_01_058: [When any other error occurs, session_send_transfer shall fail and return a non-zero value.] */
markrad 58:f50b97b08851 1404 result = SESSION_SEND_TRANSFER_ERROR;
markrad 58:f50b97b08851 1405 }
markrad 58:f50b97b08851 1406 else
markrad 58:f50b97b08851 1407 {
markrad 58:f50b97b08851 1408 AMQP_VALUE transfer_value;
markrad 58:f50b97b08851 1409
markrad 58:f50b97b08851 1410 transfer_value = amqpvalue_create_transfer(transfer);
markrad 58:f50b97b08851 1411 if (transfer_value == NULL)
markrad 58:f50b97b08851 1412 {
markrad 58:f50b97b08851 1413 /* Codes_SRS_SESSION_01_058: [When any other error occurs, session_send_transfer shall fail and return a non-zero value.] */
markrad 58:f50b97b08851 1414 result = SESSION_SEND_TRANSFER_ERROR;
markrad 58:f50b97b08851 1415 }
markrad 58:f50b97b08851 1416 else
markrad 58:f50b97b08851 1417 {
markrad 58:f50b97b08851 1418 uint32_t available_frame_size;
markrad 58:f50b97b08851 1419 size_t encoded_size;
markrad 58:f50b97b08851 1420
markrad 58:f50b97b08851 1421 if ((connection_get_remote_max_frame_size(session_instance->connection, &available_frame_size) != 0) ||
markrad 58:f50b97b08851 1422 (amqpvalue_get_encoded_size(transfer_value, &encoded_size) != 0))
markrad 58:f50b97b08851 1423 {
markrad 58:f50b97b08851 1424 result = SESSION_SEND_TRANSFER_ERROR;
markrad 58:f50b97b08851 1425 }
markrad 58:f50b97b08851 1426 else
markrad 58:f50b97b08851 1427 {
markrad 58:f50b97b08851 1428 payload_size = 0;
markrad 58:f50b97b08851 1429
markrad 58:f50b97b08851 1430 for (i = 0; i < payload_count; i++)
markrad 58:f50b97b08851 1431 {
markrad 58:f50b97b08851 1432 payload_size += payloads[i].length;
markrad 58:f50b97b08851 1433 }
markrad 58:f50b97b08851 1434
markrad 58:f50b97b08851 1435 available_frame_size -= (uint32_t)encoded_size;
markrad 58:f50b97b08851 1436 available_frame_size -= 8;
markrad 58:f50b97b08851 1437
markrad 58:f50b97b08851 1438 if (available_frame_size >= payload_size)
markrad 58:f50b97b08851 1439 {
markrad 58:f50b97b08851 1440 /* Codes_SRS_SESSION_01_055: [The encoding of the frame shall be done by calling connection_encode_frame and passing as arguments: the connection handle associated with the session, the transfer performative and the payload chunks passed to session_send_transfer.] */
markrad 58:f50b97b08851 1441 if (connection_encode_frame(session_instance->endpoint, transfer_value, payloads, payload_count, on_send_complete, callback_context) != 0)
markrad 58:f50b97b08851 1442 {
markrad 58:f50b97b08851 1443 /* Codes_SRS_SESSION_01_056: [If connection_encode_frame fails then session_send_transfer shall fail and return a non-zero value.] */
markrad 58:f50b97b08851 1444 result = SESSION_SEND_TRANSFER_ERROR;
markrad 58:f50b97b08851 1445 }
markrad 58:f50b97b08851 1446 else
markrad 58:f50b97b08851 1447 {
markrad 58:f50b97b08851 1448 /* Codes_SRS_SESSION_01_018: [is incremented after each successive transfer according to RFC-1982 [RFC1982] serial number arithmetic.] */
markrad 58:f50b97b08851 1449 session_instance->next_outgoing_id++;
markrad 58:f50b97b08851 1450 session_instance->remote_incoming_window--;
markrad 58:f50b97b08851 1451 session_instance->outgoing_window--;
markrad 58:f50b97b08851 1452
markrad 58:f50b97b08851 1453 /* Codes_SRS_SESSION_01_053: [On success, session_send_transfer shall return 0.] */
markrad 58:f50b97b08851 1454 result = SESSION_SEND_TRANSFER_OK;
markrad 58:f50b97b08851 1455 }
markrad 58:f50b97b08851 1456 }
markrad 58:f50b97b08851 1457 else
markrad 58:f50b97b08851 1458 {
markrad 58:f50b97b08851 1459 size_t current_payload_index = 0;
markrad 58:f50b97b08851 1460 uint32_t current_payload_pos = 0;
markrad 58:f50b97b08851 1461
markrad 58:f50b97b08851 1462 /* break it down into different deliveries */
markrad 58:f50b97b08851 1463 while (payload_size > 0)
markrad 58:f50b97b08851 1464 {
markrad 58:f50b97b08851 1465 uint32_t transfer_frame_payload_count = 0;
markrad 58:f50b97b08851 1466 uint32_t current_transfer_frame_payload_size = (uint32_t)payload_size;
markrad 58:f50b97b08851 1467 uint32_t byte_counter;
markrad 58:f50b97b08851 1468 size_t temp_current_payload_index = current_payload_index;
markrad 58:f50b97b08851 1469 uint32_t temp_current_payload_pos = current_payload_pos;
markrad 58:f50b97b08851 1470 AMQP_VALUE multi_transfer_amqp_value;
markrad 58:f50b97b08851 1471 bool more;
markrad 58:f50b97b08851 1472
markrad 58:f50b97b08851 1473 if (current_transfer_frame_payload_size > available_frame_size)
markrad 58:f50b97b08851 1474 {
markrad 58:f50b97b08851 1475 current_transfer_frame_payload_size = available_frame_size;
markrad 58:f50b97b08851 1476 }
markrad 58:f50b97b08851 1477
markrad 58:f50b97b08851 1478 if (available_frame_size >= payload_size)
markrad 58:f50b97b08851 1479 {
markrad 58:f50b97b08851 1480 more = false;
markrad 58:f50b97b08851 1481 }
markrad 58:f50b97b08851 1482 else
markrad 58:f50b97b08851 1483 {
markrad 58:f50b97b08851 1484 more = true;
markrad 58:f50b97b08851 1485 }
markrad 58:f50b97b08851 1486
markrad 58:f50b97b08851 1487 if (transfer_set_more(transfer, more) != 0)
markrad 58:f50b97b08851 1488 {
markrad 58:f50b97b08851 1489 break;
markrad 58:f50b97b08851 1490 }
markrad 58:f50b97b08851 1491
markrad 58:f50b97b08851 1492 multi_transfer_amqp_value = amqpvalue_create_transfer(transfer);
markrad 58:f50b97b08851 1493 if (multi_transfer_amqp_value == NULL)
markrad 58:f50b97b08851 1494 {
markrad 58:f50b97b08851 1495 break;
markrad 58:f50b97b08851 1496 }
markrad 58:f50b97b08851 1497
markrad 58:f50b97b08851 1498 byte_counter = current_transfer_frame_payload_size;
markrad 58:f50b97b08851 1499 while (byte_counter > 0)
markrad 58:f50b97b08851 1500 {
markrad 58:f50b97b08851 1501 if (payloads[temp_current_payload_index].length - temp_current_payload_pos >= byte_counter)
markrad 58:f50b97b08851 1502 {
markrad 58:f50b97b08851 1503 /* more data than we need */
markrad 58:f50b97b08851 1504 temp_current_payload_pos += byte_counter;
markrad 58:f50b97b08851 1505 byte_counter = 0;
markrad 58:f50b97b08851 1506 }
markrad 58:f50b97b08851 1507 else
markrad 58:f50b97b08851 1508 {
markrad 58:f50b97b08851 1509 byte_counter -= (uint32_t)payloads[temp_current_payload_index].length - temp_current_payload_pos;
markrad 58:f50b97b08851 1510 temp_current_payload_index++;
markrad 58:f50b97b08851 1511 temp_current_payload_pos = 0;
markrad 58:f50b97b08851 1512 }
markrad 58:f50b97b08851 1513 }
markrad 58:f50b97b08851 1514
markrad 58:f50b97b08851 1515 transfer_frame_payload_count = (uint32_t)(temp_current_payload_index - current_payload_index + 1);
markrad 58:f50b97b08851 1516 PAYLOAD* transfer_frame_payloads = (PAYLOAD*)amqpalloc_malloc(transfer_frame_payload_count * sizeof(PAYLOAD));
markrad 58:f50b97b08851 1517 if (transfer_frame_payloads == NULL)
markrad 58:f50b97b08851 1518 {
markrad 58:f50b97b08851 1519 amqpvalue_destroy(multi_transfer_amqp_value);
markrad 58:f50b97b08851 1520 break;
markrad 58:f50b97b08851 1521 }
markrad 58:f50b97b08851 1522
markrad 58:f50b97b08851 1523 /* copy data */
markrad 58:f50b97b08851 1524 byte_counter = current_transfer_frame_payload_size;
markrad 58:f50b97b08851 1525 transfer_frame_payload_count = 0;
markrad 58:f50b97b08851 1526
markrad 58:f50b97b08851 1527 while (byte_counter > 0)
markrad 58:f50b97b08851 1528 {
markrad 58:f50b97b08851 1529 if (payloads[current_payload_index].length - current_payload_pos > byte_counter)
markrad 58:f50b97b08851 1530 {
markrad 58:f50b97b08851 1531 /* more data than we need */
markrad 58:f50b97b08851 1532 transfer_frame_payloads[transfer_frame_payload_count].bytes = payloads[current_payload_index].bytes + current_payload_pos;
markrad 58:f50b97b08851 1533 transfer_frame_payloads[transfer_frame_payload_count].length = byte_counter;
markrad 58:f50b97b08851 1534 current_payload_pos += byte_counter;
markrad 58:f50b97b08851 1535 byte_counter = 0;
markrad 58:f50b97b08851 1536 }
markrad 58:f50b97b08851 1537 else
markrad 58:f50b97b08851 1538 {
markrad 58:f50b97b08851 1539 /* copy entire payload and move to the next */
markrad 58:f50b97b08851 1540 transfer_frame_payloads[transfer_frame_payload_count].bytes = payloads[current_payload_index].bytes + current_payload_pos;
markrad 58:f50b97b08851 1541 transfer_frame_payloads[transfer_frame_payload_count].length = payloads[current_payload_index].length - current_payload_pos;
markrad 58:f50b97b08851 1542 byte_counter -= (uint32_t)payloads[current_payload_index].length - current_payload_pos;
markrad 58:f50b97b08851 1543 current_payload_index++;
markrad 58:f50b97b08851 1544 current_payload_pos = 0;
markrad 58:f50b97b08851 1545 }
markrad 58:f50b97b08851 1546
markrad 58:f50b97b08851 1547 transfer_frame_payload_count++;
markrad 58:f50b97b08851 1548 }
markrad 58:f50b97b08851 1549
markrad 58:f50b97b08851 1550 if (connection_encode_frame(session_instance->endpoint, multi_transfer_amqp_value, transfer_frame_payloads, transfer_frame_payload_count, on_send_complete, callback_context) != 0)
markrad 58:f50b97b08851 1551 {
markrad 58:f50b97b08851 1552 amqpalloc_free(transfer_frame_payloads);
markrad 58:f50b97b08851 1553 amqpvalue_destroy(multi_transfer_amqp_value);
markrad 58:f50b97b08851 1554 break;
markrad 58:f50b97b08851 1555 }
markrad 58:f50b97b08851 1556
markrad 58:f50b97b08851 1557 amqpalloc_free(transfer_frame_payloads);
markrad 58:f50b97b08851 1558 amqpvalue_destroy(multi_transfer_amqp_value);
markrad 58:f50b97b08851 1559 payload_size -= current_transfer_frame_payload_size;
markrad 58:f50b97b08851 1560 }
markrad 58:f50b97b08851 1561
markrad 58:f50b97b08851 1562 if (payload_size > 0)
markrad 58:f50b97b08851 1563 {
markrad 58:f50b97b08851 1564 result = SESSION_SEND_TRANSFER_ERROR;
markrad 58:f50b97b08851 1565 }
markrad 58:f50b97b08851 1566 else
markrad 58:f50b97b08851 1567 {
markrad 58:f50b97b08851 1568 /* Codes_SRS_SESSION_01_018: [is incremented after each successive transfer according to RFC-1982 [RFC1982] serial number arithmetic.] */
markrad 58:f50b97b08851 1569 session_instance->next_outgoing_id++;
markrad 58:f50b97b08851 1570 session_instance->remote_incoming_window--;
markrad 58:f50b97b08851 1571 session_instance->outgoing_window--;
markrad 58:f50b97b08851 1572
markrad 58:f50b97b08851 1573 result = SESSION_SEND_TRANSFER_OK;
markrad 58:f50b97b08851 1574 }
markrad 58:f50b97b08851 1575 }
markrad 58:f50b97b08851 1576 }
markrad 58:f50b97b08851 1577
markrad 58:f50b97b08851 1578 amqpvalue_destroy(transfer_value);
markrad 58:f50b97b08851 1579 }
markrad 58:f50b97b08851 1580 }
markrad 58:f50b97b08851 1581 }
markrad 58:f50b97b08851 1582 }
markrad 58:f50b97b08851 1583 }
markrad 58:f50b97b08851 1584 }
markrad 58:f50b97b08851 1585
markrad 58:f50b97b08851 1586 return result;
markrad 58:f50b97b08851 1587 }