/*
 * Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.
 *
 * This software is dual-licensed to you under the MIT License (MIT) and
 * the Universal Permissive License (UPL). See the LICENSE file in the root
 * directory for license terms. You may choose either license, or both.
 */

#include <stdio.h>
#include <string.h>
#include <limits.h>

#include "advanced/iotcs_messaging.h"
#include "advanced/iotcs_message.h"
#include "messaging/msg_private.h"
#include "device_model/device_model_private.h"
#include "iotcs/iotcs_private.h"
#include "device_model/device_model_capability.h"
#include "message_persistence.h"
#include "util/util_thread_private.h"
#include "iotcs_port_thread.h"
#include "iotcs_port_system.h"
#include "iotcs_port_queue.h"
#include "iotcs_port_ssl.h"
#include "policy/device_policy.h"
#include "util/util.h"
#include "util/util_memory.h"
#include "trusted_assets_manager/iotcs_tam.h"

#include "log/log.h"
#define IOTCSP_MODULE_LOG_CHANNEL LOG_CHANNEL_MSG
#include "log/log_template.h"

#define STOP_MESSAGE_BASE 1
#define UPDATE_POLLING_INTERVAL_MESSAGE_BASE 2

#ifndef IOTCS_SETTLE_TIME_MS
#define IOTCS_SETTLE_TIME_MS 5000
#endif

#define MAXIMUM_ITERATION_WITHOUT_PROGRESS 5

#define MSG_STOP_RECEIVE_REQUEST ((iotcs_request_message *)1)

#define MSG_INF_POLLING_INTERVAL -1

#ifndef IOTCS_ASYNC_FAIL_REASON_BUFFER_SIZE
#define IOTCS_ASYNC_FAIL_REASON_BUFFER_SIZE 256
#endif
static char g_async_send_fail_reason_buf[IOTCS_ASYNC_FAIL_REASON_BUFFER_SIZE];

static iotcs_message STOP_MESSAGE = {.base = (const iotcs_message_base*) STOP_MESSAGE_BASE};
static iotcs_message UPDATE_POLLING_INTERVAL_MESSAGE = {.base = (const iotcs_message_base*) UPDATE_POLLING_INTERVAL_MESSAGE_BASE};
static iotcs_port_message_queue g_send_queue;
static iotcs_port_thread g_send_thread;
static iotcs_port_thread g_recv_thread;
static volatile int64_t g_polling_timeout_ms;

static iotcs_message_dispatcher_delivery_callback g_delivery_cb = NULL;
iotcs_message_dispatcher_error_callback g_error_cb = NULL;

static cl_message_dispatcher_counters g_message_dispatcher_counters;

#ifdef IOTCS_LONG_POLLING
static iotcs_bool g_use_long_polling = IOTCS_FALSE;
#endif

int g_batch_message_size = 0;
int64_t g_batch_message_time = 0;
iotcs_message_severity g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_LOW;
static int64_t g_settle_time;

static void set_settle_time() {
    g_settle_time = iotcs_port_get_current_time_millis() + IOTCS_SETTLE_TIME_MS;
}

iotcs_bool cl_is_settle_time_passed() {
    return iotcs_port_get_current_time_millis() > g_settle_time ? IOTCS_TRUE : IOTCS_FALSE;
}

void iotcs_message_dispatcher_set_delivery_callback(iotcs_message_dispatcher_delivery_callback delivery_cb) {
    g_delivery_cb = delivery_cb;
}

void iotcs_message_dispatcher_set_error_callback(iotcs_message_dispatcher_error_callback error_cb) {
    g_error_cb = error_cb;
}
void release_alert_msg_copy(iotcs_message *msg) {
    int i;
    int attr_num = 0;
    iotcs_data_item_desc* items_desc;
    iotcs_value *values;
    if (!msg) {
        return;
    }

    values = msg->u.alert.items_value;
    items_desc = msg->u.alert.items_desc;
    i = 0;
    if (values && items_desc) {
        while (items_desc[i].type != IOTCS_VALUE_TYPE_NONE) {
            i++;
        }
    }
    attr_num = i;

    items_desc = msg->u.alert.items_desc;
    values = msg->u.alert.items_value;


    for (i = 0; i < attr_num; i++) {
        if (msg->u.alert.items_desc[i].type == IOTCS_VALUE_TYPE_STRING) {
            util_free((void*)values[i].string_value);
        }
#if defined IOTCS_STORAGE_SUPPORT
        else if (msg->u.alert.items_desc[i].type == IOTCS_VALUE_TYPE_URI) {
            dm_free_uri_object((void*)values[i].uri_object);
        }
#endif
        util_free((void*)items_desc[i].key);
    }
    if (msg->base) {
        util_free((void*)msg->base->sender);
        util_free((void*)msg->base->source);
        util_free((void*)msg->base->destination);
    }
    util_free((void*)msg->base);
    if (msg->u.alert.base) {
        util_free((void*)msg->u.alert.base->format);
        util_free((void*)msg->u.alert.base->description);
    }
    util_free((void*)msg->u.alert.base);
    util_free((void*)items_desc);
    util_free((void*)values);
}

iotcs_result copy_alert_msg(iotcs_message *msg_src, iotcs_message **msg_dst) {
    int i;
    iotcs_data_item_desc* items_desc = NULL;
    iotcs_alert_message_base* alert_base = NULL;
    iotcs_value *values = NULL;
    int attr_num = 0;
    *msg_dst = NULL;
    iotcs_message_base *msg_base = NULL;
    iotcs_result rv = IOTCS_RESULT_FAIL;

    if (msg_src->base->type == IOTCS_MESSAGE_ALERT) {
        values = msg_src->u.alert.items_value;
        items_desc = msg_src->u.alert.items_desc;
        i = 0;
        if (values && items_desc) {
            while (items_desc[i].type != IOTCS_VALUE_TYPE_NONE) {
                i++;
            }
        }
        attr_num = i;
    }

    iotcs_message *msg;
    GOTO_ERR_OOM(rv, msg = (iotcs_message*)util_malloc(sizeof(iotcs_message)));
    memset(msg, 0, sizeof (iotcs_message));

    GOTO_ERR_OOM(rv, values = (iotcs_value*) util_malloc(sizeof (iotcs_value) * attr_num));
    memset(values, 0, sizeof (iotcs_value) * attr_num);

    GOTO_ERR_OOM(rv, items_desc = (iotcs_data_item_desc*) util_malloc(sizeof (iotcs_data_item_desc) * (attr_num + 1)));
    GOTO_ERR_OOM(rv, msg_base = (iotcs_message_base *)util_malloc(sizeof(iotcs_message_base)));
    msg->base = msg_base;
    msg_base->type = msg_src->base->type;
    msg_base->reliability = msg_src->base->reliability;
    msg_base->priority = msg_src->base->priority;
    if (msg_src->base->destination) {
        GOTO_ERR_OOM(rv, msg_base->destination = util_safe_strcpy(msg_src->base->destination));
    } else {
        GOTO_ERR_OOM(rv, msg_base->destination = util_safe_strcpy("iotserver"));
    }
    if (msg_src->base->sender) {
        GOTO_ERR_OOM(rv, msg_base->sender = util_safe_strcpy(msg_src->base->sender));
    } else {
        msg_base->sender = NULL;
    }
    if (msg_src->base->source) {
        GOTO_ERR_OOM(rv, msg_base->source = util_safe_strcpy(msg_src->base->source));
    } else {
        msg_base->source = NULL;
    }

    for (i = 0; i < attr_num; i++) {
        if (msg_src->u.alert.items_desc[i].type == IOTCS_VALUE_TYPE_STRING) {
            if (msg_src->u.alert.items_value[i].string_value) {
                GOTO_ERR_OOM(rv, values[i].string_value = util_safe_strcpy(msg_src->u.alert.items_value[i].string_value));
            } else {
                GOTO_ERR_OOM(rv, values[i].string_value = util_safe_strcpy(""));
            }
#if defined IOTCS_STORAGE_SUPPORT
        } else if (msg_src->u.alert.items_desc[i].type == IOTCS_VALUE_TYPE_URI) {
            values[i].uri_object = msg_src->u.alert.items_value[i].uri_object;
			msg_src->u.alert.items_value[i].uri_object->ref_cnt++;
#endif
        } else if (msg_src->u.alert.items_desc[i].type == IOTCS_VALUE_TYPE_INT) {
            values[i].int_value = msg_src->u.alert.items_value[i].int_value;
        } else if (msg_src->u.alert.items_desc[i].type == IOTCS_VALUE_TYPE_NUMBER) {
            values[i].number_value = msg_src->u.alert.items_value[i].number_value;
        } else {
            values[i] = msg_src->u.alert.items_value[i];
        }

        /* set description */
        GOTO_ERR_OOM(rv, items_desc[i].key = util_safe_strcpy(msg_src->u.alert.items_desc[i].key));
        items_desc[i].type = msg_src->u.alert.items_desc[i].type;
    }
    items_desc[i].key = NULL;
    items_desc[i].type = IOTCS_VALUE_TYPE_NONE;
    GOTO_ERR_OOM(rv, alert_base = (iotcs_alert_message_base*)util_malloc(sizeof(iotcs_alert_message_base)));
    memset(alert_base, 0, sizeof (iotcs_alert_message_base));
    if (msg_src->u.alert.base->format) {
        GOTO_ERR_OOM(rv, alert_base->format = util_safe_strcpy(msg_src->u.alert.base->format));
    } else {
        GOTO_ERR_OOM(rv, alert_base->format = util_safe_strcpy(""));
    }
    if (msg_src->u.alert.base->description) {
        GOTO_ERR_OOM(rv, alert_base->description = util_safe_strcpy(msg_src->u.alert.base->description));
    } else {
        GOTO_ERR_OOM(rv, alert_base->description = util_safe_strcpy(""));
    }
    alert_base->severity_level = msg_src->u.alert.base->severity_level;
    msg->u.alert.base = alert_base;
    msg->u.alert.items_value = values;
    msg->u.alert.items_desc = items_desc;
    msg->message_mark = (void*)IOTCS_OFFERED_ALERT_MARK;
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
    msg->error_callback = msg_src->error_callback;
    msg->error_callback_cpp = msg_src->error_callback_cpp;
#endif
    *msg_dst = msg;

    return IOTCS_RESULT_OK;
    error:
    msg->base = msg_base;
    msg->u.alert.base = alert_base;
    msg->u.alert.items_value = values;
    msg->u.alert.items_desc = items_desc;
    release_alert_msg_copy(msg);
    return rv;
    
}

iotcs_result copy_msg(iotcs_message *msg_src, iotcs_message **msg_dst) {
    int i;
    iotcs_data_item_desc* items_desc = NULL;
    iotcs_value *values = NULL;
    int attr_num = 0;
    *msg_dst = NULL;
    iotcs_message *msg = NULL;
    iotcs_message_base * msg_base = NULL;
    iotcs_data_message_base* data_base = NULL;
    iotcs_result rv = IOTCS_RESULT_FAIL;

    if (msg_src->base->type == IOTCS_MESSAGE_DATA) {
        values = msg_src->u.data.items_value;
        items_desc = msg_src->u.data.items_desc;
        i = 0;
        if (values && items_desc) {
            while (items_desc[i].type != IOTCS_VALUE_TYPE_NONE) {
                i++;
            }
        }
        attr_num = i;
    }

    GOTO_ERR_OOM(rv, msg = (iotcs_message*)util_malloc(sizeof(iotcs_message)));
    memset(msg, 0, sizeof (iotcs_message));

    GOTO_ERR_OOM(rv, values = (iotcs_value*) util_malloc(sizeof (iotcs_value) * attr_num));
    memset(values, 0, sizeof (iotcs_value) * attr_num);
    if (NULL == values) {
        return IOTCS_RESULT_FAIL;
    }

    GOTO_ERR_OOM(rv, items_desc = (iotcs_data_item_desc*) util_malloc(sizeof (iotcs_data_item_desc) * (attr_num + 1)));
    memset(items_desc, 0, sizeof (iotcs_data_item_desc));
    GOTO_ERR_OOM(rv, msg_base = (iotcs_message_base *)util_malloc(sizeof(iotcs_message_base)));
    memset(msg_base, 0, sizeof (iotcs_message_base));
    msg_base->type = msg_src->base->type;
    msg_base->reliability = msg_src->base->reliability;
    msg_base->priority = msg_src->base->priority;
    if (msg_src->base->destination) {
        GOTO_ERR_OOM(rv, msg_base->destination = util_safe_strcpy(msg_src->base->destination));
    } else {
        GOTO_ERR_OOM(rv, msg_base->destination = util_safe_strcpy("iotserver"));
    }
    if (msg_src->base->sender) {
        GOTO_ERR_OOM(rv, msg_base->sender = util_safe_strcpy(msg_src->base->sender));
    } else {
        msg_base->sender = NULL;
    }
    if (msg_src->base->source) {
        GOTO_ERR_OOM(rv, msg_base->source = util_safe_strcpy(msg_src->base->source));
    } else {
        msg_base->source = NULL;
    }

    msg->base = msg_base;
    for (i = 0; i < attr_num; i++) {
        if (msg_src->u.data.items_desc[i].type == IOTCS_VALUE_TYPE_STRING) {
            if (msg_src->u.data.items_value[i].string_value) {
                GOTO_ERR_OOM(rv, values[i].string_value = util_safe_strcpy(msg_src->u.data.items_value[i].string_value));
            } else {
                GOTO_ERR_OOM(rv, values[i].string_value = util_safe_strcpy(""));
            }
#if defined IOTCS_STORAGE_SUPPORT
        } else if (msg_src->u.data.items_desc[i].type == IOTCS_VALUE_TYPE_URI) {
            values[i].uri_object = msg_src->u.data.items_value[i].uri_object;
			msg_src->u.data.items_value[i].uri_object->ref_cnt++;
#endif
        } else if (msg_src->u.data.items_desc[i].type == IOTCS_VALUE_TYPE_INT) {
            values[i].int_value = msg_src->u.data.items_value[i].int_value;
        } else if (msg_src->u.data.items_desc[i].type == IOTCS_VALUE_TYPE_NUMBER) {
            values[i].number_value = msg_src->u.data.items_value[i].number_value;
        } else {
            values[i] = msg_src->u.data.items_value[i];
        }

        /* set description */
        items_desc[i].key = util_safe_strcpy(msg_src->u.data.items_desc[i].key);
        items_desc[i].type = msg_src->u.data.items_desc[i].type;
    }
    items_desc[i].key = NULL;
    items_desc[i].type = IOTCS_VALUE_TYPE_NONE;
    GOTO_ERR_OOM(rv, data_base = (iotcs_data_message_base*)util_malloc(sizeof(iotcs_data_message_base)));
    memset(data_base, 0, sizeof (iotcs_data_message_base));
    if (msg_src->u.data.base->format) {
        GOTO_ERR_OOM(rv, data_base->format = util_safe_strcpy(msg_src->u.data.base->format));
    } else {
        GOTO_ERR_OOM(rv, data_base->format = util_safe_strcpy(""));
    }
    msg->u.data.base = data_base;
    msg->u.data.items_value = values;
    msg->u.data.items_desc = items_desc;
    msg->message_mark = (void*)IOTCS_OFFERED_MARK;
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
    msg->error_callback = msg_src->error_callback;
    msg->error_callback_cpp = msg_src->error_callback_cpp;
#endif
    *msg_dst = msg;

    return IOTCS_RESULT_OK;
    error:
    msg->base = msg_base;
    msg->u.data.base = data_base;
    msg->u.data.items_value = values;
    msg->u.data.items_desc = items_desc;
    release_msg_copy(msg);
    return rv;
}

void release_msg_copy(iotcs_message *msg) {
    int i;
    int attr_num = 0;
    iotcs_data_item_desc* items_desc;
    iotcs_value *values;
    if (!msg || msg->base->type != IOTCS_MESSAGE_DATA) {
        return;
    }

    if (msg->base->type == IOTCS_MESSAGE_DATA) {
        values = msg->u.data.items_value;
        items_desc = msg->u.data.items_desc;
        i = 0;
        if (values && items_desc) {
            while (items_desc[i].type != IOTCS_VALUE_TYPE_NONE) {
                i++;
            }
        }
        attr_num = i;
    }

    items_desc = msg->u.data.items_desc;
    values = msg->u.data.items_value;


    for (i = 0; i < attr_num; i++) {
        if (msg->u.data.items_desc[i].type == IOTCS_VALUE_TYPE_STRING) {
            util_free((void*)values[i].string_value);
        }
#if defined IOTCS_STORAGE_SUPPORT
        else if (msg->u.data.items_desc[i].type == IOTCS_VALUE_TYPE_URI) {
            dm_free_uri_object((void*)values[i].uri_object);
        }
#endif
        util_free((void*)items_desc[i].key);
    }
    util_free((void*)msg->base->destination);
    util_free((void*)msg->base->sender);
    util_free((void*)msg->base->source);
    util_free((void*)msg->base);
    util_free((void*)msg->u.data.base->format);
    util_free((void*)msg->u.data.base);
    util_free((void*)items_desc);
    util_free((void*)values);
}

#ifdef IOTCS_IMPLICIT_EDGE_COMPUTING
iotcs_result iotcs_offer(iotcs_message *message) {
    iotcs_result result = IOTCS_RESULT_FAIL;
    int has_policy = 0;
    iotcs_message *msg;
    iotcs_data_item_desc* items_desc;
    iotcs_value *values;

    if (!message) {
        return IOTCS_RESULT_INVALID_ARGUMENT;
    }

    if (message->base->type != IOTCS_MESSAGE_DATA) {
        return IOTCS_RESULT_FAIL;
    }

    if (message->base->type == IOTCS_MESSAGE_DATA) {
        values = message->u.data.items_value;
        items_desc = message->u.data.items_desc;
        if (!values || !items_desc) {
            return IOTCS_RESULT_FAIL;
        }
    } else {
        return iotcs_port_message_queue_put(g_send_queue, message, -1);
    }
    if (copy_msg(message, &msg) == IOTCS_RESULT_OK) {
        if (!apply_policy(NULL, msg)) {
            result = iotcs_port_message_queue_put(g_send_queue, msg, -1);
            return result;
        }
        return IOTCS_RESULT_OK;
    }

    return IOTCS_RESULT_FAIL;
}
#endif


#ifdef IOTCS_MESSAGE_PERSISTENCE
extern int g_message_uid;

void guaranted_serialize_and_save(iotcs_message *message) {
    UTIL_LOCK_LL();
    char *tmp_message_buffer = util_get_payload_buffer();
    char *endpoint = tam_get_endpoint_id();
    int written = 0;
    written = msg_toJSON(message, tmp_message_buffer,
            UTIL_PAYLOAD_BUFFER_SIZE);
    message->uid = g_message_uid++;
    iotcs_message_persistence_save_message(message, tmp_message_buffer, endpoint);
    UTIL_UNLOCK_LL();
}
#endif

iotcs_result iotcs_message_dispatcher_queue(iotcs_message *message) {
    iotcs_result result = IOTCS_RESULT_FAIL;
    iotcs_message *msg = NULL;
    if (!message) {
        return IOTCS_RESULT_INVALID_ARGUMENT;
    }
    if (message->base->type == IOTCS_MESSAGE_DATA) {
        if ((result = copy_msg(message, &msg)) == IOTCS_RESULT_OK) {
            result = iotcs_port_message_queue_put(g_send_queue, msg, -1);
#ifdef IOTCS_MESSAGE_PERSISTENCE
            if (message->base && message->base->reliability == IOTCS_MESSAGE_RELIABILITY_GUARANTED_DELIVERY &&
                result != IOTCS_RESULT_OK) {
                guaranted_serialize_and_save(message);
            }
#endif
#ifndef IOTCS_VIRTUALIZATION_SUPPORT
            if (result != IOTCS_RESULT_OK) {
                if (g_error_cb) {
                    (*g_error_cb)(message, result, "Can not put message to queue");
                }
                release_msg_copy(msg);
            }
#endif
        }
        util_free(msg);
    } else if (message->base->type == IOTCS_MESSAGE_ALERT) {
        if ((result = copy_alert_msg(message, &msg)) == IOTCS_RESULT_OK) {
            result = iotcs_port_message_queue_put(g_send_queue, msg, -1);
#ifdef IOTCS_MESSAGE_PERSISTENCE
            if (message->base && message->base->reliability == IOTCS_MESSAGE_RELIABILITY_GUARANTED_DELIVERY &&
                result != IOTCS_RESULT_OK) {
                guaranted_serialize_and_save(message);
            }
#endif
#ifndef IOTCS_VIRTUALIZATION_SUPPORT
            if (result != IOTCS_RESULT_OK) {
                if (g_error_cb) {
                    (*g_error_cb)(message, result, "Can not put message to queue");
                }
                release_alert_msg_copy(msg);
            }
#endif
        }
        util_free(msg);
    } else {
        result = iotcs_port_message_queue_put(g_send_queue, message, -1);
    }

    return result;
}

void cl_message_dispatcher_counters_add_bytes_sent(int64_t value) {
    if (value > 0) {
        g_message_dispatcher_counters.total_bytes_sent += value;
    }
}

void cl_message_dispatcher_counters_add_bytes_received(int64_t value) {
    if (value > 0) {
        g_message_dispatcher_counters.total_bytes_received += value;
    }
}

int64_t cl_message_dispatcher_get_polling_interval(void) {
    return g_polling_timeout_ms == MSG_INF_POLLING_INTERVAL ? 0 : g_polling_timeout_ms;
}

void cl_message_dispatcher_set_polling_interval(int64_t value) {
    g_polling_timeout_ms = value > 0 ? value : MSG_INF_POLLING_INTERVAL;
    iotcs_port_message_queue_put(g_send_queue, &UPDATE_POLLING_INTERVAL_MESSAGE, -1);
}

cl_message_dispatcher_counters* cl_message_dispatcher_counters_get(void) {
    g_message_dispatcher_counters.load = ((float) iotcs_port_message_queue_count(g_send_queue)) / ((float) IOTCS_PORT_MESSAGE_QUEUE_SIZE);
    return &g_message_dispatcher_counters;
}

void cl_message_dispatcher_counters_reset() {
    g_message_dispatcher_counters.total_bytes_received = 0;
    g_message_dispatcher_counters.total_bytes_sent = 0;
    g_message_dispatcher_counters.total_messages_received = 0;
    g_message_dispatcher_counters.total_messages_retried = 0;
    g_message_dispatcher_counters.total_messages_sent = 0;
    g_message_dispatcher_counters.total_protocols_errors = 0;
}

static void send_all(iotcs_message *message, int count) {
    int posted = 0, iteration_counter = 0;
    int first = 0;
    cl_fail_string_desc desc = {g_async_send_fail_reason_buf, IOTCS_ASYNC_FAIL_REASON_BUFFER_SIZE};

    if (!message || count <= 0) {
        return;
    }

    while (count > posted) {
        iotcs_result current_result = cl_internal_send(message, count, &posted, &desc);
        if (current_result == IOTCS_RESULT_OK) {
            g_message_dispatcher_counters.total_messages_sent += posted;
        } else {
            g_message_dispatcher_counters.total_protocols_errors += posted;
        }
        // If number of posted messages is zero in current iteration
        if (first == posted) {
            iteration_counter++;
        } else {
            iteration_counter = 0;
        }
        // If there was 5 iterations in a row without posted messages then return.
        if (iteration_counter > MAXIMUM_ITERATION_WITHOUT_PROGRESS) {
            posted = count;
            current_result = IOTCS_RESULT_FAIL;
        }

        int i;
        for (i = first; i < posted; i++) {
            if (i >= IOTCS_MAX_MESSAGES_FOR_SEND) {
                break;
            }
            if (current_result == IOTCS_RESULT_OK) {
                if (message[i].base->type && message[i].base->type == IOTCS_MESSAGE_RESPONSE) {
                    if (message[i].u.response.request) {
                        cl_internal_request_message_free_unsafe(message[i].u.response.request);
                    }
                } else if (g_delivery_cb) {
                    (*g_delivery_cb)(&message[i]);
                }
#ifndef IOTCS_VIRTUALIZATION_SUPPORT
                if (message[i].base->type == IOTCS_MESSAGE_DATA && message[i].message_mark == IOTCS_OFFERED_MARK) {
                    release_msg_copy(&message[i]);
                } else if (message[i].base->type == IOTCS_MESSAGE_ALERT && message[i].message_mark == IOTCS_OFFERED_ALERT_MARK) {
                    release_alert_msg_copy(&message[i]);
                }
#endif
            } else {
                if (message[i].base->type && message[i].base->type == IOTCS_MESSAGE_RESPONSE) {
                    if (message[i].u.response.request) {
                        cl_internal_request_message_free_unsafe(message[i].u.response.request);
                    }
                } else if (g_error_cb) {
                    (*g_error_cb)(&message[i], current_result, g_async_send_fail_reason_buf);
                }
#ifndef IOTCS_VIRTUALIZATION_SUPPORT
                if (message[i].base->type == IOTCS_MESSAGE_DATA && message[i].message_mark == IOTCS_OFFERED_MARK) {
                    release_msg_copy(&message[i]);
                } else if (message[i].base->type == IOTCS_MESSAGE_ALERT && message[i].message_mark == IOTCS_OFFERED_ALERT_MARK) {
                    release_alert_msg_copy(&message[i]);
                }
#endif
            }
        }
        first = posted;
    };
#ifdef IOTCS_MESSAGE_PERSISTENCE
    iotcs_batch_by_clear_messages();
#endif
}

iotcs_bool is_ready_to_send(int count, int64_t timer) {
    if (g_batch_message_size && count < g_batch_message_size) {
        return IOTCS_FALSE;
    }
    if (g_batch_message_time && (timer + g_batch_message_time - iotcs_port_get_current_time_millis() > 0)) {
        return IOTCS_FALSE;
    }
    return IOTCS_TRUE;
}

iotcs_bool is_need_to_persist(int count) {
    if (count < IOTCS_MAX_MESSAGES_FOR_SEND) {
        return IOTCS_FALSE;
    } else {
        return IOTCS_TRUE;
    }
}

iotcs_bool is_expected_alert(iotcs_message *message) {
    if (message->base && message->base->type == IOTCS_MESSAGE_ALERT && message->u.alert.base->severity_level <= g_expected_alert_severity) {
        return IOTCS_TRUE;
    } else {
        return IOTCS_FALSE;
    }
}

#ifdef IOTCS_MESSAGE_PERSISTENCE
void batch_serialize_and_save(iotcs_message *message) {
    UTIL_LOCK_LL();
    char *tmp_message_buffer = util_get_payload_buffer();
    int written = 0;
    written = msg_toJSON(message, tmp_message_buffer,
            UTIL_PAYLOAD_BUFFER_SIZE);
    message->uid = g_message_uid++;
    iotcs_batch_by_save_message(message, tmp_message_buffer);
    UTIL_UNLOCK_LL();
}
#endif

void* messaging_send_thread(void *arg) {
    (void) arg;
    int stop_message_received = 0;
    iotcs_message message[IOTCS_MAX_MESSAGES_FOR_SEND];
    iotcs_message *tmp = NULL;
    iotcs_message cur_msg;
    iotcs_result result;
    iotcs_bool need_to_send = IOTCS_FALSE;
    int64_t timeout = 0;
    int64_t timer = iotcs_port_get_current_time_millis();

    set_settle_time();

    while (!stop_message_received) {
        int count = 0;
        need_to_send = IOTCS_FALSE;
        result = iotcs_port_message_queue_get(g_send_queue, &message[count],
                g_polling_timeout_ms < INT_MAX ? (int) g_polling_timeout_ms : INT_MAX);
        tmp = &message[count];
        if (g_batch_message_size) {
            timeout = 10000;
        }
        if (g_batch_message_time) {
            timeout = timer + g_batch_message_time - iotcs_port_get_current_time_millis();
            if (timeout < 0) {
                timeout = 0;
            }
        }
        if (tmp->base == (iotcs_message_base*)STOP_MESSAGE_BASE) {
            stop_message_received = 1;
            break;
        }
        if (result == IOTCS_RESULT_OK && tmp->base != (iotcs_message_base*)STOP_MESSAGE_BASE) {
            count++;
            do {
                if (tmp->base == (iotcs_message_base*)STOP_MESSAGE_BASE) {
                    stop_message_received = 1;
                    break;
                }

                /*if update polling interval message is not first in a chain then just send all previous messages*/
                if (!g_batch_message_size && !g_batch_message_time && tmp->base == (iotcs_message_base*)UPDATE_POLLING_INTERVAL_MESSAGE_BASE) {
                    break;
                }

#ifdef IOTCS_MESSAGE_PERSISTENCE
                if (is_need_to_persist(count)) {
                    result = iotcs_port_message_queue_get(g_send_queue, &cur_msg, timeout);
                    if (IOTCS_RESULT_OK == result) {
                        tmp = &cur_msg;
                        if (tmp->base == (iotcs_message_base*)STOP_MESSAGE_BASE) {
                            stop_message_received = 1;
                            break;
                        }

                        /*if update polling interval message is not first in a chain then just send all previous messages*/
                        if (!g_batch_message_size && !g_batch_message_time && tmp->base == (iotcs_message_base*)UPDATE_POLLING_INTERVAL_MESSAGE_BASE) {
                            break;
                        }
                        count++;
                        if (tmp->base && tmp->base->reliability == IOTCS_MESSAGE_RELIABILITY_GUARANTED_DELIVERY &&
                            (tmp->base->type == IOTCS_MESSAGE_DATA ||
                             tmp->base->type == IOTCS_MESSAGE_ALERT)) {
                            guaranted_serialize_and_save(tmp);
                        } else {
                            batch_serialize_and_save(tmp);
                            if (tmp->base && tmp->base->type == IOTCS_MESSAGE_RESPONSE) {
                                cl_internal_request_message_free_unsafe(tmp->u.response.request);
                                tmp->u.response.request = NULL;
                            }
                        }
                        need_to_send = is_expected_alert(tmp);
 
                        if (tmp->base->type == IOTCS_MESSAGE_DATA && tmp->message_mark == IOTCS_OFFERED_MARK) {
                            release_msg_copy(tmp);
                        } else if (tmp->base->type == IOTCS_MESSAGE_ALERT && tmp->message_mark == IOTCS_OFFERED_ALERT_MARK) {
                            release_alert_msg_copy(tmp);
                        } else if (g_delivery_cb) {
                            (*g_delivery_cb)(tmp);
                        }
                    } else {
                        break;
                    }
                } else {
#endif
                    result = iotcs_port_message_queue_get(g_send_queue, &message[count], timeout);
                    if (IOTCS_RESULT_OK == result) {
                        tmp = &message[count];
                        if (tmp->base == (iotcs_message_base*)STOP_MESSAGE_BASE) {
                            stop_message_received = 1;
                            break;
                        }

                        /*if update polling interval message is not first in a chain then just send all previous messages*/
                        if (!g_batch_message_size && !g_batch_message_time && tmp->base == (iotcs_message_base*)UPDATE_POLLING_INTERVAL_MESSAGE_BASE) {
                            break;
                        }
                        count++;
                        need_to_send = is_expected_alert(tmp);
                        message[count].uid = 0;
                    } else {
                        break;
                    }
#ifdef IOTCS_MESSAGE_PERSISTENCE
                }
#endif
            } while (!is_ready_to_send(count, timer) &&
                     !need_to_send &&
                    IOTCS_RESULT_OK == result);
            timer = iotcs_port_get_current_time_millis();
            send_all(&message[0], count);
        } else {
#ifdef IOTCS_LONG_POLLING
            if (!g_use_long_polling)
#endif
            {/* No messages to send - go to server and ask it one time */
                cl_get_requests_messages();
            }
        }
    }

#ifdef IOTCS_LONG_POLLING    
    UTIL_LOCK_LP();
    cl_long_polling_receive_should_be_stopped();
    UTIL_UNLOCK_LP();
#endif
    cl_put_special_request_message(MSG_STOP_RECEIVE_REQUEST, -1);

    if (g_recv_thread != NULL) {
        iotcs_port_thread_join(g_recv_thread);
    }

    // Then finalize send thread
    while (IOTCS_RESULT_OK == iotcs_port_message_queue_get(g_send_queue, &message[0], 0)) {
        send_all(&message[0], 1);
    }

#ifndef IOTCS_LONG_POLLING
    /* Free all received messages */
    {
        iotcs_request_message *request;
        while (NULL != (request = cl_internal_receive(0))) {
            g_message_dispatcher_counters.total_messages_received++;
            cl_internal_request_message_free(request);
        }
    }
#endif

    iotcs_port_thread_cleanup();

    return NULL;
}

void* messaging_recv_thread(void *arg) {
    (void) arg;
    iotcs_request_message *request;
#if (IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE <= 1)
    iotcs_message message;
#endif

    while (1) {
#ifndef IOTCS_LONG_POLLING
        request = cl_get_request_message_blocking(-1);

        if (request == MSG_STOP_RECEIVE_REQUEST) {
            break;
        }
#else
        if (cl_is_long_polling_receive_should_be_stopped()) {
            break;
        }

        if (g_use_long_polling) {
            request = cl_get_request_message_blocking(0);
        } else {
            request = cl_get_request_message_blocking(-1);
        }

        if (request == MSG_STOP_RECEIVE_REQUEST) {
            break;
        }
#endif

        if (request != NULL) {
            /* Process all received messages */
            g_message_dispatcher_counters.total_messages_received++;

            if (IOTCS_RESULT_OK == device_model_process_capability_request(request)) {
                continue; /* it was a capability request handled by library */
            }
#ifdef IOTCS_VIRTUALIZATION_SUPPORT            
  #if  defined(IOTCS_MESSAGE_DISPATCHER) && (IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE > 1)
           /* dispatch the request using one of request handler thread pool threads */
           cl_internal_request_handler_pool_dispatcher_dispatch(cl_internal_dispatch_request,request);
  #else
           cl_internal_dispatch_request(request, &message);
  #endif   
#else 
  #if  defined(IOTCS_MESSAGE_DISPATCHER) && (IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE > 1)
           cl_internal_request_handler_pool_dispatcher_dispatch(cl_internal_request_dispatcher_dispatch,request);
  #else
           cl_internal_request_dispatcher_dispatch(request, &message);
  #endif
#endif

#if (IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE <= 1)
           cl_internal_async_message_dispatcher_process_response_message(request, &message);
#endif
        }
#ifdef IOTCS_LONG_POLLING 
        else if (g_use_long_polling) {
            char* response = NULL;
            if (IOTCS_RESULT_OK == cl_post_long_polling_request(&response, NULL, IOTCS_LP_BIG_TIMEOUT_MS)) {
                if (response) {
                    cl_process_message_from_json(response);
                }
            }
        }
#endif
    }

    iotcs_port_thread_cleanup();

    return NULL;
}

void cl_internal_async_message_dispatcher_process_response_message(iotcs_request_message *request, iotcs_message* message) {
    //if settle time isn't passed and request wasn't processed then try to put it into receive queue
    //we will try to hold 1 slot for it until is_settle_time_passed() == IOTCS_FALSE 
    if (message->u.response.status_code == PROTOCOL_RESPONSE_CODE_NOT_FOUND &&
            cl_is_settle_time_passed() == IOTCS_FALSE &&
            strstr(message->u.response.request->url, "manage/resources/") == NULL) {
        cl_put_special_request_message(request, 0);
    } else if (strcmp(request->source, INTERNAL_REQUEST_SOURCE) == 0) {
        //Do nothing
    } else {
        iotcs_port_message_queue_put(g_send_queue, message, -1);
    }
}

iotcs_result cl_internal_async_message_dispatcher_init(iotcs_message_dispatcher_delivery_callback delivery_cb,
        iotcs_message_dispatcher_error_callback error_cb, int polling_timeout_ms, iotcs_bool long_polling_request_required) {

    g_polling_timeout_ms = polling_timeout_ms;
    g_delivery_cb = delivery_cb;
    g_error_cb = error_cb;
#ifdef IOTCS_LONG_POLLING
    g_use_long_polling = long_polling_request_required;
#endif

    if (NULL == (g_send_queue = iotcs_port_message_queue_create())) {
        return IOTCS_RESULT_FAIL;
    }

    if (NULL == (g_send_thread = iotcs_port_thread_create(messaging_send_thread))) {
        iotcs_port_message_queue_destroy(g_send_queue);
        return IOTCS_RESULT_FAIL;
    }

    if (NULL == (g_recv_thread = iotcs_port_thread_create(messaging_recv_thread))) {
        iotcs_port_message_queue_put(g_send_queue, &STOP_MESSAGE, -1);
        iotcs_port_thread_join(g_send_thread);
        iotcs_port_message_queue_destroy(g_send_queue);
        return IOTCS_RESULT_FAIL;
    }

    return IOTCS_RESULT_OK;
}

void cl_internal_async_message_dispatcher_finalize(void) {
    iotcs_port_message_queue_put(g_send_queue, &STOP_MESSAGE, -1);
    iotcs_port_thread_join(g_send_thread);
    iotcs_port_message_queue_destroy(g_send_queue);
}
