Azure IoT
A small memory footprint AMQP implimentation
Dependents: iothub_client_sample_amqp remote_monitoring simplesample_amqp
Diff: header_detect_io.c
- Revision:
- 28:add19eb7defa
- Parent:
- 27:d74f1cea23e1
- Child:
- 29:4a11413cf217
--- a/header_detect_io.c Fri Jun 02 15:53:07 2017 -0700
+++ b/header_detect_io.c Fri Jun 30 10:41:22 2017 -0700
@@ -9,25 +9,25 @@
typedef enum IO_STATE_TAG
{
- IO_STATE_NOT_OPEN,
- IO_STATE_OPENING_UNDERLYING_IO,
- IO_STATE_WAIT_FOR_HEADER,
- IO_STATE_OPEN,
- IO_STATE_CLOSING,
- IO_STATE_ERROR
+ IO_STATE_NOT_OPEN,
+ IO_STATE_OPENING_UNDERLYING_IO,
+ IO_STATE_WAIT_FOR_HEADER,
+ IO_STATE_OPEN,
+ IO_STATE_CLOSING,
+ IO_STATE_ERROR
} IO_STATE;
typedef struct HEADER_DETECT_IO_INSTANCE_TAG
{
- XIO_HANDLE underlying_io;
- size_t header_pos;
- IO_STATE io_state;
- ON_IO_OPEN_COMPLETE on_io_open_complete;
- ON_IO_CLOSE_COMPLETE on_io_close_complete;
- ON_IO_ERROR on_io_error;
- ON_BYTES_RECEIVED on_bytes_received;
- void* on_io_open_complete_context;
- void* on_io_close_complete_context;
+ XIO_HANDLE underlying_io;
+ size_t header_pos;
+ IO_STATE io_state;
+ ON_IO_OPEN_COMPLETE on_io_open_complete;
+ ON_IO_CLOSE_COMPLETE on_io_close_complete;
+ ON_IO_ERROR on_io_error;
+ ON_BYTES_RECEIVED on_bytes_received;
+ void* on_io_open_complete_context;
+ void* on_io_close_complete_context;
void* on_io_error_context;
void* on_bytes_received_context;
} HEADER_DETECT_IO_INSTANCE;
@@ -36,18 +36,18 @@
static void indicate_error(HEADER_DETECT_IO_INSTANCE* header_detect_io_instance)
{
- if (header_detect_io_instance->on_io_error != NULL)
- {
- header_detect_io_instance->on_io_error(header_detect_io_instance->on_io_error_context);
- }
+ if (header_detect_io_instance->on_io_error != NULL)
+ {
+ header_detect_io_instance->on_io_error(header_detect_io_instance->on_io_error_context);
+ }
}
static void indicate_open_complete(HEADER_DETECT_IO_INSTANCE* header_detect_io_instance, IO_OPEN_RESULT open_result)
{
- if (header_detect_io_instance->on_io_open_complete != NULL)
- {
- header_detect_io_instance->on_io_open_complete(header_detect_io_instance->on_io_open_complete_context, open_result);
- }
+ if (header_detect_io_instance->on_io_open_complete != NULL)
+ {
+ header_detect_io_instance->on_io_open_complete(header_detect_io_instance->on_io_open_complete_context, open_result);
+ }
}
static void indicate_close_complete(HEADER_DETECT_IO_INSTANCE* header_detect_io_instance)
@@ -67,189 +67,189 @@
static void on_underlying_io_bytes_received(void* context, const unsigned char* buffer, size_t size)
{
- HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)context;
+ HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)context;
- while (size > 0)
- {
- switch (header_detect_io_instance->io_state)
- {
- default:
- break;
+ while (size > 0)
+ {
+ switch (header_detect_io_instance->io_state)
+ {
+ default:
+ break;
- case IO_STATE_WAIT_FOR_HEADER:
- if (amqp_header[header_detect_io_instance->header_pos] != buffer[0])
- {
+ case IO_STATE_WAIT_FOR_HEADER:
+ if (amqp_header[header_detect_io_instance->header_pos] != buffer[0])
+ {
/* Send expected header, then close as per spec. We do not care if we fail */
(void)xio_send(header_detect_io_instance->underlying_io, amqp_header, sizeof(amqp_header), on_send_complete_close, context);
header_detect_io_instance->io_state = IO_STATE_NOT_OPEN;
- indicate_open_complete(header_detect_io_instance, IO_OPEN_ERROR);
- size = 0;
- }
- else
- {
- header_detect_io_instance->header_pos++;
- size--;
- buffer++;
- if (header_detect_io_instance->header_pos == sizeof(amqp_header))
- {
- if (xio_send(header_detect_io_instance->underlying_io, amqp_header, sizeof(amqp_header), NULL, NULL) != 0)
- {
- header_detect_io_instance->io_state = IO_STATE_NOT_OPEN;
- indicate_open_complete(header_detect_io_instance, IO_OPEN_ERROR);
- }
- else
- {
- header_detect_io_instance->io_state = IO_STATE_OPEN;
- indicate_open_complete(header_detect_io_instance, IO_OPEN_OK);
- }
- }
- }
- break;
+ indicate_open_complete(header_detect_io_instance, IO_OPEN_ERROR);
+ size = 0;
+ }
+ else
+ {
+ header_detect_io_instance->header_pos++;
+ size--;
+ buffer++;
+ if (header_detect_io_instance->header_pos == sizeof(amqp_header))
+ {
+ if (xio_send(header_detect_io_instance->underlying_io, amqp_header, sizeof(amqp_header), NULL, NULL) != 0)
+ {
+ header_detect_io_instance->io_state = IO_STATE_NOT_OPEN;
+ indicate_open_complete(header_detect_io_instance, IO_OPEN_ERROR);
+ }
+ else
+ {
+ header_detect_io_instance->io_state = IO_STATE_OPEN;
+ indicate_open_complete(header_detect_io_instance, IO_OPEN_OK);
+ }
+ }
+ }
+ break;
- case IO_STATE_OPEN:
- header_detect_io_instance->on_bytes_received(header_detect_io_instance->on_bytes_received_context, buffer, size);
- size = 0;
- break;
- }
- }
+ case IO_STATE_OPEN:
+ header_detect_io_instance->on_bytes_received(header_detect_io_instance->on_bytes_received_context, buffer, size);
+ size = 0;
+ break;
+ }
+ }
}
static void on_underlying_io_close_complete(void* context)
{
- HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)context;
+ HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)context;
- switch (header_detect_io_instance->io_state)
- {
- default:
- break;
+ switch (header_detect_io_instance->io_state)
+ {
+ default:
+ break;
case IO_STATE_CLOSING:
header_detect_io_instance->io_state = IO_STATE_NOT_OPEN;
indicate_close_complete(header_detect_io_instance);
break;
- case IO_STATE_WAIT_FOR_HEADER:
- case IO_STATE_OPENING_UNDERLYING_IO:
- header_detect_io_instance->io_state = IO_STATE_NOT_OPEN;
- indicate_open_complete(header_detect_io_instance, IO_OPEN_ERROR);
- break;
- }
+ case IO_STATE_WAIT_FOR_HEADER:
+ case IO_STATE_OPENING_UNDERLYING_IO:
+ header_detect_io_instance->io_state = IO_STATE_NOT_OPEN;
+ indicate_open_complete(header_detect_io_instance, IO_OPEN_ERROR);
+ break;
+ }
}
static void on_underlying_io_open_complete(void* context, IO_OPEN_RESULT open_result)
{
- HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)context;
+ HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)context;
- switch (header_detect_io_instance->io_state)
- {
- default:
- break;
+ switch (header_detect_io_instance->io_state)
+ {
+ default:
+ break;
- case IO_STATE_OPENING_UNDERLYING_IO:
- if (open_result == IO_OPEN_OK)
- {
- header_detect_io_instance->io_state = IO_STATE_WAIT_FOR_HEADER;
- }
- else
- {
- if (xio_close(header_detect_io_instance->underlying_io, on_underlying_io_close_complete, header_detect_io_instance) != 0)
- {
- header_detect_io_instance->io_state = IO_STATE_NOT_OPEN;
- indicate_open_complete(header_detect_io_instance, IO_OPEN_ERROR);
- }
- }
+ case IO_STATE_OPENING_UNDERLYING_IO:
+ if (open_result == IO_OPEN_OK)
+ {
+ header_detect_io_instance->io_state = IO_STATE_WAIT_FOR_HEADER;
+ }
+ else
+ {
+ if (xio_close(header_detect_io_instance->underlying_io, on_underlying_io_close_complete, header_detect_io_instance) != 0)
+ {
+ header_detect_io_instance->io_state = IO_STATE_NOT_OPEN;
+ indicate_open_complete(header_detect_io_instance, IO_OPEN_ERROR);
+ }
+ }
- break;
- }
+ break;
+ }
}
static void on_underlying_io_error(void* context)
{
- HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)context;
+ HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)context;
- switch (header_detect_io_instance->io_state)
- {
- default:
- break;
+ switch (header_detect_io_instance->io_state)
+ {
+ default:
+ break;
- case IO_STATE_WAIT_FOR_HEADER:
- case IO_STATE_OPENING_UNDERLYING_IO:
- header_detect_io_instance->io_state = IO_STATE_NOT_OPEN;
- indicate_open_complete(header_detect_io_instance, IO_OPEN_ERROR);
- break;
+ case IO_STATE_WAIT_FOR_HEADER:
+ case IO_STATE_OPENING_UNDERLYING_IO:
+ header_detect_io_instance->io_state = IO_STATE_NOT_OPEN;
+ indicate_open_complete(header_detect_io_instance, IO_OPEN_ERROR);
+ break;
- case IO_STATE_OPEN:
- header_detect_io_instance->io_state = IO_STATE_ERROR;
- indicate_error(header_detect_io_instance);
- break;
- }
+ case IO_STATE_OPEN:
+ header_detect_io_instance->io_state = IO_STATE_ERROR;
+ indicate_error(header_detect_io_instance);
+ break;
+ }
}
CONCRETE_IO_HANDLE headerdetectio_create(void* io_create_parameters)
{
- HEADER_DETECT_IO_INSTANCE* result;
+ HEADER_DETECT_IO_INSTANCE* result;
- if (io_create_parameters == NULL)
- {
- result = NULL;
- }
- else
- {
- HEADERDETECTIO_CONFIG* header_detect_io_config = (HEADERDETECTIO_CONFIG*)io_create_parameters;
- result = (HEADER_DETECT_IO_INSTANCE*)malloc(sizeof(HEADER_DETECT_IO_INSTANCE));
- if (result != NULL)
- {
- result->underlying_io = header_detect_io_config->underlying_io;
- result->on_io_open_complete = NULL;
+ if (io_create_parameters == NULL)
+ {
+ result = NULL;
+ }
+ else
+ {
+ HEADERDETECTIO_CONFIG* header_detect_io_config = (HEADERDETECTIO_CONFIG*)io_create_parameters;
+ result = (HEADER_DETECT_IO_INSTANCE*)malloc(sizeof(HEADER_DETECT_IO_INSTANCE));
+ if (result != NULL)
+ {
+ result->underlying_io = header_detect_io_config->underlying_io;
+ result->on_io_open_complete = NULL;
result->on_io_close_complete = NULL;
- result->on_io_error = NULL;
- result->on_bytes_received = NULL;
- result->on_io_open_complete_context = NULL;
+ result->on_io_error = NULL;
+ result->on_bytes_received = NULL;
+ result->on_io_open_complete_context = NULL;
result->on_io_close_complete_context = NULL;
result->on_io_error_context = NULL;
result->on_bytes_received_context = NULL;
- result->io_state = IO_STATE_NOT_OPEN;
- }
- }
+ result->io_state = IO_STATE_NOT_OPEN;
+ }
+ }
- return result;
+ return result;
}
void headerdetectio_destroy(CONCRETE_IO_HANDLE header_detect_io)
{
- if (header_detect_io != NULL)
- {
- HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)header_detect_io;
- (void)headerdetectio_close(header_detect_io, NULL, NULL);
- xio_destroy(header_detect_io_instance->underlying_io);
- free(header_detect_io);
- }
+ if (header_detect_io != NULL)
+ {
+ HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)header_detect_io;
+ (void)headerdetectio_close(header_detect_io, NULL, NULL);
+ xio_destroy(header_detect_io_instance->underlying_io);
+ free(header_detect_io);
+ }
}
int headerdetectio_open(CONCRETE_IO_HANDLE header_detect_io, ON_IO_OPEN_COMPLETE on_io_open_complete, void* on_io_open_complete_context, ON_BYTES_RECEIVED on_bytes_received, void* on_bytes_received_context, ON_IO_ERROR on_io_error, void* on_io_error_context)
{
- int result;
+ int result;
- if (header_detect_io == NULL)
- {
- result = __FAILURE__;
- }
- else
- {
- HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)header_detect_io;
+ if (header_detect_io == NULL)
+ {
+ result = __FAILURE__;
+ }
+ else
+ {
+ HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)header_detect_io;
if (header_detect_io_instance->io_state != IO_STATE_NOT_OPEN &&
header_detect_io_instance->io_state != IO_STATE_OPEN)
{
result = __FAILURE__;
}
- else
- {
- header_detect_io_instance->on_bytes_received = on_bytes_received;
- header_detect_io_instance->on_io_open_complete = on_io_open_complete;
- header_detect_io_instance->on_io_error = on_io_error;
+ else
+ {
+ header_detect_io_instance->on_bytes_received = on_bytes_received;
+ header_detect_io_instance->on_io_open_complete = on_io_open_complete;
+ header_detect_io_instance->on_io_error = on_io_error;
header_detect_io_instance->on_bytes_received_context = on_bytes_received_context;
header_detect_io_instance->on_io_open_complete_context = on_io_open_complete_context;
header_detect_io_instance->on_io_error_context = on_io_error_context;
@@ -273,93 +273,93 @@
result = 0;
}
}
- }
- }
+ }
+ }
- return result;
+ return result;
}
int headerdetectio_close(CONCRETE_IO_HANDLE header_detect_io, ON_IO_CLOSE_COMPLETE on_io_close_complete, void* callback_context)
{
- int result;
+ int result;
- if (header_detect_io == NULL)
- {
- result = __FAILURE__;
- }
- else
- {
- HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)header_detect_io;
+ if (header_detect_io == NULL)
+ {
+ result = __FAILURE__;
+ }
+ else
+ {
+ HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)header_detect_io;
- if ((header_detect_io_instance->io_state == IO_STATE_NOT_OPEN) ||
- (header_detect_io_instance->io_state == IO_STATE_CLOSING))
- {
- result = __FAILURE__;
- }
- else
- {
- header_detect_io_instance->io_state = IO_STATE_CLOSING;
+ if ((header_detect_io_instance->io_state == IO_STATE_NOT_OPEN) ||
+ (header_detect_io_instance->io_state == IO_STATE_CLOSING))
+ {
+ result = __FAILURE__;
+ }
+ else
+ {
+ header_detect_io_instance->io_state = IO_STATE_CLOSING;
header_detect_io_instance->on_io_close_complete = on_io_close_complete;
header_detect_io_instance->on_io_close_complete_context = callback_context;
- if (xio_close(header_detect_io_instance->underlying_io, on_underlying_io_close_complete, header_detect_io_instance) != 0)
- {
- result = __FAILURE__;
- }
- else
- {
- result = 0;
- }
- }
- }
+ if (xio_close(header_detect_io_instance->underlying_io, on_underlying_io_close_complete, header_detect_io_instance) != 0)
+ {
+ result = __FAILURE__;
+ }
+ else
+ {
+ result = 0;
+ }
+ }
+ }
- return result;
+ return result;
}
int headerdetectio_send(CONCRETE_IO_HANDLE header_detect_io, const void* buffer, size_t size, ON_SEND_COMPLETE on_send_complete, void* callback_context)
{
- int result;
+ int result;
- if (header_detect_io == NULL)
- {
- result = __FAILURE__;
- }
- else
- {
- HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)header_detect_io;
+ if (header_detect_io == NULL)
+ {
+ result = __FAILURE__;
+ }
+ else
+ {
+ HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)header_detect_io;
- if (header_detect_io_instance->io_state != IO_STATE_OPEN)
- {
- result = __FAILURE__;
- }
- else
- {
- if (xio_send(header_detect_io_instance->underlying_io, buffer, size, on_send_complete, callback_context) != 0)
- {
- result = __FAILURE__;
- }
- else
- {
- result = 0;
- }
- }
- }
+ if (header_detect_io_instance->io_state != IO_STATE_OPEN)
+ {
+ result = __FAILURE__;
+ }
+ else
+ {
+ if (xio_send(header_detect_io_instance->underlying_io, buffer, size, on_send_complete, callback_context) != 0)
+ {
+ result = __FAILURE__;
+ }
+ else
+ {
+ result = 0;
+ }
+ }
+ }
- return result;
+ return result;
}
void headerdetectio_dowork(CONCRETE_IO_HANDLE header_detect_io)
{
- if (header_detect_io != NULL)
- {
- HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)header_detect_io;
+ if (header_detect_io != NULL)
+ {
+ HEADER_DETECT_IO_INSTANCE* header_detect_io_instance = (HEADER_DETECT_IO_INSTANCE*)header_detect_io;
- if ((header_detect_io_instance->io_state != IO_STATE_NOT_OPEN) &&
- (header_detect_io_instance->io_state != IO_STATE_ERROR))
- {
- xio_dowork(header_detect_io_instance->underlying_io);
- }
- }
+ if ((header_detect_io_instance->io_state != IO_STATE_NOT_OPEN) &&
+ (header_detect_io_instance->io_state != IO_STATE_ERROR))
+ {
+ xio_dowork(header_detect_io_instance->underlying_io);
+ }
+ }
}
int headerdetectio_setoption(CONCRETE_IO_HANDLE header_detect_io, const char* optionName, const void* value)
@@ -422,16 +422,16 @@
static const IO_INTERFACE_DESCRIPTION header_detect_io_interface_description =
{
headerdetectio_retrieveoptions,
- headerdetectio_create,
- headerdetectio_destroy,
- headerdetectio_open,
- headerdetectio_close,
- headerdetectio_send,
- headerdetectio_dowork,
+ headerdetectio_create,
+ headerdetectio_destroy,
+ headerdetectio_open,
+ headerdetectio_close,
+ headerdetectio_send,
+ headerdetectio_dowork,
headerdetectio_setoption
};
const IO_INTERFACE_DESCRIPTION* headerdetectio_get_interface_description(void)
{
- return &header_detect_io_interface_description;
+ return &header_detect_io_interface_description;
}
