Diff: private/connector_data_service.h

Revision:

0:1c358ea10753

diff -r 000000000000 -r 1c358ea10753 private/connector_data_service.h
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/private/connector_data_service.h	Tue Dec 03 13:34:02 2013 +0000
@@ -0,0 +1,959 @@
+/*
+ * 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
+ * =======================================================================
+ */
+
+typedef enum
+{
+    data_service_opcode_put_request,
+    data_service_opcode_put_response,
+    data_service_opcode_device_request,
+    data_service_opcode_device_response
+} data_service_opcode_t;
+
+typedef struct
+{
+    void * callback_context;
+    connector_request_data_service_send_t const * header;
+    connector_bool_t dp_request;
+    connector_request_id_data_service_t request_type;
+} data_service_context_t;
+
+static void set_data_service_error(msg_service_request_t * const service_request, connector_session_error_t const error_code)
+{
+    service_request->error_value = error_code;
+    service_request->service_type = msg_service_type_error;
+}
+
+
+static connector_status_t call_ds_receive_callback(connector_data_t * const connector_ptr,
+                                                   data_service_context_t * const data_service,
+                                                   void * const data)
+{
+    connector_status_t result = connector_working;
+    connector_request_id_t request_id;
+
+    request_id.data_service_request = data_service->request_type;
+    switch (request_id.data_service_request)
+    {
+        case connector_request_id_data_service_receive_reply_length:
+            request_id.data_service_request = connector_request_id_data_service_receive_reply_data;
+            /* fall thru for error (not-handle) response data */
+        case connector_request_id_data_service_receive_target:
+        case connector_request_id_data_service_receive_data:
+        case connector_request_id_data_service_receive_status:
+        case connector_request_id_data_service_receive_reply_data:
+        {
+            connector_callback_status_t const status = connector_callback(connector_ptr->callback, connector_class_id_data_service, request_id, data);
+            switch (status)
+            {
+                case connector_callback_continue:
+                    break;
+                case connector_callback_error:
+                    data_service->request_type = connector_request_id_data_service_receive_reply_length;
+                    break;
+                case connector_callback_busy:
+                    result = connector_pending;
+                    break;
+                default:
+                    result = connector_abort;
+                    break;
+
+            }
+            break;
+        }
+
+        default:
+            ASSERT(connector_false);
+            break;
+    }
+
+    return result;
+}
+
+static connector_status_t process_ds_receive_target(connector_data_t * const connector_ptr,
+                                                    data_service_context_t * const data_service,
+                                                    uint8_t const * const data,
+                                                    size_t * const data_length)
+{
+    /* 1st message so let's parse message-start packet:
+     *
+     * Data Service Device request format:
+     *  -------------------------------------------------------------------------------------------------------
+     * |   0    |   1    |  2+N   |    +1     |     +1      |    +1       |    +M       |  ...       | +P      |
+     *  -------------------------------------------------------------------------------------------------------
+     * | Opcode | Target | Target | Parameter | Parameter 1 | Parameter 1 | Parameter 1 | Additional | Payload |
+     * |        | length | string |   count   |     ID      | data length |    data     | parameters |         |
+     *  -------------------------------------------------------------------------------------------------------
+     *
+     */
+    enum {
+        field_define(ds_device_request, opcode, uint8_t),
+        field_define(ds_device_request, target_length, uint8_t),
+        record_end(ds_device_request_header)
+    };
+
+    enum {
+        field_define(ds_device_request, parameter_count, uint8_t)
+    };
+
+    enum {
+        field_define(ds_device_request, parameter_id, uint8_t),
+        field_define(ds_device_request, parameter_length, uint8_t),
+        record_end(ds_device_request_parameter)
+    };
+
+    connector_status_t result = connector_working;
+    uint8_t const * ds_device_request = data;
+
+    char * target_string = NULL;
+    uint8_t const target_length =  message_load_u8(ds_device_request, target_length);
+
+    size_t const min_data_length = (size_t)(target_length +
+                      record_bytes(ds_device_request_header) +
+                      field_named_data(ds_device_request, parameter_count, size));
+
+    ASSERT_GOTO((message_load_u8(ds_device_request, opcode) == data_service_opcode_device_request), done);
+    ASSERT_GOTO(*data_length >= min_data_length, done);
+
+    ds_device_request += record_bytes(ds_device_request_header);
+
+    target_string = (char *)ds_device_request;
+    ds_device_request += target_length;
+
+    {
+        /* TODO: Parse and process each parameter in the future.
+         *      Ignore all parameters now.
+         */
+
+        uint8_t const parameter_count = message_load_u8(ds_device_request, parameter_count);
+        uint8_t i;
+
+        ds_device_request += field_named_data(ds_device_request, parameter_count, size);
+
+        for (i=0; i < parameter_count; i++)
+        {
+             unsigned int const parameter_length = message_load_u8(ds_device_request, parameter_length);
+             size_t const min_parameter_length = min_data_length + record_bytes(ds_device_request_parameter) + parameter_length;
+             ASSERT_GOTO(*data_length >= min_parameter_length, done);
+
+             ds_device_request += record_bytes(ds_device_request_parameter); /* skip id and length */
+             ds_device_request += parameter_length;
+
+        }
+    }
+
+    *data_length = (ds_device_request - data);
+
+    /* Add NUL to the target string. Must NULL-terminate it after parsing all parameters.
+     * The NUL char is on parameter_count field in the request.
+     */
+    target_string[target_length] = '\0';
+
+    switch (data_service->request_type)
+    {
+        case connector_request_id_data_service_receive_target:
+        {
+
+            connector_data_service_receive_target_t device_request;
+
+            device_request.transport = connector_transport_tcp;
+            device_request.user_context = data_service->callback_context;
+            device_request.target = target_string;
+            device_request.response_required = connector_true;
+
+            result = call_ds_receive_callback(connector_ptr, data_service, &device_request);
+            data_service->callback_context = device_request.user_context;
+            break;
+        }
+        default:
+            /* just skip the header and return */
+            break;
+    }
+
+done:
+    return result;
+}
+
+static connector_status_t process_ds_receive_data(connector_data_t * const connector_ptr,
+                                                  data_service_context_t * const data_service,
+                                                  uint8_t const * const data,
+                                                  size_t const data_length,
+                                                  unsigned int const flags)
+{
+    connector_status_t result;
+
+    connector_data_service_receive_data_t device_request;
+    device_request.transport = connector_transport_tcp;
+    device_request.user_context = data_service->callback_context;
+    device_request.buffer = data;
+    device_request.bytes_used = data_length;
+    device_request.more_data = MsgIsLastData(flags) ? connector_false : connector_true;
+
+    result = call_ds_receive_callback(connector_ptr, data_service, &device_request);
+    data_service->callback_context = device_request.user_context;
+
+    return result;
+}
+static connector_status_t process_data_service_device_request(connector_data_t * const connector_ptr,
+                                                 msg_service_request_t * const service_request)
+{
+    connector_status_t result = connector_working;
+    msg_session_t * const session = service_request->session;
+    data_service_context_t * data_service = session->service_context;
+    msg_service_data_t * const service_data = service_request->have_data;
+
+    uint8_t const * ds_device_request = service_data->data_ptr;
+    size_t ds_device_request_length = service_data->length_in_bytes;
+
+    connector_bool_t const isFirstRequest = connector_bool(MsgIsStart(service_data->flags));
+
+    if (isFirstRequest)
+    {
+        if (data_service == NULL)
+        {
+            /* 1st time here so let's allocate service context memory for device request service */
+            void * ptr;
+
+            result = malloc_data_buffer(connector_ptr, sizeof *data_service, named_buffer_id(msg_service), &ptr);
+            if (result != connector_working)
+            {
+                goto done;
+            }
+
+            data_service = ptr;
+            session->service_context = data_service;
+            data_service->callback_context = NULL;
+            data_service->dp_request = connector_false;
+            data_service->request_type = connector_request_id_data_service_receive_target;
+        }
+    }
+
+    switch (data_service->request_type)
+    {
+        case connector_request_id_data_service_receive_target:
+            ds_device_request_length = service_data->length_in_bytes;
+
+            result = process_ds_receive_target(connector_ptr, data_service, ds_device_request, &ds_device_request_length);
+
+            switch (result)
+            {
+            case connector_working:
+                if (data_service->request_type == connector_request_id_data_service_receive_target)
+                    data_service->request_type = connector_request_id_data_service_receive_data;
+                /* make it return busy so it comes here to skip parsing the header for the 1st data block */
+                result = connector_pending;
+                break;
+            default:
+                break;
+            }
+            break;
+
+        case connector_request_id_data_service_receive_data:
+            if (isFirstRequest)
+            {
+                /* skip the header; just get the point to data  */
+                result = process_ds_receive_target(connector_ptr, data_service, ds_device_request, &ds_device_request_length);
+                if (result != connector_working) goto done;
+                ds_device_request += ds_device_request_length;
+            }
+            {
+                size_t const data_length = (size_t)(ds_device_request - (uint8_t *)service_data->data_ptr);
+                ds_device_request_length = service_data->length_in_bytes - data_length;
+
+                result = process_ds_receive_data(connector_ptr, data_service, ds_device_request, ds_device_request_length, service_data->flags);
+                switch (result)
+                {
+                    case connector_working:
+                        /* data_service->request_type may have been changed inside process_ds_receive_data() */
+                        switch (data_service->request_type)
+                        {
+                            case connector_request_id_data_service_receive_reply_length:
+                                break;
+
+                            default:
+                                if (MsgIsLastData(service_data->flags))
+                                    data_service->request_type = connector_request_id_data_service_receive_reply_data;
+                                break;
+                        }
+                        break;
+                    default:
+                        break;
+                }
+                break;
+            }
+        case connector_request_id_data_service_receive_reply_length:
+            /* We set this when callback returns error.
+             * We need to respond with an error after receiving all request data.
+             */
+            break;
+
+        default:
+            ASSERT(connector_false);
+            goto done;
+    }
+
+done:
+    return result;
+}
+
+static connector_status_t process_data_service_device_response(connector_data_t * const connector_ptr,
+                                                               msg_service_request_t * const service_request)
+{
+    /* Data Service Device response format:
+     *  ---------------------------------
+     * |   0    |   1    |     2...      |
+     *  ---------------------------------
+     * | Opcode | status | Response Data |
+     *  ---------------------------------
+     */
+    enum {
+        field_define(ds_device_response, opcode, uint8_t),
+        field_define(ds_device_response, status, uint8_t),
+        record_end(ds_device_response_header)
+    };
+
+    connector_status_t result = connector_working;
+    msg_service_data_t * const service_data = service_request->need_data;
+    msg_session_t * const session = service_request->session;
+    data_service_context_t * const data_service = session->service_context;
+    connector_bool_t const isFirstResponse = connector_bool(MsgIsStart(service_data->flags));
+
+    /* save some room for response header on 1st response data */
+    size_t const header_length = isFirstResponse == connector_true ? record_bytes(ds_device_response_header) : 0;
+    uint8_t * const data_ptr = service_data->data_ptr;
+
+
+    connector_data_service_receive_reply_data_t device_request;
+
+    device_request.transport = connector_transport_tcp;
+    device_request.user_context = data_service->callback_context;
+    device_request.buffer = data_ptr + header_length;
+    device_request.bytes_available = service_data->length_in_bytes - header_length;
+    device_request.bytes_used = 0;
+    device_request.more_data = connector_false;
+
+    {
+        connector_request_id_data_service_t const request_type = data_service->request_type;
+
+        switch (request_type)
+        {
+            case connector_request_id_data_service_receive_reply_length:
+            /* We got here because callback returns error for request data.
+             * Just get the replay data.
+             */
+             /* fall thru */
+            case connector_request_id_data_service_receive_reply_data:
+            {
+
+                result = call_ds_receive_callback(connector_ptr, data_service, &device_request);
+                data_service->callback_context = device_request.user_context;
+                if (request_type != data_service->request_type &&
+                    data_service->request_type == connector_request_id_data_service_receive_reply_length)
+                {
+                    /* callback returns error on reply data
+                     * We need to cancel the message in messaging layer.
+                     */
+                    device_request.bytes_used = 0;
+                    device_request.more_data = connector_false;
+                    set_data_service_error(service_request, connector_session_error_cancel);
+                }
+                break;
+            }
+            default:
+                /* should be here */
+                ASSERT(connector_false);
+                goto done;
+        }
+    }
+    if (isFirstResponse)
+    {
+
+        enum {
+            connector_data_service_device_success,
+            connector_data_service_device_not_handled
+        };
+
+        /* Add header for 1st response message */
+        uint8_t * const ds_device_response = service_data->data_ptr;
+        uint8_t const target_status = (data_service->request_type == connector_request_id_data_service_receive_reply_length) ?
+                                connector_data_service_device_not_handled: connector_data_service_device_success;
+
+        message_store_u8(ds_device_response, opcode, data_service_opcode_device_response);
+        message_store_u8(ds_device_response, status, target_status);
+    }
+
+    if (!device_request.more_data)
+    {
+        data_service->dp_request = connector_true;
+        MsgSetLastData(service_data->flags);
+    }
+    service_data->length_in_bytes = device_request.bytes_used + header_length;
+
+done:
+    return result;
+}
+
+
+static connector_status_t process_data_service_device_error(connector_data_t * const connector_ptr,
+                                                       msg_service_request_t * const service_request)
+{
+    connector_status_t result = connector_working;
+
+    msg_session_t * const session = service_request->session;
+    data_service_context_t * const data_service = session->service_context;
+    connector_data_service_status_t device_request;
+
+
+    device_request.transport = connector_transport_tcp;
+    device_request.user_context = data_service->callback_context;
+    data_service->request_type = connector_request_id_data_service_receive_status;
+
+    switch (service_request->error_value)
+    {
+        case connector_session_error_none:
+            device_request.status = connector_data_service_status_complete;
+            break;
+
+        case connector_session_error_cancel:
+            device_request.status = connector_data_service_status_cancel;
+            break;
+
+        case connector_session_error_timeout:
+            device_request.status = connector_data_service_status_timeout;
+            break;
+
+        default:
+            device_request.status = connector_data_service_status_session_error;
+            device_request.session_error = service_request->error_value;
+            break;
+    }
+
+    result = call_ds_receive_callback(connector_ptr, data_service, &device_request);
+
+    data_service->callback_context = device_request.user_context;
+
+    return result;
+}
+
+static connector_status_t data_service_device_request_callback(connector_data_t * const connector_ptr, msg_service_request_t * const service_request)
+{
+    connector_status_t status = connector_working;
+
+    switch (service_request->service_type)
+    {
+    case msg_service_type_need_data:
+        status = process_data_service_device_response(connector_ptr, service_request);
+        break;
+
+    case msg_service_type_have_data:
+        status = process_data_service_device_request(connector_ptr, service_request);
+        break;
+
+    case msg_service_type_error:
+    {
+        msg_session_t * const session = service_request->session;
+        session->error = service_request->error_value;
+        status = process_data_service_device_error(connector_ptr, service_request);
+        break;
+    }
+
+    case msg_service_type_free:
+        {
+            msg_session_t * const session = service_request->session;
+#if (CONNECTOR_VERSION >= 0x02010000)
+            if (session->error == connector_session_error_none)
+            {
+                /* If there is no error, call the user to inform that session is done */
+                status = process_data_service_device_error(connector_ptr, service_request);
+                if (status != connector_working)
+                    break;
+            }
+#endif
+            status = free_data_buffer(connector_ptr, named_buffer_id(msg_service), session->service_context);
+            break;
+        }
+
+    default:
+        ASSERT(connector_false);
+        break;
+    }
+
+    return status;
+}
+
+static size_t fill_put_request_header(connector_request_data_service_send_t const * const request, uint8_t * const data)
+{
+    uint8_t * ptr = data;
+
+    *ptr++ = data_service_opcode_put_request;
+
+    /* fill path */
+    if (request->path != NULL)
+    {
+        uint8_t const bytes = (uint8_t) strlen(request->path);
+
+        ASSERT(strlen(request->path) <= UCHAR_MAX);
+        *ptr++ = bytes;
+        memcpy(ptr, request->path, bytes);
+        ptr += bytes;
+    }
+
+    /* fill parameters */
+    {
+        connector_bool_t const have_type = connector_bool(request->content_type != NULL);
+        uint8_t const parameter_requested = 1;
+        uint8_t params = have_type ? 1 : 0;
+
+        enum
+        {
+            parameter_id_content_type,
+            parameter_id_archive,
+            parameter_id_append,
+            parameter_id_transient
+        };
+
+        if (request->option != connector_data_service_send_option_overwrite) params++;
+
+        *ptr++ = params;
+
+        if (have_type)
+        {
+            uint8_t const bytes = (uint8_t) strlen(request->content_type);
+
+            ASSERT(strlen(request->content_type) <= UCHAR_MAX);
+            *ptr++ = parameter_id_content_type;
+            *ptr++ = bytes;
+            memcpy(ptr, request->content_type, bytes);
+            ptr += bytes;
+        }
+
+        switch(request->option)
+        {
+            case connector_data_service_send_option_archive:
+                *ptr++ = parameter_id_archive;
+                *ptr++ = parameter_requested;
+                break;
+
+            case connector_data_service_send_option_append:
+                *ptr++ = parameter_id_append;
+                *ptr++ = parameter_requested;
+                break;
+
+            case connector_data_service_send_option_transient:
+                *ptr++ = parameter_id_transient;
+                *ptr++ = parameter_requested;
+                break;
+
+            default:
+                break;
+        }
+    }
+
+    return (size_t)(ptr - data);
+}
+
+static connector_status_t call_put_request_user(connector_data_t * const connector_ptr, msg_service_request_t * const service_request, connector_request_id_data_service_t const request_id, void * const cb_data)
+{
+    connector_status_t status = connector_working;
+    msg_session_t * const session = service_request->session;
+    data_service_context_t * const context = (session != NULL) ? session->service_context : NULL;
+    connector_callback_status_t callback_status = connector_callback_continue;
+
+    if ((context == NULL) || (context->dp_request == connector_false))
+    {
+        connector_request_id_t request;
+
+        request.data_service_request = request_id;
+        callback_status = connector_callback(connector_ptr->callback, connector_class_id_data_service, request, cb_data);
+    }
+    #if (defined CONNECTOR_DATA_POINTS)
+    else
+    {
+        callback_status = dp_handle_callback(connector_ptr, request_id, cb_data);
+    }
+    #endif
+
+    switch (callback_status)
+    {
+        case connector_callback_continue:
+            status = connector_working;
+            break;
+
+        case connector_callback_error:
+            set_data_service_error(service_request, connector_session_error_cancel);
+            status = connector_working;
+            break;
+
+        case connector_callback_busy:
+            status = connector_pending;
+            break;
+
+        default:
+            status = connector_abort;
+            break;
+    }
+
+    return status;
+}
+
+static connector_status_t process_send_request(connector_data_t * const connector_ptr, msg_service_request_t * const service_request, data_service_context_t * const ds_ptr)
+{
+    connector_status_t status = connector_working;
+    msg_service_data_t * const service_data = service_request->need_data;
+    connector_data_service_send_data_t user_data;
+
+    user_data.transport = connector_transport_tcp;
+    user_data.user_context = ds_ptr->callback_context;
+    user_data.bytes_used = 0;
+    user_data.more_data = connector_false;
+
+    if (MsgIsStart(service_data->flags))
+    {
+        uint8_t * dptr = service_data->data_ptr;
+        size_t const bytes = fill_put_request_header(ds_ptr->header, dptr);
+
+        if (bytes >= service_data->length_in_bytes)
+        {
+            connector_debug_printf("process_send_request: required bytes [%" PRIsize "] is more than available [%" PRIsize "]\n", bytes, service_data->length_in_bytes);
+            goto error;
+        }
+
+        user_data.buffer = dptr + bytes;
+        user_data.bytes_available = service_data->length_in_bytes - bytes;
+        service_data->length_in_bytes = bytes;
+    }
+    else
+    {
+        user_data.buffer = service_data->data_ptr;
+        user_data.bytes_available = service_data->length_in_bytes;
+        service_data->length_in_bytes = 0;
+    }
+
+    status = call_put_request_user(connector_ptr, service_request, connector_request_id_data_service_send_data, &user_data);
+    if (status == connector_working)
+    {
+        service_data->flags = 0;
+        service_data->length_in_bytes += user_data.bytes_used;
+        if (user_data.more_data == connector_false)
+            MsgSetLastData(service_data->flags);
+    }
+    goto done;
+
+error:
+    set_data_service_error(service_request, connector_session_error_format);
+
+done:
+    return status;
+}
+
+static connector_status_t process_send_response(connector_data_t * const connector_ptr, msg_service_request_t * const service_request, data_service_context_t * const ds_ptr)
+{
+    connector_status_t status = connector_working;
+    connector_data_service_send_response_t user_data;
+
+    /* Data Service put response format:
+     *  ---------------------------------
+     * |   0    |   1    |     2...      |
+     *  ---------------------------------
+     * | Opcode | status | Response Data |
+     *  ---------------------------------
+     */
+    enum
+    {
+        field_define(put_response, opcode, uint8_t),
+        field_define(put_response, status, uint8_t),
+        record_end(put_response)
+    };
+
+    enum
+    {
+        ds_data_success,
+        ds_data_bad_request,
+        ds_data_service_unavailable,
+        ds_data_cloud_error
+    };
+
+    msg_service_data_t * const service_data = service_request->have_data;
+    uint8_t * const put_response = service_data->data_ptr;
+    uint8_t const opcode = message_load_u8(put_response, opcode);
+    uint8_t const result = message_load_u8(put_response, status);
+
+    ASSERT_GOTO(MsgIsStart(service_data->flags), error);
+    ASSERT_GOTO(opcode == data_service_opcode_put_response, error);
+
+    user_data.transport = connector_transport_tcp;
+    user_data.user_context = ds_ptr->callback_context;
+    if (service_data->length_in_bytes > record_end(put_response))
+    {
+        int const max_hint_length = (MSG_MAX_RECV_PACKET_SIZE - PACKET_EDP_HEADER_SIZE - record_end(start_packet) - record_end(put_response));
+        uint8_t * const hint_start = put_response + record_end(put_response);
+        uint8_t * const hint_end = put_response + service_data->length_in_bytes;
+        size_t hint_length = (hint_end - hint_start) < max_hint_length ? hint_end - hint_start : max_hint_length;
+        char * const hint = (char *)hint_start;
+
+        /* Add a null-terminator only if necessary */
+        if (hint[hint_length - 1] != '\0')
+            hint[hint_length] = '\0';
+        user_data.hint = hint;
+    }
+    else
+    {
+        user_data.hint = NULL;
+    }
+
+    switch (result)
+    {
+    case ds_data_success:
+        user_data.response = connector_data_service_send_response_success;
+        break;
+
+    case ds_data_bad_request:
+        user_data.response = connector_data_service_send_response_bad_request;
+        break;
+
+    case ds_data_service_unavailable:
+        user_data.response = connector_data_service_send_response_unavailable;
+        break;
+
+    case ds_data_cloud_error:
+        user_data.response = connector_data_service_send_response_cloud_error;
+        break;
+
+    default:
+        ASSERT(connector_false);
+        break;
+    }
+
+    status = call_put_request_user(connector_ptr, service_request, connector_request_id_data_service_send_response, &user_data);
+    goto done;
+
+error:
+    set_data_service_error(service_request, connector_session_error_format);
+
+done:
+    return status;
+}
+
+static connector_status_t process_send_error(connector_data_t * const connector_ptr, msg_service_request_t * const service_request, void * const cb_context)
+{
+    connector_status_t status = connector_working;
+    connector_data_service_status_t user_data;
+
+    user_data.transport = connector_transport_tcp;
+    user_data.user_context = cb_context;
+    user_data.session_error = connector_session_error_none;
+
+    switch (service_request->error_value)
+    {
+        case connector_session_error_none:
+            user_data.status = connector_data_service_status_complete;
+            break;
+
+        case connector_session_error_cancel:
+            user_data.status = connector_data_service_status_cancel;
+            break;
+
+        case connector_session_error_timeout:
+            user_data.status = connector_data_service_status_timeout;
+            break;
+
+        default:
+            user_data.status = connector_data_service_status_session_error;
+            user_data.session_error = service_request->error_value;
+            break;
+    }
+
+    status = call_put_request_user(connector_ptr, service_request, connector_request_id_data_service_send_status, &user_data);
+
+    return status;
+}
+
+static connector_status_t data_service_put_request_callback(connector_data_t * const connector_ptr, msg_service_request_t * const service_request)
+{
+    connector_status_t status;
+    msg_session_t * const session = service_request->session;
+    data_service_context_t * const ds_ptr = session->service_context;
+
+    switch (service_request->service_type)
+    {
+        case msg_service_type_need_data:
+            status = process_send_request(connector_ptr, service_request, ds_ptr);
+            break;
+
+        case msg_service_type_have_data:
+            status = process_send_response(connector_ptr, service_request, ds_ptr);
+            break;
+
+        case msg_service_type_error:
+        {
+            msg_session_t * const session = service_request->session;
+            session->error = service_request->error_value;
+            status = process_send_error(connector_ptr, service_request, ds_ptr->callback_context);
+            break;
+        }
+
+        case msg_service_type_free:
+            {
+#if (CONNECTOR_VERSION >= 0x02010000)
+                msg_session_t * const session = service_request->session;
+                if (session->error == connector_session_error_none)
+                {
+                    /* If there is no error, call the user to inform that session is done */
+                    status = process_send_error(connector_ptr, service_request, ds_ptr->callback_context);
+                    if (status != connector_working)
+                        break;
+                }
+#endif
+                if (ds_ptr != NULL)
+                    status = free_data_buffer(connector_ptr, named_buffer_id(put_request), ds_ptr);
+                else
+                    status = connector_working;
+                break;
+            }
+        default:
+            status = connector_idle;
+            ASSERT(connector_false);
+            break;
+    }
+
+    return status;
+}
+
+static connector_status_t data_service_put_request_init(connector_data_t * const connector_ptr, msg_service_request_t * const service_request)
+{
+    connector_status_t status = connector_working;
+    connector_session_error_t result = service_request->error_value;
+    msg_session_t * const session = service_request->session;
+    connector_request_data_service_send_t * send_ptr = (void *)service_request->have_data;
+    data_service_context_t * ds_ptr = NULL;
+
+    if (send_ptr != NULL)
+    {
+        void * ptr;
+
+        if ((result != connector_session_error_none) || (session == NULL)) goto error;
+
+        status = malloc_data_buffer(connector_ptr, sizeof *ds_ptr, named_buffer_id(put_request), &ptr);
+        if (status != connector_working)
+            goto error;
+
+        ds_ptr = ptr;
+    }
+    else
+    {
+        status = connector_invalid_data;
+        ASSERT_GOTO(connector_false, done);
+    }
+
+    ds_ptr->header = send_ptr;
+    ds_ptr->callback_context = send_ptr->user_context;
+    ds_ptr->request_type = connector_request_id_data_service_send_data;
+    ds_ptr->dp_request = connector_false;
+    session->service_context = ds_ptr;
+
+    #if (defined CONNECTOR_DATA_POINTS)
+    {
+        char const data_point_prefix[] = "DataPoint/";
+
+        ds_ptr->dp_request = connector_bool(!strncmp(ds_ptr->header->path, data_point_prefix, strlen(data_point_prefix)));
+    }
+    #endif
+
+    goto done;
+
+error:
+    set_data_service_error(service_request, result);
+    process_send_error(connector_ptr, service_request, send_ptr->user_context);
+
+done:
+    return status;
+}
+
+static connector_status_t data_service_callback(connector_data_t * const connector_ptr, msg_service_request_t * const service_request)
+{
+    connector_status_t status = connector_idle;
+    msg_session_t * session;
+    data_service_context_t * ds_ptr;
+
+    ASSERT_GOTO(connector_ptr != NULL, done);
+    ASSERT_GOTO(service_request != NULL, done);
+
+    session = service_request->session;
+
+    if (service_request->service_type == msg_service_type_pending_request || session == NULL)
+    {
+        status = data_service_put_request_init(connector_ptr, service_request);
+        goto done;
+    }
+
+    ds_ptr = session->service_context;
+
+    if (ds_ptr == NULL)
+    {
+        status = data_service_device_request_callback(connector_ptr, service_request);
+        goto done;
+    }
+
+    switch (ds_ptr->request_type)
+    {
+        case connector_request_id_data_service_send_length:
+        case connector_request_id_data_service_send_data:
+        case connector_request_id_data_service_send_status:
+        case connector_request_id_data_service_send_response:
+            status = data_service_put_request_callback(connector_ptr, service_request);
+            break;
+
+        case connector_request_id_data_service_receive_target:
+        case connector_request_id_data_service_receive_data:
+        case connector_request_id_data_service_receive_status:
+        case connector_request_id_data_service_receive_reply_length:
+        case connector_request_id_data_service_receive_reply_data:
+            status = data_service_device_request_callback(connector_ptr, service_request);
+            break;
+
+        default:
+            ASSERT(connector_false);
+            break;
+    }
+
+done:
+    return status;
+}
+
+static connector_status_t connector_facility_data_service_cleanup(connector_data_t * const connector_ptr)
+{
+    return msg_cleanup_all_sessions(connector_ptr,  msg_service_id_data);
+}
+
+static connector_status_t connector_facility_data_service_delete(connector_data_t * const data_ptr)
+{
+    return msg_delete_facility(data_ptr, msg_service_id_data);
+}
+
+static connector_status_t connector_facility_data_service_init(connector_data_t * const data_ptr, unsigned int const facility_index)
+{
+    return msg_init_facility(data_ptr, facility_index, msg_service_id_data, data_service_callback);
+}
+
+static connector_status_t data_service_initiate(connector_data_t * const connector_ptr,  void const * request)
+{
+    connector_status_t status = connector_invalid_data;
+
+    ASSERT_GOTO(request != NULL, error);
+
+    status = msg_initiate_request(connector_ptr, request) == connector_true ? connector_success : connector_service_busy;
+
+error:
+    return status;
+}
+
+
+