Diff: private/connector_sm_recv.h

Revision:

0:1c358ea10753

diff -r 000000000000 -r 1c358ea10753 private/connector_sm_recv.h
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/private/connector_sm_recv.h	Tue Dec 03 13:34:02 2013 +0000
@@ -0,0 +1,784 @@
+/*
+ * Copyright (c) 2013 Digi International Inc.,
+ * All rights not expressly granted are reserved.
+ *
+ * This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this file,
+ * You can obtain one at http://mozilla.org/MPL/2.0/.
+ *
+ * Digi International Inc. 11001 Bren Road East, Minnetonka, MN 55343
+ * =======================================================================
+ */
+#if (defined CONNECTOR_TRANSPORT_SMS)
+static connector_status_t sm_decode_segment(connector_data_t * const connector_ptr, connector_sm_packet_t * const recv_ptr)
+{
+    size_t const data_size = 1 + ((recv_ptr->total_bytes - recv_ptr->processed_bytes) * 4)/5;
+    void * data_ptr = NULL;
+    connector_status_t result = malloc_data_buffer(connector_ptr, data_size, named_buffer_id(sm_data_block), &data_ptr);
+
+    if (result == connector_working)
+    {
+        recv_ptr->total_bytes = sm_decode85(data_ptr, data_size, recv_ptr->data + recv_ptr->processed_bytes, recv_ptr->total_bytes - recv_ptr->processed_bytes);
+        memcpy(recv_ptr->data, data_ptr, recv_ptr->total_bytes);
+        result = free_data_buffer(connector_ptr, named_buffer_id(sm_data_block), data_ptr);
+    }
+
+    return result;
+}
+
+static connector_status_t sm_verify_sms_preamble(connector_sm_data_t * const sm_ptr)
+{
+    connector_status_t result = connector_working;
+
+    if (sm_ptr->transport.id_length > 0)
+    {
+        uint8_t * const data_ptr = sm_ptr->network.recv_packet.data;
+        char const prefix = '(';
+        char const suffix[] = "):";
+        size_t const suffix_bytes = sizeof suffix - 1;
+        size_t const prefix_bytes = sizeof prefix;
+        size_t const suffix_position = prefix_bytes + sm_ptr->transport.id_length;
+        connector_bool_t const valid_prefix = connector_bool(*data_ptr == prefix);
+        connector_bool_t const valid_shared_key = connector_bool(memcmp(data_ptr + prefix_bytes, sm_ptr->transport.id, sm_ptr->transport.id_length) == 0);
+        connector_bool_t const valid_suffix = connector_bool(memcmp(data_ptr + suffix_position, suffix, suffix_bytes) == 0);
+
+        if (valid_prefix && valid_shared_key && valid_suffix)
+            sm_ptr->network.recv_packet.processed_bytes = suffix_position + suffix_bytes;
+        else
+        {
+            connector_debug_printf("sm_verify_sms_preamble: valid_prefix=%d, valid_shared_key=%d, valid_suffix=%d\n", valid_prefix, valid_shared_key, valid_suffix);
+            connector_debug_printf("%s\n",(char *)data_ptr);
+
+            result = connector_invalid_response;
+        }
+    }
+
+    return result;
+}
+#endif
+
+#if (defined CONNECTOR_TRANSPORT_UDP)
+static connector_status_t sm_verify_udp_header(connector_sm_data_t * const sm_ptr)
+{
+    connector_status_t result = connector_invalid_response;
+    connector_sm_packet_t * const recv_ptr = &sm_ptr->network.recv_packet;
+    uint8_t * data_ptr = recv_ptr->data;
+    uint8_t const version_and_type = *data_ptr++;
+    uint8_t const version = version_and_type >> 4;
+    connector_sm_id_type_t const type = (connector_sm_id_type_t)(version_and_type & 0x0F);
+
+    if (version != SM_UDP_VERSION)
+    {
+        connector_debug_printf("sm_verify_udp_header: invalid SM UDP version [%d]\n", version);
+        result = connector_abort;
+        goto error;
+    }
+
+    if (type != sm_ptr->transport.id_type)
+        goto done;
+
+    if (memcmp(data_ptr, sm_ptr->transport.id, sm_ptr->transport.id_length))
+        goto done; /* not for us */
+
+    recv_ptr->processed_bytes = sm_ptr->transport.id_length + 1;
+    result = connector_working;
+
+done:
+error:
+    return result;
+}
+#endif
+
+#if (defined CONNECTOR_SM_MULTIPART)
+static connector_status_t sm_multipart_allocate(connector_data_t * const connector_ptr, connector_sm_data_t * const sm_ptr, connector_sm_session_t * const session)
+{
+    connector_status_t status = connector_working;
+    size_t const max_payload_bytes = sm_get_max_payload_bytes(sm_ptr);
+    size_t const max_session_bytes = (((max_payload_bytes * session->segments.count) + 3)/4) * 4; /* make sure it is word aligned */
+    size_t const size_array_bytes = sizeof(*session->segments.size_array) * session->segments.count;
+
+    ASSERT_GOTO(session->in.data == NULL, error);
+    session->in.bytes = max_session_bytes + size_array_bytes;
+    status = sm_allocate_user_buffer(connector_ptr, &session->in);
+    ASSERT_GOTO(status == connector_working, error);
+    session->segments.size_array = (void *)(session->in.data + max_session_bytes); /* alignment issue is taken care where max_session_bytes is defined */
+    memset(session->segments.size_array, 0, size_array_bytes);
+
+error:
+    return status;
+}
+#endif
+
+static connector_status_t sm_more_data_callback(connector_data_t * const connector_ptr, connector_sm_data_t * const sm_ptr)
+{
+    connector_status_t result;
+    connector_request_id_t request_id;
+    connector_callback_status_t callback_status;
+    connector_sm_more_data_t cb_data;
+
+    cb_data.transport = sm_ptr->network.transport;
+    request_id.sm_request = connector_request_id_sm_more_data;
+    callback_status = connector_callback(connector_ptr->callback, connector_class_id_short_message, request_id, &cb_data);
+    result = sm_map_callback_status_to_connector_status(callback_status);
+
+    return result;
+}
+
+typedef struct
+{
+    uint32_t request_id;
+    size_t bytes;
+
+    connector_bool_t isRequest;
+    connector_bool_t isResponseNeeded;
+    connector_bool_t isPackCmd;
+    connector_bool_t isMultipart;
+    connector_bool_t isCompressed;
+    connector_bool_t isError;
+
+    connector_sm_cmd_t command;
+
+    enum
+    {
+        sm_single_segment,
+        sm_multipart_first_segment,
+        sm_multipart_subsequent_segment
+    } type;
+
+    struct
+    {
+        uint8_t count;
+        uint8_t number;
+    } segment;
+
+    uint16_t crc16;
+    uint8_t cmd_status;
+} sm_header_t;
+
+static connector_status_t sm_process_header(connector_sm_packet_t * const recv_ptr, sm_header_t * const header)
+{
+    connector_status_t result = connector_invalid_payload_packet;
+    uint8_t * data_ptr = &recv_ptr->data[recv_ptr->processed_bytes];
+    uint8_t * const segment = data_ptr;
+
+    header->segment.count = 1;
+    header->segment.number = 0;
+    header->bytes = 0;
+    header->command = connector_sm_cmd_opaque_response;
+    header->isError = connector_false;
+    header->isCompressed = connector_false;
+
+    {
+        uint8_t const info_field = message_load_u8(segment, info);
+        uint8_t const request_id_high_bits_mask = 0x03;
+
+        header->request_id = (info_field & request_id_high_bits_mask) << 8;
+        header->request_id |= message_load_u8(segment, request);
+
+        header->isMultipart = SmIsMultiPart(info_field);
+        header->isRequest = SmIsRequest(info_field);
+        header->isResponseNeeded = SmIsResponseNeeded(info_field);
+    }
+
+    if (header->isMultipart)
+    {
+        #if (defined CONNECTOR_SM_MULTIPART)
+        {
+            uint8_t * const segment0 = data_ptr;
+
+            header->segment.number = message_load_u8(segment0, segment);
+            if (header->segment.number > 0)
+            {
+                uint8_t * const segmentn = data_ptr;
+
+                header->cmd_status = 0;
+                header->type = sm_multipart_subsequent_segment;
+
+                if (header->isPackCmd)
+                {
+                    header->bytes = record_end(segmentn) - sizeof header->crc16;
+                }
+                else
+                {
+                    header->crc16 = message_load_be16(segmentn, crc);
+                    message_store_be16(segmentn, crc, 0);
+                    header->bytes = record_end(segmentn);
+                }
+            }
+            else
+            {
+                header->type = sm_multipart_first_segment;
+                header->segment.count = message_load_u8(segment0, count);
+                header->cmd_status = message_load_u8(segment0, cmd_status);
+
+                if (header->isPackCmd)
+                {
+                    header->bytes = record_end(segment0) - sizeof header->crc16;
+                }
+                else
+                {
+                    header->crc16 = message_load_be16(segment0, crc);
+                    message_store_be16(segment0, crc, 0);
+                    header->bytes = record_end(segment0);
+                }
+            }
+        }
+        #else
+        {
+            connector_debug_printf("Received multipart packet, but multipart is disabled\n");
+            connector_debug_printf("Review CONNECTOR_SM_MULTIPART and CONNECTOR_SM_MAX_RX_SEGMENTS defines\n");
+            goto error;
+        }
+        #endif
+    }
+    else
+    {
+        header->cmd_status = message_load_u8(segment, cmd_status);
+        if (header->isPackCmd)
+        {
+            header->bytes = record_end(segment) - sizeof header->crc16;
+        }
+        else
+        {
+            header->crc16 = message_load_be16(segment, crc);
+            message_store_be16(segment, crc, 0);
+            header->bytes = record_end(segment);
+        }
+    }
+
+    header->isCompressed = SmIsCompressed(header->cmd_status);
+    #if !(defined CONNECTOR_COMPRESSION)
+    if (header->isCompressed)
+    {
+        connector_debug_printf("sm_process_header: Received compressed packet, but compression is disabled!\n");
+        goto error;
+    }
+    #endif
+
+    if (header->isRequest)
+    {
+        header->command = (connector_sm_cmd_t)(header->cmd_status & 0x7F);
+        if (header->isPackCmd && (header->command == connector_sm_cmd_pack))
+        {
+            connector_debug_printf("sm_process_header: Pack command inside a pack command is not allowed!\n");
+            goto error;
+        }
+        header->isError = connector_false;
+    }
+    else
+    {
+        header->isError = SmIsError(header->cmd_status);
+        header->command = connector_sm_cmd_opaque_response;
+    }
+ 
+    if (!header->isPackCmd)
+    {
+        size_t const sm_bytes = recv_ptr->total_bytes - recv_ptr->processed_bytes;
+        uint16_t calculated_crc = 0;
+
+        calculated_crc = sm_calculate_crc16(calculated_crc, data_ptr, sm_bytes);
+        if(calculated_crc != header->crc16)
+        {
+            connector_debug_printf("sm_process_header: crc error!\n");
+            goto error;
+        }
+
+        if (header->isRequest)
+            header->isPackCmd = connector_bool(header->command == connector_sm_cmd_pack);
+    }
+
+    recv_ptr->processed_bytes += header->bytes;
+    result = connector_working;
+
+error:
+    return result;
+}
+
+static connector_status_t sm_update_session(connector_data_t * const connector_ptr, connector_sm_data_t * const sm_ptr,
+                                            sm_header_t * const header, size_t const payload_bytes)
+{
+    connector_status_t result = connector_invalid_payload_packet;
+    connector_sm_packet_t * const recv_ptr = &sm_ptr->network.recv_packet;
+    connector_bool_t const client_originated = connector_bool(!header->isRequest);
+    connector_sm_session_t * session = get_sm_session(sm_ptr, header->request_id, client_originated);
+
+    if (session == NULL)
+    {
+        session = sm_create_session(connector_ptr, sm_ptr, connector_false);
+        if (session == NULL)
+        {
+            result = connector_pending;
+            goto error;
+        }
+
+        session->request_id = header->request_id;
+        session->command = client_originated ? connector_sm_cmd_opaque_response : header->command;
+    }
+
+    if (header->segment.number == 0)
+    {
+        if (header->isCompressed) SmSetCompressed(session->flags);
+
+        if (header->isRequest)
+        {
+            if (header->isResponseNeeded) SmSetResponseNeeded(session->flags);
+            /* If the first segment of a multipart SMS that arrived was not segment0, previously stored session->command information is wrong as 
+             * for multipart messages, type is only included in the 0th segment. Update it here */
+            session->command = header->command;
+        }
+        else if (header->isError)
+        {
+            uint16_t const error_value = LoadBE16(&recv_ptr->data[recv_ptr->processed_bytes]);
+
+            switch (error_value)
+            {
+                case connector_sm_error_in_request:
+                    session->error = connector_sm_error_in_request;
+                    break;
+
+                case connector_sm_error_unavailable:
+                    session->error = connector_sm_error_unavailable;
+                    break;
+
+                default:
+                    session->error = connector_sm_error_unknown;
+                    break;
+            }
+
+            SmSetError(session->flags);
+            recv_ptr->processed_bytes += sizeof error_value;
+        }
+
+        session->segments.count = header->segment.count;
+    }
+
+    #if (defined CONNECTOR_SM_MULTIPART)
+    if (header->isMultipart)
+    {
+        SmSetMultiPart(session->flags);
+        if (session->in.data == NULL)
+        {
+            if (header->segment.number > 0)
+                session->segments.count = sm_ptr->session.max_segments;
+
+            result = sm_multipart_allocate(connector_ptr, sm_ptr, session);
+            ASSERT_GOTO(result == connector_working, error);
+            ASSERT_GOTO(session->in.data != NULL, error);
+        }
+
+        if (session->segments.size_array[header->segment.number] == 0)
+        {
+            size_t const max_payload_bytes = sm_get_max_payload_bytes(sm_ptr);
+            uint8_t * copy_to = session->in.data + (header->segment.number * max_payload_bytes);
+
+            session->segments.size_array[header->segment.number] = payload_bytes;
+            memcpy(copy_to, &recv_ptr->data[recv_ptr->processed_bytes], payload_bytes);
+            session->segments.processed++;
+        }
+        else
+        {
+            connector_debug_printf("sm_update_session: duplicate segment %d, in id %d\n", header->segment.number, session->request_id);
+        }
+    }
+    else
+    #endif
+    {
+        session->in.bytes = payload_bytes;
+        if (payload_bytes > 0)
+        {
+            result = sm_allocate_user_buffer(connector_ptr, &session->in);
+            ASSERT_GOTO(result == connector_working, error);
+            memcpy(session->in.data, &recv_ptr->data[recv_ptr->processed_bytes], payload_bytes);
+        }
+        else
+            session->in.data = NULL;
+        session->segments.processed++;
+    }
+
+    if (session->segments.processed >= session->segments.count)
+    {
+        session->bytes_processed = 0;
+        session->segments.processed = 0;
+        session->sm_state = connector_sm_state_process_payload;
+
+        #if (defined CONNECTOR_COMPRESSION)
+        if (SmIsCompressed(session->flags))
+        {
+            session->compress.out.data = NULL;
+            session->sm_state = connector_sm_state_decompress;
+        }
+        #endif
+    }
+
+    recv_ptr->processed_bytes += payload_bytes;
+    result = connector_working;
+
+error:
+    return result;
+}
+
+static connector_status_t sm_process_packet(connector_data_t * const connector_ptr, connector_sm_data_t * const sm_ptr)
+{
+    sm_header_t sm_header;
+    connector_status_t result;
+    connector_sm_packet_t * const recv_ptr = &sm_ptr->network.recv_packet;
+    size_t sm_bytes = recv_ptr->total_bytes - recv_ptr->processed_bytes;
+
+    sm_header.isPackCmd = connector_false;
+    sm_header.crc16 = 0;
+    do
+    {
+        result = sm_process_header(recv_ptr, &sm_header);
+        if (result != connector_working) goto error;
+
+        if ((sm_header.segment.count > sm_ptr->session.max_segments) || (sm_header.segment.number >= sm_ptr->session.max_segments))
+        {
+            connector_debug_printf("sm_process_packet: Exceeded maximum segments [%" PRIsize "/%" PRIsize "]!\n", sm_header.segment.count, sm_ptr->session.max_segments);
+            connector_debug_printf("Review CONNECTOR_SM_MULTIPART and CONNECTOR_SM_MAX_RX_SEGMENTS defines\n");
+            goto error;
+        }
+
+        if (sm_header.command == connector_sm_cmd_pack)
+        {
+            enum sm_pack_t
+            {
+                field_define(pack_header, flag, uint8_t),
+                field_define(pack_header, length, uint16_t),
+                record_end(pack_header)
+            };
+            uint8_t * const pack_header = &recv_ptr->data[recv_ptr->processed_bytes];
+            uint8_t const flag = message_load_u8(pack_header, flag);
+
+            #define SM_MESSAGE_PENDING 0x01
+            if ((flag & SM_MESSAGE_PENDING) == SM_MESSAGE_PENDING)
+            {
+                result = sm_more_data_callback(connector_ptr, sm_ptr);
+                if (result != connector_working) goto error;
+            }
+            #undef SM_MESSAGE_PENDING
+
+            sm_bytes = message_load_be16(pack_header, length);
+            recv_ptr->processed_bytes += record_end(pack_header);
+            continue;
+        }
+
+        {
+            size_t const payload_bytes = sm_bytes - sm_header.bytes;
+
+            ASSERT(sm_bytes >= sm_header.bytes);
+            result = sm_update_session(connector_ptr, sm_ptr, &sm_header, payload_bytes);
+        }
+
+        if (!sm_header.isPackCmd) break;
+
+        {
+            size_t const sm_header_size = 5;
+            size_t const remaining_bytes = recv_ptr->total_bytes - recv_ptr->processed_bytes;
+
+            if (remaining_bytes < sm_header_size) break;
+        }
+
+        sm_bytes = LoadBE16(&recv_ptr->data[recv_ptr->processed_bytes]);
+        recv_ptr->processed_bytes += sizeof(uint16_t);
+
+    } while (recv_ptr->processed_bytes < recv_ptr->total_bytes);
+
+    result = connector_working;
+
+error:
+    if (result == connector_invalid_payload_packet)  /* unreliable method, so silently ignore the packet */
+        result = connector_working;
+
+    recv_ptr->total_bytes = 0;
+    recv_ptr->processed_bytes = 0;
+
+    return result;
+}
+
+static connector_status_t sm_receive_data(connector_data_t * const connector_ptr, connector_sm_data_t * const sm_ptr)
+{
+    connector_status_t result = connector_pending;
+    connector_sm_packet_t * const recv_ptr = &sm_ptr->network.recv_packet;
+    connector_callback_status_t status;
+    connector_request_id_t request_id;
+    connector_network_receive_t read_data;
+
+    if (recv_ptr->total_bytes > 0) goto done;
+
+    read_data.handle = sm_ptr->network.handle;
+    read_data.buffer = recv_ptr->data;
+    read_data.bytes_available = sm_ptr->transport.mtu;
+    read_data.bytes_used = 0;
+
+    request_id.network_request = connector_request_id_network_receive;
+    status = connector_callback(connector_ptr->callback, sm_ptr->network.class_id, request_id, &read_data);
+    ASSERT(status != connector_callback_unrecognized);
+    switch (status)
+    {
+        case connector_callback_busy:
+            result = connector_idle;
+            goto done;
+
+        case connector_callback_continue:
+            recv_ptr->total_bytes = read_data.bytes_used;
+            recv_ptr->processed_bytes = 0;
+
+            switch (sm_ptr->network.transport)
+            {
+                #if (defined CONNECTOR_TRANSPORT_SMS)
+                case connector_transport_sms:
+                    result = sm_verify_sms_preamble(sm_ptr);
+                    switch(result)
+                    {
+                        case connector_working:
+                            break;
+                        case connector_invalid_response:
+                            /* not Device Cloud packet? Ignore the packet */
+                            recv_ptr->total_bytes = 0;
+                            recv_ptr->processed_bytes = 0;
+                            result = connector_working;
+                            goto done;
+                        default:
+                            goto done;
+                    }
+                                                                
+                    result = sm_decode_segment(connector_ptr, recv_ptr);
+
+                    /* Remove sms preamble */
+                    sm_ptr->network.recv_packet.processed_bytes = 0;
+
+                    break;
+                #endif
+
+                #if (defined CONNECTOR_TRANSPORT_UDP)
+                case connector_transport_udp:
+                    result = sm_verify_udp_header(sm_ptr);
+                    break;
+                #endif
+
+                default:
+                    ASSERT_GOTO(connector_false, done);
+                    break;
+            }
+
+            if(result != connector_working) goto done; /* not Device Cloud packet? */
+                result = sm_process_packet(connector_ptr, sm_ptr);
+            break;
+
+        case connector_callback_abort:
+            result = connector_abort;
+            break;
+
+        default:
+            connector_debug_printf("sm_receive_data: callback returned error [%d]\n", status);
+            result = connector_device_error;
+            break;
+    }
+
+    sm_verify_result(sm_ptr, &result);
+
+done:
+    return result;
+}
+
+#if (defined CONNECTOR_COMPRESSION)
+static connector_status_t sm_decompress_data(connector_data_t * const connector_ptr, connector_sm_data_t * const sm_ptr, connector_sm_session_t * const session)
+{
+    connector_status_t status = connector_working;
+    size_t const max_payload_bytes = sm_get_max_payload_bytes(sm_ptr);
+    uint8_t zlib_header[] = {0x58, 0xC3};
+    z_streamp const zlib_ptr = &session->compress.zlib;
+    int zret;
+
+    if (session->compress.out.data == NULL)
+    {
+        session->compress.out.bytes = max_payload_bytes;
+        status = sm_allocate_user_buffer(connector_ptr, &session->compress.out);
+        ASSERT_GOTO(status == connector_working, done);
+
+        memset(zlib_ptr, 0, sizeof *zlib_ptr);
+        zret = inflateInit(zlib_ptr);
+        ASSERT_GOTO(zret == Z_OK, error);
+        zlib_ptr->next_out = session->compress.out.data;
+        zlib_ptr->avail_out = session->compress.out.bytes;
+        zlib_ptr->next_in = zlib_header;
+        zlib_ptr->avail_in = sizeof zlib_header;
+    }
+
+    while (zlib_ptr->avail_out > 0)
+    {
+        if (zlib_ptr->avail_in == 0)
+        {
+            if (session->segments.processed == session->segments.count)
+            {
+                SmSetLastData(session->flags);
+                break;
+            }
+            else
+            {
+                size_t const data_index = session->segments.processed * max_payload_bytes;
+
+                zlib_ptr->next_in = &session->in.data[data_index];
+                zlib_ptr->avail_in = SmIsMultiPart(session->flags) ? session->segments.size_array[session->segments.processed] : session->in.bytes;
+                session->segments.processed++;
+            }
+        }
+
+        zret = inflate(zlib_ptr, Z_NO_FLUSH);
+        switch(zret)
+        {
+            case Z_STREAM_END:
+            case Z_BUF_ERROR:
+            case Z_OK:
+                break;
+
+            default:
+                status = connector_abort;
+                connector_debug_printf("ZLIB Return value [%d]\n", zret);
+                ASSERT_GOTO(connector_false, error);
+                break;
+        }
+    }
+
+    {
+        size_t const payload_bytes = session->compress.out.bytes - zlib_ptr->avail_out;
+
+        status = sm_pass_user_data(connector_ptr, session, session->compress.out.data, payload_bytes);
+        switch (status)
+        {
+            case connector_pending:
+                goto done;
+
+            case connector_working:
+                if (SmIsNotLastData(session->flags))
+                {
+                    zlib_ptr->next_out = session->compress.out.data;
+                    zlib_ptr->avail_out = session->compress.out.bytes;
+                    goto done;
+                }
+                break;
+
+            default:
+                break;
+        }
+    }
+
+error:
+    zret = inflateEnd(zlib_ptr);
+    ASSERT(zret == Z_OK);
+
+    if (status != connector_abort)
+    {
+        status = free_data_buffer(connector_ptr, named_buffer_id(sm_data_block), session->compress.out.data);
+        session->compress.out.data = NULL;
+    }
+
+done:
+    return status;
+}
+#endif
+
+static connector_status_t sm_handle_error(connector_data_t * const connector_ptr, connector_sm_session_t * const session)
+{
+    connector_status_t result = connector_working;
+    connector_sm_state_t next_state = connector_sm_state_complete;
+
+    if (SmIsCloudOwned(session->flags) && SmIsRequest(session->flags))
+        next_state = connector_sm_state_send_data;
+
+    SmSetError(session->flags);
+    if (session->user.context != NULL) /* let the user know */
+    {
+        result = sm_inform_session_complete(connector_ptr, session);
+        if (result != connector_working) goto error;
+    }
+
+    result = sm_switch_path(connector_ptr, session, next_state);
+
+error:
+    return result;
+}
+
+static connector_status_t sm_handle_complete(connector_data_t * const connector_ptr, connector_sm_data_t * const sm_ptr, connector_sm_session_t * const session)
+{
+    connector_status_t result = connector_working;
+
+    if (session->in.data != NULL)
+    {
+        result = free_data_buffer(connector_ptr, named_buffer_id(sm_data_block), session->in.data);
+        if (result != connector_working) goto error;
+    }
+
+    if (SmIsReboot(session->flags))
+        result = sm_process_reboot(connector_ptr);
+
+    switch(result)
+    {
+        case connector_abort:
+            /* Keep connector_abort as the result */
+            sm_delete_session(connector_ptr, sm_ptr, session);
+            break;
+        case connector_pending:
+            break;
+        default:
+            result = sm_delete_session(connector_ptr, sm_ptr, session);
+    }
+
+error:
+    return result;
+}
+
+static connector_status_t sm_process_recv_path(connector_data_t * const connector_ptr, connector_sm_data_t * const sm_ptr, connector_sm_session_t * const session)
+{
+    connector_status_t result = connector_abort;
+
+    ASSERT_GOTO(session != NULL, error);
+    switch (session->sm_state)
+    {
+        case connector_sm_state_receive_data:
+            if (sm_ptr->timeout_in_seconds != SM_WAIT_FOREVER)
+            {
+                unsigned long current_time = 0;
+
+                result = get_system_time(connector_ptr, &current_time);
+                ASSERT_GOTO(result == connector_working, error);
+                if (current_time > (session->start_time + sm_ptr->timeout_in_seconds))
+                {
+                    session->sm_state = connector_sm_state_error;
+                    session->error = connector_sm_error_timeout;
+                    connector_debug_printf("Sm session [%u] timeout... start time:%u, current time:%u\n", session->request_id, session->start_time, current_time);
+                }
+            }
+
+            result = connector_idle; /* still receiving data, handled in sm_receive_data() */
+            break;
+
+        #if (defined CONNECTOR_COMPRESSION)
+        case connector_sm_state_decompress:
+            result = sm_decompress_data(connector_ptr, sm_ptr, session);
+            break;
+        #endif
+
+        case connector_sm_state_process_payload:
+            result = sm_process_payload(connector_ptr, sm_ptr, session);
+            break;
+
+        case connector_sm_state_complete:
+            result = sm_handle_complete(connector_ptr, sm_ptr, session);
+            break;
+
+        case connector_sm_state_error:
+            result = sm_handle_error(connector_ptr, session);
+            break;
+
+        default:
+            ASSERT(connector_false);
+            break;
+    }
+
+    sm_verify_result(sm_ptr, &result);
+
+error:
+    return result;
+}
+