/*
 * Copyright (c) 2017, 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 <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <ctype.h>
#include <stdarg.h>
#include <time.h>
#include <math.h>
#include <limits.h>
#include <assert.h>
#include "device_model/device_model_private.h"
#include "iotcs_device.h"
#include "util/util.h"
#include "json/json_helper.h"
#include "protocol/http/http_wrapper.h"
#include "iotcs/iotcs_private.h"
#include "trusted_assets_manager/iotcs_tam.h"
#include "advanced/iotcs_messaging.h"
#include "messaging/msg_private.h"
#include "iotcs_port_system.h"
#include "util/util_memory.h"
#include "util/util_thread_private.h"
#include "util/util_buffer.h"
#include "util/util.h"
#include "iotcs_port_thread.h"
#include "policy/device_policy.h"
#include "iotcs_port_queue.h"
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
#include "iotcs_virtual_device.h"
int get_action_idx(iotcs_device_model_handle_t* hmodel, const char *name);
#endif

#include "iotcs_port_persistence.h"

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

typedef struct endpoint_info_t {
    char *endpoint_id;
    char *device_model;
    char *policy_id;
    struct endpoint_info_t *next;
} endpoint_info;

static endpoint_info* g_endpoint_list;

static iotcs_port_thread g_policy_thread = NULL;
iotcs_policy_handle_t* g_policy_list = NULL;

iotcs_port_mutex iotcs_policy_mutex;
iotcs_window_handle_t window_array[WINDOWS_COUNT];

extern int g_batch_message_size;
extern int64_t g_batch_message_time;
extern iotcs_message_severity g_expected_alert_severity;

#define MAXOPSTACK 64
#define MAXNUMSTACK 64
#define EPSILON 0.000001

#define MAXOPSTACK 64
#define MAXNUMSTACK 64
#define EPSILON 0.000001
#define STANDARD_DEVIATION_SIZE 100
 

iotcs_bool apply_pipe(iotcs_pipeline_function_list_t *cur_pipe, char *attribute_name, iotcs_virtual_device_handle_t* virtual_device_handle, iotcs_message *message);
float get_float(char *tstart, iotcs_virtual_device_handle_t *device_handle, iotcs_message *msg);

float eval_uminus(float a1, float a2)
{
  return -a1;
}

float eval_big(float a1, float a2) {
    if (a1 > a2)
        return 1;
    else
        return 0;
}
float eval_big_eq(float a1, float a2) {
    if (a1 >= a2)
        return 1;
    else
        if (a2 - a1 < EPSILON) {
            return 1;
        }
    return 0;
}
float eval_less(float a1, float a2) {
    if (a1 < a2)
        return 1;
    else
        return 0;
}
float eval_less_eq(float a1, float a2) {
    if (a1 <= a2)
        return 1;
    else
        if (a1 - a2 < EPSILON) {
            return 1;
        }
    return 0;
}
float eval_eq(float a1, float a2) {
    if (a1 - a2 < EPSILON && a2 - a1 < EPSILON)
        return 1;
    else
        return 0;
}
float eval_mul(float a1, float a2)
{
    return a1*a2;
}
float eval_div(float a1, float a2)
{
    if (!a2) {
        LOG_ERRS("ERROR: Division by zeros");
        exit(EXIT_FAILURE);
    }
    return a1/a2;
}
float eval_add(float a1, float a2)
{
    return a1+a2;
}
float eval_sub(float a1, float a2)
{
    return a1-a2;
}

float eval_or(float a1, float a2)
{
    return a1||a2;
}
float eval_and(float a1, float a2)
{
    return a1&&a2;
}

enum {ASSOC_NONE=0, ASSOC_LEFT, ASSOC_RIGHT};

struct op_s {
  char op;
  int prec;
  int assoc;
  int unary;
  float (*eval)(float a1, float a2);
} ops[]={
  {'_', 10, ASSOC_RIGHT, 1, eval_uminus},
  {'*', 8, ASSOC_LEFT, 0, eval_mul},
  {'/', 8, ASSOC_LEFT, 0, eval_div},
  {'+', 5, ASSOC_LEFT, 0, eval_add},
  {'-', 5, ASSOC_LEFT, 0, eval_sub},
  {'|', 5, ASSOC_LEFT, 0, eval_or},
  {'&', 5, ASSOC_LEFT, 0, eval_and},
  {'>', 4, ASSOC_LEFT, 0, eval_big},
  {'}', 4, ASSOC_LEFT, 0, eval_big_eq},
  {'<', 4, ASSOC_LEFT, 0, eval_less},
  {'{', 4, ASSOC_LEFT, 0, eval_less_eq},
  {'=', 4, ASSOC_LEFT, 0, eval_eq},
  {'(', 0, ASSOC_NONE, 0, NULL},
  {')', 0, ASSOC_NONE, 0, NULL}
};

struct op_s *getop(char ch) {
    int i;
    for (i=0; i<sizeof ops/sizeof ops[0]; ++i) {
        if (ops[i].op==ch) return ops+i;
    }
    return NULL;
}

struct op_s *opstack[MAXOPSTACK];
int nopstack=0;

float numstack[MAXNUMSTACK];
int nnumstack=0;

void push_opstack(struct op_s *op)
{
    if (nopstack>MAXOPSTACK-1) {
        LOG_ERRS("Formula: Operator stack overflow");
        return;
    }
    opstack[nopstack++]=op;
}

struct op_s *pop_opstack()
{
    if (!nopstack) {
        LOG_ERRS("Formula: Operator stack empty");
        return NULL;
    }
    return opstack[--nopstack];
}

void push_numstack(float num)
{
    if (nnumstack>MAXNUMSTACK-1) {
        LOG_ERRS("Formula: Number stack overflow");
        return;
    }
    numstack[nnumstack++]=num;
}

int pop_numstack()
{
    if (!nnumstack) {
        LOG_ERRS("Formula: Number stack empty");
        return 0;
    }
    return numstack[--nnumstack];
}


void shunt_op(struct op_s *op){
    struct op_s *pop;
    int n1, n2;
    if (op->op=='(') {
        push_opstack(op);
        return;
    } else if (op->op==')') {
        while(nopstack>0 && opstack[nopstack-1]->op!='(') {
            pop=pop_opstack();
            n1=pop_numstack();
            if (pop->unary) push_numstack(pop->eval(n1, 0));
            else {
                n2=pop_numstack();
                push_numstack(pop->eval(n2, n1));
            }
        }
         if (!(pop=pop_opstack()) || pop->op!='(') {
            LOG_ERRS("Formula: Stack error. No matching \'(\'");
            return;
        }
        return;
    }

    if (op->assoc==ASSOC_RIGHT) {
        while(nopstack && op->prec<opstack[nopstack-1]->prec) {
            pop=pop_opstack();
            n1=pop_numstack();
            if (pop->unary) push_numstack(pop->eval(n1, 0));
            else {
                n2=pop_numstack();
                push_numstack(pop->eval(n2, n1));
            }
        }
    } else {
        while(nopstack && op->prec<=opstack[nopstack-1]->prec) {
            pop=pop_opstack();
            n1=pop_numstack();
            if (pop->unary) push_numstack(pop->eval(n1, 0));
            else {
                n2=pop_numstack();
                push_numstack(pop->eval(n2, n1));
            }
        }
    }
    push_opstack(op);
}

float calculateSub(char *expr, iotcs_virtual_device_handle_t* device_handle, iotcs_message *msg) {
    char *tstart=NULL;
    struct op_s startop={'X', 0, ASSOC_NONE, 0, NULL};
    struct op_s *op=NULL;
    int n1, n2;
    struct op_s *lastop=&startop;
    nopstack=0;
    nnumstack=0;

    for(; *expr ; ++expr) {
        if (!tstart) {
            if ((op = getop(*expr))) {
                if  (lastop && (lastop==&startop || lastop->op!=')')) {
                    if (op->op=='-') op=getop('_');
                    else if (op->op!='(') {
                        LOG_ERR("Formula: Stack error. No matching \'(\' at %c %c", *(expr -1), *expr);
                        return 0;
                    }
                }
                shunt_op(op);
                lastop=op;
            } else if (isdigit(*expr) || *expr == '.' || *expr == '$' ) {
                tstart=expr;
            } else if (!isspace(*expr)) {
                if (*expr == '0') break;
                LOG_ERR("Formula: Syntax error at %c %c", *(expr -1), *expr);
                return 0;
            }
        } else {
            if (isspace(*expr) || *expr == '.' || *expr == '$' || (*expr == '(' && *(expr-1) == '$')) {
                push_numstack(get_float(tstart, device_handle, msg));
                if (*expr == '$'|| (*expr == '(' && *(expr-1) == '$')) {
                    while (*expr != ')' && *expr != '\0') {
                        expr++;
                    }
                }
                if (*expr == '\0')
                    expr -= 1;
                tstart=NULL;
                lastop=NULL;
            } else if ((op=getop(*expr))) {
                push_numstack(get_float(tstart, device_handle, msg));
                if (*expr == '<' && *(expr+1) == '=') {
                    op=getop('{');
                    expr++;
                }
                if (*expr == '>' && *(expr+1) == '=') {
                    op=getop('}');
                    expr++;
                }
                if (*expr == '=' && *(expr+1) == '=') {
                    op=getop('=');
                    expr++;
                }
                if ((*expr == 'O' && *(expr+1) == 'R') ||
                    (*expr == '|' && *(expr+1) == '|')) {
                    op=getop('|');
                    expr++;
                }
                if ((*expr == 'A' && *(expr+1) == 'N' && *(expr+1) == 'D')  ||
                    (*expr == '&' && *(expr+1) == '&')) {
                    op=getop('&');
                    expr++;
                    if (*expr == 'D') expr++;
                }
                if (*tstart == '$'|| (*expr == '(' && *(expr-1) == '$')) {
                    while (*expr != ')' && *expr != '\0') {
                        expr++;
                    }
                }
                if (*expr == '\0')
                    expr -= 1;
                tstart=NULL;
                shunt_op(op);
                lastop=op;
            } else if (!isalnum(*expr) && *expr != ',' && *expr != '$') {
                LOG_ERR("Formula: Syntax error at %c %c", *(expr -1), *expr);
                return 0;
            }
        }
    }
    if (tstart) push_numstack(get_float(tstart, device_handle, msg));

    while(nopstack) {
        op=pop_opstack();
        n1=pop_numstack();
        if (op->unary) push_numstack(op->eval(n1, 0));
        else {
            n2=pop_numstack();
            push_numstack(op->eval(n2, n1));
        }
    }
    if (nnumstack!=1) {
        LOG_ERR("Formula: Number stack has %d elements after evaluation. Should be 1.", nnumstack);
        return 0;
    }

    return numstack[0];
}

float calculate(char *expr, iotcs_virtual_device_handle_t* device_handle, iotcs_message *msg) {
    int i = 0;
    float result;
    char *cond = NULL, *lexpr = NULL, *rexpr = NULL;

    for (; expr[i]; i++) {
        if (expr[i] == '?') {
            cond = util_safe_strncpy(expr, i);
            lexpr = expr + i + 1;
        }
        if (expr[i] == ':' && cond) {
            lexpr = util_safe_strncpy(lexpr, i - (lexpr - expr));
            rexpr = util_safe_strcpy(expr + i + 1);
        }
    }
    if (cond && lexpr && rexpr) {
        if (calculateSub(cond, device_handle, msg) != 0) {
            result = calculateSub(lexpr, device_handle, msg);
        } else {
            result = calculateSub(rexpr, device_handle, msg);
        }
    } else {
        result = calculateSub(expr, device_handle, msg);
    }

    util_free(cond);
    if (rexpr) {
        util_free(lexpr);
        util_free(rexpr);
    }

    return result;
}

float get_float(char *tstart, iotcs_virtual_device_handle_t *device_handle, iotcs_message *msg) {
    char *attribute_name;
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
    int int_value;
    int idx = 0;
#endif
    float result_value = 0;
    iotcs_bool is_current = IOTCS_FALSE;
    int i = 0, j;
    if (*tstart == '$') {
        if (*(tstart+1) == '$') {
            is_current = IOTCS_TRUE;
            tstart++;
        }
        while (tstart[i+2] != ')') {
            i++;
        }
        attribute_name = util_safe_strncpy(tstart+2, i);
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
        idx = get_attribute_idx(device_handle->model_handle, attribute_name);
        if (idx >= 0) {
            if (device_handle->model_handle->attributes.desc[idx].type == IOTCS_VALUE_TYPE_INT) {
                iotcs_virtual_device_get_integer(device_handle, attribute_name, &int_value);
                return (float)int_value;
            } else if (device_handle->model_handle->attributes.desc[idx].type == IOTCS_VALUE_TYPE_NUMBER) {
                iotcs_virtual_device_get_float(device_handle, attribute_name, &result_value);
                return result_value;
            }
        }
#endif
        j = 0;
        if (msg != NULL) {
            while (1) {
                if (msg->base->type == IOTCS_MESSAGE_DATA) {
                    if (msg->u.data.items_desc[j].type == IOTCS_VALUE_TYPE_NONE) {
                        break;
                    }
                    if (strcmp(msg->u.data.items_desc[j].key, attribute_name) == 0) {
                        if (msg->u.data.items_desc[j].type == IOTCS_VALUE_TYPE_NUMBER) {
                            return msg->u.data.items_value[j].number_value;
                        } else if (msg->u.data.items_desc[j].type == IOTCS_VALUE_TYPE_INT) {
                            return (float)msg->u.data.items_value[j].int_value;
                        }
                    }
                } else {
                    if (msg->u.alert.items_desc[j].type == IOTCS_VALUE_TYPE_NONE) {
                        break;
                    }
                    if (strcmp(msg->u.alert.items_desc[j].key, attribute_name) == 0) {
                        if (msg->u.alert.items_desc[j].type == IOTCS_VALUE_TYPE_NUMBER) {
                            return msg->u.alert.items_value[j].number_value;
                        } else if (msg->u.data.items_desc[j].type == IOTCS_VALUE_TYPE_INT) {
                            return (float)msg->u.alert.items_value[j].int_value;
                        }
                    }
                }
                j++;
            }
        }
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
        idx = get_attribute_idx(device_handle->model_handle, attribute_name);
        if (idx >= 0) {
            if (device_handle->model_handle->attributes.desc[idx].type == IOTCS_VALUE_TYPE_INT) {
                iotcs_virtual_device_get_integer(device_handle, attribute_name, &int_value);
                return (float)int_value;
            } else if (device_handle->model_handle->attributes.desc[idx].type == IOTCS_VALUE_TYPE_NUMBER) {
                iotcs_virtual_device_get_float(device_handle, attribute_name, &result_value);
                return result_value;
            }
        }
#endif
        util_free(attribute_name);
        return (float)result_value;
    } else {
        return atof(tstart);
    }
}

/*
 * This function updates policy_id of pair endpoint_id and device_model.
 * endpoint_id or device_model must not be NULL. policy_id may be NULL.
 */
void put_policy_id_for_endpoint(const char *endpoint_id, const char *device_model, const char *policy_id) {
    iotcs_result rv = IOTCS_RESULT_FAIL;
    if (!endpoint_id || !device_model) {
        return;
    }
    endpoint_info *cur_endpoint = g_endpoint_list, *last_endpoint = g_endpoint_list, *new_endpoint = NULL;
    while (cur_endpoint) {
        if (strcmp(cur_endpoint->endpoint_id, endpoint_id) == 0 &&
            strcmp(cur_endpoint->device_model, device_model) == 0) {
            if (cur_endpoint->policy_id) {
                util_free(cur_endpoint->policy_id);
            }
            if (policy_id) {
                GOTO_ERR_OOM(rv, cur_endpoint->policy_id = util_safe_strcpy(policy_id));
            } else {
                cur_endpoint->policy_id = NULL;
            }
            return;
        }
        last_endpoint = cur_endpoint;
        cur_endpoint = cur_endpoint->next;
    }
    GOTO_ERR_OOM(rv, new_endpoint = (endpoint_info*)util_malloc(sizeof(endpoint_info)));
    if (last_endpoint) {
        last_endpoint->next = new_endpoint;
    } else {
        g_endpoint_list = new_endpoint;
    }
    new_endpoint->endpoint_id = NULL;
    new_endpoint->device_model = NULL;
    new_endpoint->policy_id = NULL;
    new_endpoint->next = NULL;
    GOTO_ERR_OOM(rv, new_endpoint->endpoint_id = util_safe_strcpy(endpoint_id));
    GOTO_ERR_OOM(rv, new_endpoint->device_model = util_safe_strcpy(device_model));
    if (policy_id) {
        GOTO_ERR_OOM(rv, new_endpoint->policy_id = util_safe_strcpy(policy_id));
    } else {
        new_endpoint->policy_id = NULL;
    }
    return;
    error:
    if (new_endpoint) {
        util_free(new_endpoint->endpoint_id);
        util_free(new_endpoint->device_model);
        util_free(new_endpoint->policy_id);
        util_free(new_endpoint);
    }
}

/* Return policy_id for endpoint_id with device_model or NULL.
 * if policy_id is NULL then it return NULL too
 */
char* get_policy_id_for_endpoint(const char *endpoint_id, const char *device_model) {
    endpoint_info* cur_endpoint = g_endpoint_list;
    char *result = NULL;
    if (!endpoint_id || !device_model) {
        return NULL;
    }
    while (cur_endpoint) {
        if (strcmp(cur_endpoint->endpoint_id, endpoint_id) == 0 &&
            strcmp(cur_endpoint->device_model, device_model) == 0) {
            result = cur_endpoint->policy_id;
            break;
        }
        cur_endpoint = cur_endpoint->next;
    }
    return result;
}

/* Return endpoint_id for policy_id at position (like [position]). Poisition must be >= 0.
 */
char* get_next_endpoint_by_policy_id(const char *policy_id, int position) {
    endpoint_info* cur_endpoint = g_endpoint_list;
    char *result = NULL;
    if (!policy_id) {
        return NULL;
    }

    while (cur_endpoint) {
        if (cur_endpoint->policy_id && strcmp(cur_endpoint->policy_id, policy_id) == 0) {
            if (position <= 0) {
                result = cur_endpoint->endpoint_id;
                break;
            }
            position--;
        }
        cur_endpoint = cur_endpoint->next;
    }
    return result;
}

/* If list contains endpoint_id and device_model then CL tried to download this policy before.
 * So it should be in the g_policy_list or there are no policy.
 */
iotcs_bool has_policy_id_for_endpoint(const char *endpoint_id, const char *device_model) {
    endpoint_info* cur_endpoint = g_endpoint_list;
    if (!endpoint_id || !device_model) {
        return IOTCS_FALSE;
    }
    while (cur_endpoint) {
        if (strcmp(cur_endpoint->endpoint_id, endpoint_id) == 0 &&
            strcmp(cur_endpoint->device_model, device_model) == 0) {
            return IOTCS_TRUE;
        }
        cur_endpoint = cur_endpoint->next;
    }
    return IOTCS_FALSE;
}

/* Trying to apply policy for message. If IOTCS_VIRTUALIZATION_SUPPORT is switched on then it expects virtual_device_handle.
 * If it is first message for endpoint_id/device_model then it downloads device policy.
 * @return IOTCS_TRUE if there are policy
 */
iotcs_bool apply_policy(iotcs_virtual_device_handle_t* virtual_device_handle, iotcs_message *message) {
    UTIL_POLICY_LOCK();
    iotcs_policy_handle_t *cur_policy = g_policy_list;
    iotcs_pipeline_t *cur_pipeline;
    iotcs_bool has_policy = IOTCS_FALSE;
    int i;
    char *urn;
    char *attribute_name;
    char *m_endpoint_id;
    iotcs_message *msg_copy = NULL;
    if (message->base->type == IOTCS_MESSAGE_DATA) {
        urn = util_safe_strcpy(message->u.data.base->format);
    } else if (message->base->type == IOTCS_MESSAGE_ALERT) {
        urn = util_safe_strcpy(message->u.alert.base->format);
    } else {
        UTIL_POLICY_UNLOCK();
        return IOTCS_FALSE;
    }
    if (message->base->source) {
        m_endpoint_id = message->base->source;
    } else {
        m_endpoint_id = tam_get_endpoint_id();
    }


    for (i = 0; urn[i] != '\0'; i++) { }
    for (; urn[i] != ':'; i--) { }
    urn[i] = '\0';

    if (message->base && message->base->type == IOTCS_MESSAGE_DATA && !has_policy_id_for_endpoint(m_endpoint_id, urn)) {
        /* Mark endpoint_id with urn as known without policy and download policy.
         * If download policy will be with no error then dp_get_policy_unsafe will update policy_id.
         */
        put_policy_id_for_endpoint(m_endpoint_id, urn, NULL);
        UTIL_LOCK_LL();
        /* No policy or error */
        if (dp_get_policy_unsafe(m_endpoint_id, urn) != IOTCS_RESULT_OK) {
            util_free(urn);
            UTIL_UNLOCK_LL();
            UTIL_POLICY_UNLOCK();
            return IOTCS_FALSE;
        }
        UTIL_UNLOCK_LL();
    }

    while (cur_policy) {
        if (strcmp(cur_policy->endpoint_id, m_endpoint_id) == 0) {
            cur_pipeline = cur_policy->pipe_list;
            while (cur_pipeline) {
                i = 0;
                while (1) {
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        if (message->u.data.items_desc[i].type == IOTCS_VALUE_TYPE_NONE) {
                            break;
                        }
                        attribute_name = util_safe_strcpy(message->u.data.items_desc[i].key);
                        if (strcmp(cur_pipeline->attribute, attribute_name) == 0 && strcmp(cur_policy->urn, urn) == 0) {
                            if (cur_pipeline) {
                                copy_msg(message, &msg_copy);
                                has_policy = apply_pipe(cur_pipeline->function_list, attribute_name, virtual_device_handle, msg_copy);
                            }
                        }
                        util_free(attribute_name);
                    } else if (message->base->type == IOTCS_MESSAGE_ALERT) {
                        if (message->u.alert.items_desc[i].type == IOTCS_VALUE_TYPE_NONE) {
                            break;
                        }
                        attribute_name = util_safe_strcpy(message->u.alert.items_desc[i].key);
                        if (strcmp(cur_pipeline->attribute, attribute_name) == 0 && strcmp(cur_policy->urn, urn) == 0) {
                            if (cur_pipeline) {
                                copy_msg(message, &msg_copy);
                                has_policy = apply_pipe(cur_pipeline->function_list, attribute_name, virtual_device_handle, msg_copy) || has_policy;
                            }
                        }
                        util_free(attribute_name);
                    } else {
                        util_free(urn);
                        return IOTCS_FALSE;
                    }
                    i++;
                }
                cur_pipeline = cur_pipeline->next;
            }
        }
        cur_policy = cur_policy->next;
    }
    UTIL_POLICY_UNLOCK();
    util_free(urn);
    return has_policy;
}

const char * DEFAULT_DESTINATION = "iotserver";
const char * DEFAULT_ALERT_DESCRIPTION = "Policy alert";

/* This function is applying pipeline from function cur_function for attribute_name in message. */
iotcs_bool apply_pipe(iotcs_pipeline_function_list_t *cur_function, char *attribute_name, iotcs_virtual_device_handle_t* virtual_device_handle, iotcs_message *message) {
    char *formula;
    double cond, result;
    int k, j, idx;
    iotcs_typed_value value;
    
    k = 0;
    value.type = IOTCS_VALUE_TYPE_NONE;
    while (1) {
        if (message->base->type == IOTCS_MESSAGE_DATA) {
            if (message->u.data.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                break;
            }
            if (strcmp(message->u.data.items_desc[k].key, attribute_name) == 0) {
                value.type = message->u.data.items_desc[k].type;
                if (value.type == IOTCS_VALUE_TYPE_INT) {
                    value.value.int_value = message->u.data.items_value[k].int_value;
                } else if (value.type == IOTCS_VALUE_TYPE_NUMBER) {
                    value.value.number_value = message->u.data.items_value[k].number_value;
                }
                break;
            }
        } else {
            if (message->u.alert.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                break;
            }
            if (strcmp(message->u.alert.items_desc[k].key, attribute_name) == 0) {
                value.type = message->u.alert.items_desc[k].type;
                if (value.type == IOTCS_VALUE_TYPE_INT) {
                    value.value.int_value = message->u.alert.items_value[k].int_value;
                } else if (value.type == IOTCS_VALUE_TYPE_NUMBER) {
                    value.value.number_value = message->u.alert.items_value[k].number_value;
                }
                break;
            }
        }
        k++;
    }

    while (cur_function) {
        switch (cur_function->type) {
            case IOTCS_FUNCTION_TYPE_MEAN:
            case IOTCS_FUNCTION_TYPE_MIN:
            case IOTCS_FUNCTION_TYPE_MAX:
            case IOTCS_FUNCTION_TYPE_STANDARD_DEVIATION:
                dp_handle_math_function(cur_function, virtual_device_handle, attribute_name, message);
                /* This types of policy generate message itself every window */
                return IOTCS_TRUE; 
            case IOTCS_FUNCTION_TYPE_ELIMINATE_DUBL:
                if (dp_handle_duplicates_function(cur_function, virtual_device_handle, attribute_name, message)) {
                    release_msg_copy(message);
                    util_free(message);
                    return IOTCS_TRUE;
                }
                break;
            case IOTCS_FUNCTION_TYPE_DETECT_DUBL:
                if (dp_handle_duplicates_function(cur_function, virtual_device_handle, attribute_name, message)) {
                    iotcs_message_base *alert_msg_base = (iotcs_message_base*)util_malloc(sizeof(iotcs_message_base));
                    iotcs_alert_message_base *alert_base = (iotcs_alert_message_base*)util_malloc(sizeof(iotcs_alert_message_base));

                    alert_msg_base->type = IOTCS_MESSAGE_ALERT;
                    alert_msg_base->reliability = IOTCS_MESSAGE_RELIABILITY_BEST_EFFORT;
                    alert_msg_base->priority = IOTCS_MESSAGE_PRIORITY_HIGHEST;
                    alert_msg_base->destination = util_safe_strcpy(DEFAULT_DESTINATION);
                    alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                    alert_base->format = NULL;
                    alert_base->description = util_safe_strcpy(DEFAULT_ALERT_DESCRIPTION);
                    if (strcmp(cur_function->function.detect_dubl_function.severity, "CRITICAL") == 0) {
                        alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_CRITICAL;
                    } else if (strcmp(cur_function->function.detect_dubl_function.severity, "SIGNIFICANT") == 0) {
                        alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                    } else if (strcmp(cur_function->function.detect_dubl_function.severity, "NORMAL") == 0) {
                        alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_NORMAL;
                    } else if (strcmp(cur_function->function.detect_dubl_function.severity, "LOW") == 0) {
                        alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_LOW;
                    } else {
                        alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                    }
                    alert_base->format = util_safe_strcpy(cur_function->function.detect_dubl_function.format);
                    iotcs_data_item_desc *desc = (iotcs_data_item_desc *)util_malloc(
                        sizeof(iotcs_data_item_desc) * (cur_function->function.detect_dubl_function.fields_len + 1));
                    iotcs_value *values = (iotcs_value *)util_malloc(
                        sizeof(iotcs_value) * (cur_function->function.detect_dubl_function.fields_len + 1));

                    for (j = 0; j < cur_function->function.detect_dubl_function.fields_len; j++) {
                        int length = 0;
                        char * value_name = cur_function->function.detect_dubl_function.fields[j].value;
                        if (*value_name == '$') {
                            while (!isalpha  (*value_name)) {
                                value_name++;
                            }
                            while (*(value_name + length) != '\0' && *(value_name + length) != ')') {
                                length++;
                            }
                            value_name = util_safe_strncpy(value_name, length);
                            k = 0;
                            while (1) {
                                if (message->base->type == IOTCS_MESSAGE_DATA) {
                                    if (message->u.data.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                                        k = -1;
                                        break;
                                    }
                                    if (strcmp(message->u.data.items_desc[k].key, value_name) == 0) {
                                        break;
                                    }
                                } else {
                                    if (message->u.alert.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                                        k = -1;
                                        break;
                                    }
                                    if (strcmp(message->u.alert.items_desc[k].key, value_name) == 0) {
                                        break;
                                    }
                                }
                                k++;
                            }
                            if (k == -1) {
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
                                idx = 0;
                                if ((idx = get_attribute_idx(virtual_device_handle->model_handle, value_name)) < 0) {
                                    LOG_ERR("Can not set value to alert \"%s\" field \"%s\". Field \"%s\"doesn't exist.", 
                                        cur_function->function.detect_dubl_function.format, cur_function->function.alert_condition.fields[j].key, value_name);
#else
                                    LOG_ERR("Can not find attribute %s", value_name);
#endif
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
                                }
                                desc[j].type = virtual_device_handle->model_handle->attributes.desc[idx].type;
                                desc[j].key = util_safe_strcpy(cur_function->function.detect_dubl_function.fields[j].key);
                                if (virtual_device_handle->model_handle->attributes.desc[idx].type == IOTCS_VALUE_TYPE_STRING) {
                                    values[j].string_value = util_safe_strcpy(virtual_device_handle->attributes[idx].value.string_value);
                                } else {
                                    values[j] = virtual_device_handle->attributes[idx].value;
                                }
#endif
                            } else {
                                if (message->base->type == IOTCS_MESSAGE_DATA) {
                                    desc[j].type = message->u.data.items_desc[k].type;
                                    desc[j].key = util_safe_strcpy(cur_function->function.detect_dubl_function.fields[j].key);
                                    if (message->u.data.items_desc[k].type == IOTCS_VALUE_TYPE_STRING) {
                                        values[j].string_value = util_safe_strcpy(message->u.data.items_value[k].string_value);
                                    } else {
                                        values[j] = message->u.data.items_value[k];
                                    }
                                } else {
                                    desc[j].type = message->u.alert.items_desc[k].type;
                                    desc[j].key = util_safe_strcpy(cur_function->function.detect_dubl_function.fields[j].key);
                                    if (message->u.alert.items_desc[k].type == IOTCS_VALUE_TYPE_STRING) {
                                        values[j].string_value = util_safe_strcpy(message->u.alert.items_value[k].string_value);
                                    } else {
                                        values[j] = message->u.alert.items_value[k];
                                    }
                                }
                            }
                        } else if (*value_name == '-' || isdigit(*value_name)) {
                            iotcs_bool is_integer = IOTCS_TRUE;
                            float fraction_part = 0;
                            int integer_part = 0;
                            int fraction_divisor = 1;
                            int is_negative = 1;
                            if (*value_name == '-') {
                                is_negative = -1;
                                value_name++;
                            }

                            while (value_name) {
                                if (isdigit(*value_name)) {
                                    if (!is_integer) {
                                        fraction_part = fraction_part * 10 + (*value_name - '0');
                                        fraction_divisor *= 10;
                                    } else {
                                        integer_part = integer_part * 10 + (*value_name - '0');
                                    }
                                    value_name++;
                                    continue;
                                }
                                if (*value_name == '.') {
                                    if (is_integer) {
                                        is_integer = IOTCS_FALSE;
                                        value_name++;
                                        continue;
                                    } else {
                                        LOG_WARN("Can not parse value of %s : %s", cur_function->function.detect_dubl_function.fields[j].key, cur_function->function.alert_condition.fields[j].value);
                                    }
                                }
                                if (isspace(*value_name) || *value_name == '\0') {
                                    break;
                                }
                                // find something unexpected
                                LOG_WARN("Can not parse value of %s : %s", cur_function->function.detect_dubl_function.fields[j].key, cur_function->function.alert_condition.fields[j].value);
                                break;
                            }
                            if (message->base->type == IOTCS_MESSAGE_DATA) {
                                desc[j].key = util_safe_strcpy(cur_function->function.detect_dubl_function.fields[j].key);
                                if (fraction_part != 0) {
                                    desc[j].type = IOTCS_VALUE_TYPE_NUMBER;
                                    values[j].number_value = (float)is_negative*((float)integer_part + fraction_part / fraction_divisor);
                                } else {
                                    desc[j].type = IOTCS_VALUE_TYPE_INT;
                                    values[j].int_value = is_negative*integer_part;
                                }
                            }
                        } else {
                            if (message->base->type == IOTCS_MESSAGE_DATA) {
                                desc[j].key = util_safe_strcpy(cur_function->function.detect_dubl_function.fields[j].key);
                                desc[j].type = IOTCS_VALUE_TYPE_STRING;
                                values[j].string_value = util_safe_strcpy(value_name);
                            }
                        }
                    }
                    if (message->base->sender) {
                        alert_msg_base->sender = util_safe_strcpy(message->base->sender);
                    } else {
                        alert_msg_base->sender = NULL;
                    }
                    if (message->base->source) {
                        alert_msg_base->source = util_safe_strcpy(message->base->source);
                    } else {
                        alert_msg_base->source = NULL;
                    }
                    desc[j].key = NULL;
                    desc[j].type = IOTCS_VALUE_TYPE_NONE;
                    iotcs_message alert_message = {
                        .id = "",
                        .base = alert_msg_base,
                        .u.alert.base = alert_base,
                        .u.alert.items_desc = desc,
                        .u.alert.items_value = values
                    };
                    alert_message.message_mark = IOTCS_OFFERED_ALERT_MARK;

                    if (iotcs_message_dispatcher_queue(&alert_message) != IOTCS_RESULT_OK) {
                        release_alert_msg_copy(&alert_message);
                    }
                }
                break;
            case IOTCS_FUNCTION_TYPE_SAMPLE_QUALITY:
                if (cur_function->function.sample_quality.rate > 0) {
                    if (cur_function->function.sample_quality.rate != cur_function->function.sample_quality.counter) {
                        cur_function->function.sample_quality.counter++;
                        release_msg_copy(message);
                        util_free(message);
                        return IOTCS_TRUE;
                    } else {
                        cur_function->function.sample_quality.counter = 0;
                    }
                } else if (cur_function->function.sample_quality.rate == IOTCS_SAMPLE_QUALITY_RANDOM) {
                    if (rand() % 100 >= 30) {
                        release_msg_copy(message);
                        util_free(message);
                        return IOTCS_TRUE;
                    }
                } else if (cur_function->function.sample_quality.rate == IOTCS_SAMPLE_QUALITY_NONE) {
                    release_msg_copy(message);
                    util_free(message);
                    return IOTCS_TRUE;
                }
                break;
            case IOTCS_FUNCTION_TYPE_ALERT_CONDITION:
                formula = cur_function->function.alert_condition.condition;
                cond = calculate(formula, virtual_device_handle, message);
                if (cond != 0) {
                    iotcs_message_base *alert_msg_base = (iotcs_message_base*)util_malloc(sizeof(iotcs_message_base));
                    iotcs_alert_message_base *alert_base = (iotcs_alert_message_base*)util_malloc(sizeof(iotcs_alert_message_base));

                    alert_msg_base->type = IOTCS_MESSAGE_ALERT;
                    alert_msg_base->reliability = IOTCS_MESSAGE_RELIABILITY_BEST_EFFORT;
                    alert_msg_base->priority = IOTCS_MESSAGE_PRIORITY_HIGHEST;
                    alert_msg_base->destination = DEFAULT_DESTINATION;
                    alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                    alert_base->format = NULL;
                    alert_base->description = DEFAULT_ALERT_DESCRIPTION;

                    if (strcmp(cur_function->function.alert_condition.severity, "CRITICAL") == 0) {
                        alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_CRITICAL;
                    } else if (strcmp(cur_function->function.alert_condition.severity, "SIGNIFICANT") == 0) {
                        alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                    } else if (strcmp(cur_function->function.alert_condition.severity, "NORMAL") == 0) {
                        alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_NORMAL;
                    } else if (strcmp(cur_function->function.alert_condition.severity, "LOW") == 0) {
                        alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_LOW;
                    } else {
                        alert_base->severity_level = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                    }
                    alert_base->format = util_safe_strcpy(cur_function->function.alert_condition.format);
                    iotcs_data_item_desc *desc = (iotcs_data_item_desc *)util_malloc(
                        sizeof(iotcs_data_item_desc) * (cur_function->function.alert_condition.fields_len + 1));
                    iotcs_value *values = (iotcs_value *)util_malloc(
                        sizeof(iotcs_value) * (cur_function->function.alert_condition.fields_len + 1));

                    for (j = 0; j < cur_function->function.alert_condition.fields_len; j++) {
                        int length = 0;
                        char * value_name = cur_function->function.alert_condition.fields[j].value;
                        if (*value_name == '$') {
                            while (!isalpha  (*value_name)) {
                                value_name++;
                            }
                            while (*(value_name + length) != '\0' && *(value_name + length) != ')') {
                                length++;
                            }
                            value_name = util_safe_strncpy(value_name, length);
                            k = 0;
                            while (1) {
                                if (message->base->type == IOTCS_MESSAGE_DATA) {
                                    if (message->u.data.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                                        k = -1;
                                        break;
                                    }
                                    if (strcmp(message->u.data.items_desc[k].key, value_name) == 0) {
                                        break;
                                    }
                                } else {
                                    if (message->u.alert.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                                        k = -1;
                                        break;
                                    }
                                    if (strcmp(message->u.alert.items_desc[k].key, value_name) == 0) {
                                        break;
                                    }
                                }
                                k++;
                            }
                            if (k == -1) {
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
                                idx = 0;
                                if ((idx = get_attribute_idx(virtual_device_handle->model_handle, value_name)) < 0) {
                                    LOG_ERR("Can not set value to alert \"%s\" field \"%s\". Field \"%s\"doesn't exist.", 
                                        cur_function->function.alert_condition.format, cur_function->function.alert_condition.fields[j].key, value_name);
#else
                                    LOG_ERR("Can not find attribute %s", value_name);
#endif
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
                                }
                                desc[j].type = virtual_device_handle->model_handle->attributes.desc[idx].type;
                                desc[j].key = util_safe_strcpy(cur_function->function.alert_condition.fields[j].key);
                                if (virtual_device_handle->model_handle->attributes.desc[idx].type == IOTCS_VALUE_TYPE_STRING) {
                                    values[j].string_value = util_safe_strcpy(virtual_device_handle->attributes[idx].value.string_value);
                                } else {
                                    values[j] = virtual_device_handle->attributes[idx].value;
                                }
#endif
                            } else {
                                if (message->base->type == IOTCS_MESSAGE_DATA) {
                                    desc[j].type = message->u.data.items_desc[k].type;
                                    desc[j].key = util_safe_strcpy(cur_function->function.alert_condition.fields[j].key);
                                    if (message->u.data.items_desc[k].type == IOTCS_VALUE_TYPE_STRING) {
                                        values[j].string_value = util_safe_strcpy(message->u.data.items_value[k].string_value);
                                    } else {
                                        values[j] = message->u.data.items_value[k];
                                    }
                                } else {
                                    desc[j].type = message->u.alert.items_desc[k].type;
                                    desc[j].key = util_safe_strcpy(cur_function->function.alert_condition.fields[j].key);
                                    if (message->u.alert.items_desc[k].type == IOTCS_VALUE_TYPE_STRING) {
                                        values[j].string_value = util_safe_strcpy(message->u.alert.items_value[k].string_value);
                                    } else {
                                        values[j] = message->u.alert.items_value[k];
                                    }
                                }
                            }
                        } else if (*value_name == '-' || isdigit(*value_name)) {
                            iotcs_bool is_integer = IOTCS_TRUE;
                            float fraction_part = 0;
                            int integer_part = 0;
                            int fraction_divisor = 1;
                            int is_negative = 1;
                            if (*value_name == '-') {
                                is_negative = -1;
                                value_name++;
                            }
                            while (value_name) {
                                if (isdigit(*value_name)) {
                                    if (!is_integer) {
                                        fraction_part = fraction_part * 10 + (*value_name - '0');
                                        fraction_divisor *= 10;
                                    } else {
                                        integer_part = integer_part * 10 + (*value_name - '0');
                                    }
                                    value_name++;
                                    continue;
                                }
                                if (*value_name == '.') {
                                    if (is_integer) {
                                        is_integer = IOTCS_FALSE;
                                        value_name++;
                                        continue;
                                    } else {
                                        LOG_WARN("Can not parse value of %s : %s", cur_function->function.alert_condition.fields[j].key, cur_function->function.alert_condition.fields[j].value);
                                    }
                                }
                                if (isspace(*value_name) || *value_name == '\0') {
                                    break;
                                }
                                // find something unexpected
                                LOG_WARN("Can not parse value of %s : %s", cur_function->function.alert_condition.fields[j].key, cur_function->function.alert_condition.fields[j].value);
                                break;
                            }
                            if (message->base->type == IOTCS_MESSAGE_DATA) {
                                desc[j].key = util_safe_strcpy(cur_function->function.alert_condition.fields[j].key);
                                if (fraction_part != 0) {
                                    desc[j].type = IOTCS_VALUE_TYPE_NUMBER;
                                    values[j].number_value = (float)is_negative*((float)integer_part + fraction_part / fraction_divisor);
                                } else {
                                    desc[j].type = IOTCS_VALUE_TYPE_INT;
                                    values[j].int_value = is_negative*integer_part;
                                }
                            }
                        } else {
                            if (message->base->type == IOTCS_MESSAGE_DATA) {
                                desc[j].key = util_safe_strcpy(cur_function->function.alert_condition.fields[j].key);
                                desc[j].type = IOTCS_VALUE_TYPE_STRING;
                                values[j].string_value = util_safe_strcpy(value_name);
                            }
                        }
                    }
                    desc[j].key = NULL;
                    desc[j].type = IOTCS_VALUE_TYPE_NONE;
                    if (message->base->sender) {
                        alert_msg_base->sender = util_safe_strcpy(message->base->sender);
                    } else {
                        alert_msg_base->sender = NULL;
                    }
                    if (message->base->source) {
                        alert_msg_base->source = util_safe_strcpy(message->base->source);
                    } else {
                        alert_msg_base->source = NULL;
                    }
                    iotcs_message alert_message = {
                        .id = "",
                        .base = alert_msg_base,
                        .u.alert.base = alert_base,
                        .u.alert.items_desc = desc,
                        .u.alert.items_value = values
                    };
                    alert_message.message_mark = IOTCS_OFFERED_ALERT_MARK;

                    if (iotcs_message_dispatcher_queue(&alert_message) != IOTCS_RESULT_OK) {
                        release_alert_msg_copy(&alert_message);
                    }
                    if (cur_function->function.alert_condition.filter) {
                        release_msg_copy(message);
                        util_free(message);
                        return IOTCS_TRUE;
                    }
                }
                break;
            case IOTCS_FUNCTION_TYPE_ACTION_CONDITION:
                formula = cur_function->function.action_condition.condition;
                cond = calculate(formula, virtual_device_handle, message);
                if (cond != 0) {
                    msg_request_message_holder *holder;
                    iotcs_request_message* req_message = NULL;

                    holder = request_get_free();
                    if (!holder) {
                        LOG_ERRS("No more room in the request message holder. Operation failed");
                        cur_function = cur_function->next;
                        continue;
                    }
                    req_message = &holder->request;

                    memset(req_message, 0, sizeof (iotcs_request_message));

                    req_message->id = util_safe_strcpy("");
                    req_message->source = util_safe_strcpy(INTERNAL_REQUEST_SOURCE);

                    if (message->base->source) {
                        req_message->destination = util_safe_strcpy(message->base->source);
                    } else {
                        req_message->destination = util_safe_strcpy(tam_get_endpoint_id());
                    }

                    int length = 0;
                    char *urn = util_safe_strcpy(message->u.data.base->format);
                    int i;
                    for (i = 0; urn[i] != '\0'; i++) { }
                    for (; urn[i] != ':'; i--) { }
                    urn[i] = '\0';
                    length = (strlen("deviceModels///") + strlen(urn) + strlen("actions")
                        + strlen(cur_function->function.action_condition.action) + 1);
                    req_message->url = (char*)util_malloc(sizeof(char) * length);
                    util_safe_snprintf(req_message->url, length, "deviceModels/%s/%s/%s", urn, "actions",
                        cur_function->function.action_condition.action);
                    req_message->method = IOTCS_REQUEST_METHOD_POST;
                    util_free(urn);
                    length += 2;
                    for (j = 0; j < cur_function->function.action_condition.fields_len; j++) {
                        length += (strlen("{\"value\":}") + strlen(cur_function->function.action_condition.arguments[j].value));
                    }
                    req_message->body = (char*)util_malloc(sizeof(char) * length);
                    req_message->body[0] = '\0';
                    int cx = 0;
                    for (j = 0; j < cur_function->function.action_condition.fields_len; j++) {
                        cx += util_safe_snprintf(req_message->body + cx, length - cx, "{\"%s\":%s}",
                                cur_function->function.action_condition.arguments[j].key,
                                cur_function->function.action_condition.arguments[j].value);
                    }

                    req_message->params_len = 0;
                    req_message->params = NULL;

                    if (req_message) {
                        cl_put_special_request_message(req_message, 0);
                    }

#ifdef IOTCS_LONG_POLLING
                    cl_long_polling_receive_should_be_interrupted();
#endif
                    if (cur_function->function.action_condition.filter) {
                        //UTIL_UNLOCK_LL();
                        release_msg_copy(message);
                        util_free(message);
                        return IOTCS_TRUE;
                    }
                    //UTIL_UNLOCK_LL();
                }
                break;
            case IOTCS_FUNCTION_TYPE_FILTER_CONDITION:
                formula = cur_function->function.filter_condition.condition;
                cond = calculate(formula, virtual_device_handle, message);
                if (cond != 0) {
                    release_msg_copy(message);
                    util_free(message);
                    return IOTCS_TRUE;
                }
                break;
            case IOTCS_FUNCTION_TYPE_COMPUTED_METRIC:
                formula = cur_function->function.computed_metric.formula;
                result = calculate(formula, virtual_device_handle, message);
                k = 0;
                while (1) {
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        if (message->u.data.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                            k = -1;
                            break;
                        }
                        if (strcmp(message->u.data.items_desc[k].key, attribute_name) == 0) {
                            if (message->u.data.items_desc[k].type == IOTCS_VALUE_TYPE_NUMBER) {
                                message->u.data.items_value[k].number_value = result;
                            } else if (message->u.data.items_desc[k].type == IOTCS_VALUE_TYPE_INT) {
                                message->u.data.items_value[k].int_value = (int)result;
                            }
                        }
                    } else {
                        if (message->u.alert.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                            k = -1;
                            break;
                        }
                        if (strcmp(message->u.alert.items_desc[k].key, attribute_name) == 0) {
                            if (message->u.alert.items_desc[k].type == IOTCS_VALUE_TYPE_NUMBER) {
                                message->u.alert.items_value[k].number_value = result;
                            } else {
                                message->u.alert.items_value[k].int_value = (int)result;
                            }
                        }
                    }
                    k++;
                }
                break;
            default:
                break;
        }
        cur_function = cur_function->next;
    }
#ifdef IOTCS_VIRTUALIZATION_SUPPORT
    j = 0;
    if (message->base->type == IOTCS_MESSAGE_DATA) {
        iotcs_virtual_device_start_update(virtual_device_handle);
        while (1) {
            if (message->u.data.items_desc[j].type == IOTCS_VALUE_TYPE_NONE) {
                break;
            }
            if (message->u.data.items_desc[j].type == IOTCS_VALUE_TYPE_INT) {
                iotcs_virtual_device_set_integer(virtual_device_handle, message->u.data.items_desc[j].key, message->u.data.items_value[j].int_value);
            } else if (message->u.data.items_desc[j].type == IOTCS_VALUE_TYPE_NUMBER) {
                iotcs_virtual_device_set_float(virtual_device_handle, message->u.data.items_desc[j].key, message->u.data.items_value[j].number_value);
            }
            j++;
        }
        iotcs_virtual_device_finish_update(virtual_device_handle);
        release_msg_copy(message);
        util_free(message);
    } else {
        return (iotcs_message_dispatcher_queue(message) == IOTCS_RESULT_OK);
    }
    return IOTCS_TRUE;
#else
    return (iotcs_message_dispatcher_queue(message) == IOTCS_RESULT_OK);
#endif
}

/* Put value to function with window */
void put_value_to_window(struct iotcs_window_handle_t *window, iotcs_typed_value *value) {
    int j, k;
    window->last_offer_time = iotcs_port_get_current_time_millis();
    if (window->time_last < window->time_delta) {
        // Case when we should drop value
        if (window->sheduled_time - window->last_offer_time > window->time_last) {
            return;
        }
    }

    switch (window->type) {
        case IOTCS_FUNCTION_TYPE_MEAN:
            for (j = 0; j < window->size; j++) {
                window->typed_value[j].type = value->type;
                if (value->type == IOTCS_VALUE_TYPE_INT) {
                    window->typed_value[j].value.int_value += value->value.int_value;
                } else if (value->type == IOTCS_VALUE_TYPE_NUMBER) {
                    window->typed_value[j].value.number_value += value->value.number_value;
                } else {
                    LOG_INFO("Unexpected type of value %d", value->type);
                }
                window->count[j]++;
            }
            break;
        case IOTCS_FUNCTION_TYPE_MIN:
            for (j = 0; j < window->size; j++) {
                window->typed_value[j].type = value->type;
                if (value->type == IOTCS_VALUE_TYPE_INT) {
                    if (value->value.int_value < window->typed_value[j].value.int_value || !window->count[j]) {
                        window->typed_value[j].value.int_value = value->value.int_value;
                        window->count[j] = 1;
                    }
                } else if (value->type == IOTCS_VALUE_TYPE_NUMBER) {
                    if (value->value.number_value < window->typed_value[j].value.number_value
                        || !window->count[j]) {
                        window->typed_value[j].value.number_value = value->value.number_value;
                        window->count[j] = 1;
                    }
                } else {
                    LOG_INFO("Unexpected type of value %d", value->type);
                }
            }
            break;
        case IOTCS_FUNCTION_TYPE_MAX:
            for (j = 0; j < window->size; j++) {
                window->typed_value[j].type = value->type;
                if (value->type == IOTCS_VALUE_TYPE_INT) {
                    if (value->value.int_value > window->typed_value[j].value.int_value
                        || !window->count[j]) {
                        window->typed_value[j].value.int_value = value->value.int_value;
                        window->count[j] = 1;
                    }
                } else if (value->type == IOTCS_VALUE_TYPE_NUMBER) {
                    if (value->value.number_value > window->typed_value[j].value.number_value
                        || !window->count[j]) {
                        window->typed_value[j].value.number_value = value->value.number_value;
                        window->count[j] = 1;
                    }
                } else {
                    LOG_INFO("Unexpected type of value %d", value->type);
                }
            }
            break;
        case IOTCS_FUNCTION_TYPE_DETECT_DUBL:
        case IOTCS_FUNCTION_TYPE_ELIMINATE_DUBL:
            for (j = 0; j < window->size; j++) {
                window->typed_value[j].type = value->type;
                if (value->type == IOTCS_VALUE_TYPE_INT) {
                    if (value->value.int_value != window->typed_value[j].value.int_value
                        || !window->count[j]) {
                        window->typed_value[j].value.int_value = value->value.int_value;
                        window->count[j] = 1;
                    }
                } else if (value->type == IOTCS_VALUE_TYPE_NUMBER) {
                    if (value->value.number_value != window->typed_value[j].value.int_value
                        || !window->count[j]) {
                        window->typed_value[j].value.number_value = value->value.number_value;
                        window->count[j] = 1;
                    }
                } else {
                    LOG_INFO("Unexpected type of value %d", value->type);
                }
            }
            break;
        case IOTCS_FUNCTION_TYPE_STANDARD_DEVIATION:
            if (window->args == NULL) {
                window->args = (void*)util_malloc(window->size * sizeof(iotcs_typed_value*));
                for (j = 0; j < window->size; j++) {
                    window->count[j] = 0;
                    ((iotcs_typed_value**)window->args)[j] = (void*)util_malloc((STANDARD_DEVIATION_SIZE + 1) * sizeof(iotcs_typed_value));
                    ((iotcs_typed_value**)window->args)[j][0].value.int_value = STANDARD_DEVIATION_SIZE;
                }
            }
            for (j = 0; j < window->size; j++) {
                iotcs_typed_value *array = ((iotcs_typed_value**)window->args)[j];
                if (window->count[j] > array[0].value.int_value) {
                    iotcs_typed_value *new_array = (iotcs_typed_value *)util_malloc((array[0].value.int_value + STANDARD_DEVIATION_SIZE + 1) * sizeof(iotcs_typed_value));
                    new_array[0].value.int_value = array[0].value.int_value + STANDARD_DEVIATION_SIZE;
                    for (k = 1; k <= window->count[j]; k++) {
                        new_array[k].type = array[k].type;
                        if (array[k].type == IOTCS_VALUE_TYPE_INT) {
                            new_array[k].value.int_value = array[k].value.int_value;
                        } else if (array[k].type == IOTCS_VALUE_TYPE_NUMBER) {
                            new_array[k].value.number_value = array[k].value.number_value;
                        }
                    }
                    ((iotcs_typed_value**)window->args)[j] = new_array;
                    util_free(array);
                    array = new_array;
                }
                if (value->type == IOTCS_VALUE_TYPE_INT) {
                    array[window->count[j] + 1].type = IOTCS_VALUE_TYPE_INT;
                    array[window->count[j] + 1].value.int_value = value->value.int_value;
                } else if (value->type == IOTCS_VALUE_TYPE_NUMBER) {
                    array[window->count[j] + 1].type = IOTCS_VALUE_TYPE_NUMBER;
                    array[window->count[j] + 1].value.number_value = value->value.number_value;
                } else {
                    LOG_INFO("Unexpected type of value %d", value->type);
                }
                window->count[j]++;
            }
            break;
        default:
            break;
    }
}

iotcs_bool dp_handle_math_function(iotcs_pipeline_function_list_t *cur_function, iotcs_virtual_device_handle_t* virtual_device_handle,
                                        char *attribute_name, iotcs_message *message) {
    int i, k, j;
    iotcs_bool result = IOTCS_FALSE;
    iotcs_typed_value value;
    int64_t window = 0, slide = 0;
    char *endpoint_id = NULL;

    if (message != NULL) {
        if (message->base->type != IOTCS_MESSAGE_DATA &&
            message->base->type != IOTCS_MESSAGE_ALERT) {
            return result;
        }
        if (message->base->source) {
            endpoint_id = message->base->source;
        } else {
            endpoint_id = tam_get_endpoint_id();
        }
    }
    k = 0;
    value.type = IOTCS_VALUE_TYPE_NONE;
    while (1) {
        if (message->base->type == IOTCS_MESSAGE_DATA) {
            if (message->u.data.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                break;
            }
            if (strcmp(message->u.data.items_desc[k].key, attribute_name) == 0) {
                value.type = message->u.data.items_desc[k].type;
                if (value.type == IOTCS_VALUE_TYPE_INT) {
                    value.value.int_value = message->u.data.items_value[k].int_value;
                } else if (value.type == IOTCS_VALUE_TYPE_NUMBER) {
                    value.value.number_value = message->u.data.items_value[k].number_value;
                }
                break;
            }
        } else {
            if (message->u.alert.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                break;
            }
            if (strcmp(message->u.alert.items_desc[k].key, attribute_name) == 0) {
                value.type = message->u.alert.items_desc[k].type;
                if (value.type == IOTCS_VALUE_TYPE_INT) {
                    value.value.int_value = message->u.alert.items_value[k].int_value;
                } else if (value.type == IOTCS_VALUE_TYPE_NUMBER) {
                    value.value.number_value = message->u.alert.items_value[k].number_value;
                }
                break;
            }
        }
        k++;
    }

    for (i = 0; i < WINDOWS_COUNT; i++) {
        if (window_array[i].device_handle == NULL &&
            window_array[i].endpoint_id == NULL)
            continue;
        if (((window_array[i].device_handle != NULL && window_array[i].device_handle == virtual_device_handle) ||
            ((window_array[i].endpoint_id   != NULL && endpoint_id != NULL &&
            strcmp(window_array[i].endpoint_id, endpoint_id) == 0))) &&
            strcmp(window_array[i].attribute_name, attribute_name) == 0 &&
            cur_function == window_array[i].cur_function) {
            if (cur_function->type != window_array[i].type) {
                continue;
            }
            result = IOTCS_TRUE;
            put_value_to_window(&window_array[i], &value);
            release_msg_copy(message);
            util_free(message);
            break;
        }
    }

    // If there was no windows then add one
    if (result == IOTCS_FALSE) {
        for (i = 0; i < WINDOWS_COUNT; i++) {
            if (window_array[i].device_handle == NULL && window_array[i].endpoint_id == NULL) {
                result = IOTCS_TRUE;
                if (virtual_device_handle) {
                    window_array[i].device_handle = (iotcs_virtual_device_handle)virtual_device_handle;
                } else {
                    window_array[i].endpoint_id = endpoint_id;
                }
                // Must be before type set to create memory for deviation function
                window_array[i].args = NULL;
                switch (cur_function->type) {
                    case IOTCS_FUNCTION_TYPE_MEAN:
                        window = cur_function->function.mean_function.window;
                        slide = cur_function->function.mean_function.slide;
                        break;
                    case IOTCS_FUNCTION_TYPE_MIN:
                        window = cur_function->function.min_function.window;
                        slide = cur_function->function.min_function.slide;
                        break;
                    case IOTCS_FUNCTION_TYPE_MAX:
                        window = cur_function->function.max_function.window;
                        slide = cur_function->function.max_function.slide;
                        break;
                    case IOTCS_FUNCTION_TYPE_ELIMINATE_DUBL:
                        window = cur_function->function.eliminate_dubl_function.window;
                        slide = 0;
                        break;
                    case IOTCS_FUNCTION_TYPE_DETECT_DUBL:
                        window = cur_function->function.detect_dubl_function.window;
                        slide = 0;
                        break;
                    case IOTCS_FUNCTION_TYPE_STANDARD_DEVIATION:
                        window = cur_function->function.standard_deviation_function.window;
                        slide = cur_function->function.standard_deviation_function.slide;
                        break;
                    default:
                        LOG_ERRS("Wrong policy function");
                        return result;
                }

                if (slide) {
                    int k = window / slide;
                    if (k * slide == window) {
                        window_array[i].size = k;
                    } else {
                        window_array[i].size = k + 1;
                    }
                } else {
                    window_array[i].size = 1;
                }

                window_array[i].value_index = 0;
                window_array[i].typed_value = (iotcs_typed_value*)util_malloc(window_array[i].size * sizeof(iotcs_typed_value));
                window_array[i].count = (int*)util_malloc(window_array[i].size * sizeof(int));
                for (j = 0; j < window_array[i].size; j++) {
                    window_array[i].count[j] = 0;
                }
                if (window_array[i].size == 1) {
                    if (slide == 0) {
                        window_array[i].sheduled_time = iotcs_port_get_current_time_millis() + window;
                        window_array[i].time_delta = window;
                        window_array[i].time_last = window;
                    } else {
                        window_array[i].sheduled_time = iotcs_port_get_current_time_millis() + window;
                        window_array[i].time_delta = slide;
                        window_array[i].time_last = window;
                    }
                } else {
                    window_array[i].sheduled_time = iotcs_port_get_current_time_millis() + slide;
                    window_array[i].time_delta = slide;
                    window_array[i].time_last = window;
                }
                window_array[i].type = cur_function->type;
                window_array[i].cur_function = cur_function;
                window_array[i].message = message;
                put_value_to_window(&window_array[i], &value);
                window_array[i].attribute_name = util_safe_strcpy(attribute_name);
                return result;
            }
        }
    }
    return result;
}


iotcs_bool dp_handle_duplicates_function(iotcs_pipeline_function_list_t *cur_function, iotcs_virtual_device_handle_t* virtual_device_handle,
                                        char *attribute_name, iotcs_message *message) {
    int i, k, j;
    iotcs_typed_value value;
    int64_t window;
    char *endpoint_id = NULL;
    int64_t offer_delta = 0;

    if (message != NULL) {
        if (message->base->type != IOTCS_MESSAGE_DATA &&
            message->base->type != IOTCS_MESSAGE_ALERT) {
            return IOTCS_FALSE;
        }
        if (message->base->source) {
            endpoint_id = message->base->source;
        } else {
            endpoint_id = tam_get_endpoint_id();
        }
    }
    k = 0;
    value.type = IOTCS_VALUE_TYPE_NONE;
    while (1) { 
        if (message->base->type == IOTCS_MESSAGE_DATA) {
            if (message->u.data.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                break;
            }
            if (strcmp(message->u.data.items_desc[k].key, attribute_name) == 0) {
                value.type = message->u.data.items_desc[k].type;
                value.value = message->u.data.items_value[k];
            }
        } else {
            if (message->u.alert.items_desc[k].type == IOTCS_VALUE_TYPE_NONE) {
                break;
            }
            if (strcmp(message->u.alert.items_desc[k].key, attribute_name) == 0) {
                value.type = message->u.alert.items_desc[k].type;
                value.value = message->u.alert.items_value[k];
            }
        }
        k++;
    }

    for (i = 0; i < WINDOWS_COUNT; i++) {
        if (window_array[i].device_handle == NULL &&
            window_array[i].endpoint_id == NULL)
            continue;
        if (((window_array[i].device_handle != NULL && window_array[i].device_handle == virtual_device_handle) ||
            ((window_array[i].endpoint_id   != NULL && endpoint_id != NULL &&
            strcmp(window_array[i].endpoint_id, endpoint_id) == 0))) &&
            strcmp(window_array[i].attribute_name, attribute_name) == 0) {
            if (cur_function->type != window_array[i].type &&
            cur_function == window_array[i].cur_function) {
                continue;
            }
            switch (window_array[i].type) {
                case IOTCS_FUNCTION_TYPE_DETECT_DUBL:
                case IOTCS_FUNCTION_TYPE_ELIMINATE_DUBL:
                    offer_delta = iotcs_port_get_current_time_millis() -  window_array[i].last_offer_time;
                    if (value.type == IOTCS_VALUE_TYPE_INT) {
                        if ((value.value.int_value != window_array[i].typed_value[0].value.int_value)
                            || (offer_delta > window_array[i].time_delta)
                            || !window_array[i].count[0]) {
                            window_array[i].typed_value[0].value.int_value = value.value.int_value;
                            window_array[i].count[0] = 1;
                            window_array[i].last_offer_time = iotcs_port_get_current_time_millis();
                            return IOTCS_FALSE;
                        } else {
                            return IOTCS_TRUE;
                        }
                    } else if (value.type == IOTCS_VALUE_TYPE_NUMBER) {
                        if ((value.value.number_value != window_array[i].typed_value[0].value.int_value)
                            || (offer_delta > window_array[i].time_delta)
                            || !window_array[i].count[0]) {
                            window_array[i].typed_value[0].value.number_value = value.value.number_value;
                            window_array[i].count[0] = 1;
                            window_array[i].last_offer_time = iotcs_port_get_current_time_millis();
                            return IOTCS_FALSE;
                        } else {
                            return IOTCS_TRUE;
                        }
                    } else {
                        LOG_INFO("Unexpected type of value %d", value.type);
                    }
                    break;
                default:
                    break;
            }
        }
    }

    for (i = 0; i < WINDOWS_COUNT; i++) {
        if (window_array[i].device_handle == NULL && window_array[i].endpoint_id == NULL) {
            if (virtual_device_handle) {
                window_array[i].device_handle = (iotcs_virtual_device_handle)virtual_device_handle;
            } else {
                window_array[i].endpoint_id = endpoint_id;
            }
            switch (cur_function->type) {
                case IOTCS_FUNCTION_TYPE_ELIMINATE_DUBL:
                    window = cur_function->function.eliminate_dubl_function.window;
                    break;
                case IOTCS_FUNCTION_TYPE_DETECT_DUBL:
                    window = cur_function->function.detect_dubl_function.window;
                    break;
                default:
                    window = -1;
                    LOG_ERRS("Wrong policy function");
                    break;
            }
            window_array[i].size = 1;

            window_array[i].value_index = 0;
            window_array[i].typed_value = (iotcs_typed_value*)util_malloc(window_array[i].size * sizeof(iotcs_typed_value));
            window_array[i].count = (int*)util_malloc(window_array[i].size * sizeof(int));
            for (j = 0; j < window_array[i].size; j++) {
                window_array[i].count[j] = 1;
                if (value.type == IOTCS_VALUE_TYPE_INT) {
                    window_array[i].typed_value[j].type = IOTCS_VALUE_TYPE_INT;
                    window_array[i].typed_value[j].value.int_value = value.value.int_value;
                } else {
                    window_array[i].typed_value[j].type = IOTCS_VALUE_TYPE_NUMBER;
                    window_array[i].typed_value[j].value.number_value = value.value.number_value;
                }
            }
            window_array[i].sheduled_time = iotcs_port_get_current_time_millis() + window;
            window_array[i].time_delta = window;
            window_array[i].last_offer_time = iotcs_port_get_current_time_millis();
            window_array[i].args = NULL;
            window_array[i].type = cur_function->type;
            window_array[i].message = NULL;
            window_array[i].attribute_name = util_safe_strcpy(attribute_name);
            return IOTCS_FALSE;
        }
    }

    return IOTCS_FALSE;
}

/* Handle function with a window */
void proceed_policy(int i) {
    iotcs_typed_value *array, tmp;
    iotcs_bool rv;
    int k, j, value_index = window_array[i].value_index;
    window_array[i].value_index = (window_array[i].value_index + 1) % window_array[i].size;
    iotcs_message *message = window_array[i].message, *msg_copy;
    
    j = 0;
    if (window_array[i].type != IOTCS_FUNCTION_TYPE_DETECT_DUBL &&
        window_array[i].type != IOTCS_FUNCTION_TYPE_ELIMINATE_DUBL
        && message) {
        while (1) {
            if (message->base->type == IOTCS_MESSAGE_DATA) {
                if (message->u.data.items_desc[j].type == IOTCS_VALUE_TYPE_NONE) {
                    j = -1;
                    break;
                }
                if (strcmp(message->u.data.items_desc[j].key, window_array[i].attribute_name) == 0) {
                    break;
                }
            } else {
                if (message->u.alert.items_desc[j].type == IOTCS_VALUE_TYPE_NONE) {
                    j = -1;
                    break;
                }
                if (strcmp(message->u.alert.items_desc[j].key, window_array[i].attribute_name) == 0) {
                    break;
                }
            }
            j++;
        }
    }

    switch (window_array[i].type) {
        case IOTCS_FUNCTION_TYPE_MEAN:
            if (window_array[i].count[value_index]) {
                if (window_array[i].typed_value[value_index].type == IOTCS_VALUE_TYPE_INT) {
                    window_array[i].typed_value[value_index].value.int_value =
                            window_array[i].typed_value[value_index].value.int_value / window_array[i].count[value_index];
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        message->u.data.items_value[j].int_value = window_array[i].typed_value[value_index].value.int_value;
                    } else {
                        message->u.alert.items_value[j].int_value = window_array[i].typed_value[value_index].value.int_value;
                    }
                } else if (window_array[i].typed_value[value_index].type == IOTCS_VALUE_TYPE_NUMBER) {
                    window_array[i].typed_value[value_index].value.number_value =
                            window_array[i].typed_value[value_index].value.number_value / window_array[i].count[value_index];
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        message->u.data.items_value[j].number_value = window_array[i].typed_value[value_index].value.number_value;
                    } else {
                        message->u.alert.items_value[j].number_value = window_array[i].typed_value[value_index].value.number_value;
                    }
                }
                if (window_array[i].typed_value[value_index].type == IOTCS_VALUE_TYPE_INT) {
                    window_array[i].typed_value[value_index].value.int_value = 0;
                } else if (window_array[i].typed_value[value_index].type == IOTCS_VALUE_TYPE_NUMBER) {
                    window_array[i].typed_value[value_index].value.number_value = 0;
                }
                window_array[i].count[value_index] = 0;
                copy_msg(window_array[i].message, &msg_copy);
                rv = apply_pipe(window_array[i].cur_function->next, window_array[i].attribute_name, window_array[i].device_handle, msg_copy);
                if (rv != IOTCS_TRUE) {
                    release_msg_copy(msg_copy);
                    util_free(msg_copy);
                }
            }
            break;
        case IOTCS_FUNCTION_TYPE_MIN:
            if (window_array[i].count[value_index]) {
                if (window_array[i].typed_value[value_index].type == IOTCS_VALUE_TYPE_INT) {
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        message->u.data.items_value[j].int_value = window_array[i].typed_value[value_index].value.int_value;
                    } else {
                        message->u.alert.items_value[j].int_value = window_array[i].typed_value[value_index].value.int_value;
                    }
                    window_array[i].typed_value[value_index].value.int_value = 0;
                } else if (window_array[i].typed_value[value_index].type == IOTCS_VALUE_TYPE_NUMBER) {
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        message->u.data.items_value[j].number_value = window_array[i].typed_value[value_index].value.number_value;
                    } else {
                        message->u.alert.items_value[j].number_value = window_array[i].typed_value[value_index].value.number_value;
                    }
                    window_array[i].typed_value[value_index].value.number_value = 0;
                }
                window_array[i].typed_value[value_index].value.int_value = 0;
                window_array[i].count[value_index] = 0;
                copy_msg(window_array[i].message, &msg_copy);
                rv = apply_pipe(window_array[i].cur_function->next, window_array[i].attribute_name, window_array[i].device_handle, msg_copy);
                if (rv != IOTCS_TRUE) {
                    release_msg_copy(msg_copy);
                    util_free(msg_copy);
                }
            }
            break;
        case IOTCS_FUNCTION_TYPE_MAX:
            if (window_array[i].count[value_index]) {
                if (window_array[i].typed_value[value_index].type == IOTCS_VALUE_TYPE_INT) {
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        message->u.data.items_value[j].int_value = window_array[i].typed_value[value_index].value.int_value;
                    } else {
                        message->u.alert.items_value[j].int_value = window_array[i].typed_value[value_index].value.int_value;
                    }
                    window_array[i].typed_value[value_index].value.int_value = 0;
                } else if (window_array[i].typed_value[value_index].type == IOTCS_VALUE_TYPE_NUMBER) {
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        message->u.data.items_value[j].number_value = window_array[i].typed_value[value_index].value.number_value;
                    } else {
                        message->u.alert.items_value[j].number_value = window_array[i].typed_value[value_index].value.number_value;
                    }
                    window_array[i].typed_value[value_index].value.number_value = 0;
                }
                window_array[i].count[value_index] = 0;
                copy_msg(window_array[i].message, &msg_copy);
                rv = apply_pipe(window_array[i].cur_function->next, window_array[i].attribute_name, window_array[i].device_handle, msg_copy);
                if (rv != IOTCS_TRUE) {
                    release_msg_copy(msg_copy);
                    util_free(msg_copy);
                }
            }
            break;
        case IOTCS_FUNCTION_TYPE_STANDARD_DEVIATION:
            if (window_array[i].count[value_index] > 1) {
                double mean = 0;
                double sqrSumm = 0;
                array = ((iotcs_typed_value**)window_array[i].args)[value_index];
                if (array[1].type == IOTCS_VALUE_TYPE_INT) {
                    array[1].type = IOTCS_VALUE_TYPE_INT;
                    tmp.type = IOTCS_VALUE_TYPE_INT;
                    tmp.value.int_value = 0;
                } else {
                    array[1].type = IOTCS_VALUE_TYPE_NUMBER;
                    tmp.type = IOTCS_VALUE_TYPE_NUMBER;
                    tmp.value.number_value = (float)0;
                }
                for (k = 1; k <= window_array[i].count[value_index]; k++) {
                    if (array[k].type == IOTCS_VALUE_TYPE_INT) {
                        tmp.value.int_value += array[k].value.int_value;
                    } else if (array[k].type == IOTCS_VALUE_TYPE_NUMBER) {
                        tmp.value.number_value += array[k].value.number_value;
                    }
                }
                if (tmp.type == IOTCS_VALUE_TYPE_INT) {
                    mean = (double)tmp.value.int_value / (double)window_array[i].count[value_index];
                } else if (array[k].type == IOTCS_VALUE_TYPE_NUMBER) {
                    mean =  tmp.value.number_value / (double)window_array[i].count[value_index];
                }
                for (k = 1; k <= window_array[i].count[value_index]; k++) {
                    if (array[k].type == IOTCS_VALUE_TYPE_INT) {
                        sqrSumm += ((float)array[k].value.int_value - mean) * ((float)array[k].value.int_value - mean);
                    } else if (array[k].type == IOTCS_VALUE_TYPE_NUMBER) {
                        sqrSumm += (array[k].value.number_value - mean) * (array[k].value.number_value - mean);
                    }
                }

                if (array[k].type == IOTCS_VALUE_TYPE_INT) {
                    sqrSumm = (int)sqrt(sqrSumm / ((double)(window_array[i].count[value_index] - 1)));
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        message->u.data.items_value[j].int_value = (int)sqrSumm;
                    } else {
                        message->u.alert.items_value[j].int_value = (int)sqrSumm;
                    }
                } else if (array[k].type == IOTCS_VALUE_TYPE_NUMBER) {
                    sqrSumm = (float)sqrt((double)tmp.value.number_value / ((double)(window_array[i].count[value_index] - 1)));
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        message->u.data.items_value[j].number_value = (float)sqrSumm;
                    } else {
                        message->u.alert.items_value[j].number_value = (float)sqrSumm;
                    }
                }
                window_array[i].count[value_index] = 0;
                copy_msg(window_array[i].message, &msg_copy);
                rv = apply_pipe(window_array[i].cur_function->next, window_array[i].attribute_name, window_array[i].device_handle, msg_copy);
                if (rv != IOTCS_TRUE) {
                    release_msg_copy(msg_copy);
                    util_free(msg_copy);
                }
            } else if (window_array[i].count[value_index] == 1) {
                if (message->u.data.items_desc[j].type == IOTCS_VALUE_TYPE_INT) {
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        message->u.data.items_value[j].int_value = 0;
                    } else {
                        message->u.alert.items_value[j].int_value = 0;
                    }
                } else if (message->u.data.items_desc[j].type == IOTCS_VALUE_TYPE_NUMBER) {
                    if (message->base->type == IOTCS_MESSAGE_DATA) {
                        message->u.data.items_value[j].number_value = (float)0;
                    } else {
                        message->u.alert.items_value[j].number_value = (float)0;
                    }
                } else if (message->base->type == IOTCS_MESSAGE_DATA) {
                    message->u.data.items_value[j].int_value = 0;
                } else {
                    message->u.alert.items_value[j].int_value = 0;
                }
                window_array[i].count[value_index] = 0;
                copy_msg(window_array[i].message, &msg_copy);
                rv = apply_pipe(window_array[i].cur_function->next, window_array[i].attribute_name, window_array[i].device_handle, msg_copy);
                if (rv != IOTCS_TRUE) {
                    release_msg_copy(msg_copy);
                    util_free(msg_copy);
                }
            }
            break;
        case IOTCS_FUNCTION_TYPE_DETECT_DUBL:
        case IOTCS_FUNCTION_TYPE_ELIMINATE_DUBL:
            if (window_array[i].last_offer_time - iotcs_port_get_current_time_millis() > window_array[i].time_delta &&
                window_array[i].count[value_index]) {
                window_array[i].count[value_index] = 0;
            }
            break;
        default:
            break;
    }
}

void check_window() {
    int i = 0;
    int64_t delta = 0, min_delta;
    min_delta = 5000;
    UTIL_POLICY_LOCK();
    for (i = 0; i < WINDOWS_COUNT; i++) {
        if (window_array[i].attribute_name && (window_array[i].device_handle || window_array[i].message)) {
            delta = window_array[i].sheduled_time - iotcs_port_get_current_time_millis();
            if (delta <= 0) {
                proceed_policy(i);
                window_array[i].sheduled_time += window_array[i].time_delta;
                delta = window_array[i].time_delta;
            } else if (delta < min_delta) {
                min_delta = delta;
            }
        }
    }
    UTIL_POLICY_UNLOCK();
    iotcs_port_sleep_millis(min_delta / 2);
}

void* policy_thread(void *arg) {
    (void) arg;
    while (1) {
        check_window();
    }
    return NULL;
}

void dp_release_policy(iotcs_policy_handle_t* policy) {
    int i = 0;
    iotcs_pipeline_t *cur_pipe = NULL, *tmp_cur_pipe = NULL;
    iotcs_pipeline_function_list_t *cur_func_list = NULL, *tmp_cur_func_list = NULL;
    iotcs_key_value *params;

    if (!policy) {
        return;
    }

    cur_pipe = policy->pipe_list;
    while (cur_pipe) {
        cur_func_list = cur_pipe->function_list;
        while (cur_func_list) {
            switch (cur_func_list->type) {
                case IOTCS_FUNCTION_TYPE_MEAN:
                case IOTCS_FUNCTION_TYPE_MIN:
                case IOTCS_FUNCTION_TYPE_MAX:
                case IOTCS_FUNCTION_TYPE_STANDARD_DEVIATION:
                case IOTCS_FUNCTION_TYPE_SAMPLE_QUALITY:
                    break;
                case IOTCS_FUNCTION_TYPE_DETECT_DUBL:
                    util_free(cur_func_list->function.detect_dubl_function.format);
                    util_free(cur_func_list->function.detect_dubl_function.severity);
                    params = cur_func_list->function.detect_dubl_function.fields;
                    for (i = 0; i < cur_func_list->function.detect_dubl_function.fields_len; i++) {
                        util_free((void*)params->key);
                        util_free((void*)params->value);
                        params++;
                    }
                    util_free(cur_func_list->function.detect_dubl_function.fields);
                    break;
                case IOTCS_FUNCTION_TYPE_ALERT_CONDITION:
                    util_free(cur_func_list->function.alert_condition.condition);
                    util_free(cur_func_list->function.alert_condition.format);
                    util_free(cur_func_list->function.alert_condition.severity);
                    params = cur_func_list->function.alert_condition.fields;
                    for (i = 0; i < cur_func_list->function.alert_condition.fields_len; i++) {
                        util_free((void*)params->key);
                        util_free((void*)params->value);
                        params++;
                    }
                    util_free(cur_func_list->function.alert_condition.fields);
                    break;
                case IOTCS_FUNCTION_TYPE_ACTION_CONDITION:
                    util_free(cur_func_list->function.action_condition.condition);
                    util_free(cur_func_list->function.action_condition.action);
                    params = cur_func_list->function.action_condition.arguments;
                    for (i = 0; i < cur_func_list->function.action_condition.fields_len; i++) {
                        util_free((void*)params->key);
                        util_free((void*)params->value);
                        params++;
                    }
                    util_free(cur_func_list->function.action_condition.arguments);
                    break;
                case IOTCS_FUNCTION_TYPE_FILTER_CONDITION:
                    util_free(cur_func_list->function.filter_condition.condition);
                    break;
                case IOTCS_FUNCTION_TYPE_COMPUTED_METRIC:
                    util_free(cur_func_list->function.computed_metric.formula);
                    break;
                case IOTCS_FUNCTION_TYPE_BATCH_TIME:
                    util_free(cur_func_list->function.batch_time.severity);
                    g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_LOW;
                    g_batch_message_time = 0;
                    break;
                case IOTCS_FUNCTION_TYPE_BATCH_SIZE:
                    util_free(cur_func_list->function.batch_size.severity);
                    g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_LOW;
                    g_batch_message_size = 0;
                    break;
                default:
                    break;
            }
            tmp_cur_func_list = cur_func_list;
            cur_func_list = cur_func_list->next;
            util_free(tmp_cur_func_list);
        }
        tmp_cur_pipe = cur_pipe;
        cur_pipe = cur_pipe->next;
        util_free(tmp_cur_pipe);
        policy->pipe_list = NULL;
    }
    util_free(policy->id);
    util_free(policy->urn);
    util_free(policy->endpoint_id);
    util_free(policy);
}

void dp_remove_policy(char* urn, char* name) {
    iotcs_policy_handle_t *cur_policy = g_policy_list, *prev_policy = NULL;
    while (cur_policy) {
        if (strcmp(cur_policy->id, name) == 0 && strcmp(cur_policy->urn, urn) == 0) {
            if (prev_policy) {
                prev_policy->next = cur_policy->next;
            } else {
                g_policy_list = g_policy_list->next;
            }
            dp_release_policy(cur_policy);
        }
        prev_policy = cur_policy;
        cur_policy = g_policy_list->next;
    }
}

iotcs_result dp_parse_policy_fromJSON(char *device_model, json_tokens_t* tok,
                                      iotcs_policy_handle_t** policy) {
    json_tokens_t pipe_arr_token, pipe_arr_current_token, pipe_token, function_token,
                parameter_list_token, parameter_list_current_token, sub_object_token;
    json_array_iterator_t pipe_iter, pipe_arr_iter, param_arr_iter;
    json_object_iterator_t field_object_iter;
    iotcs_bool enabled = IOTCS_TRUE;
    iotcs_result rv;
    char *value;
    int len;
    iotcs_key_value *params, *act_param;
    iotcs_pipeline_t *cur_pipe = NULL;
    iotcs_pipeline_t *tmp_cur_pipe = NULL;
    iotcs_pipeline_function_list_t *cur_func_list = NULL;
    iotcs_pipeline_function_list_t *tmp_cur_func_list = NULL;
    iotcs_policy_handle_t* cur_policy = NULL;

    cur_policy = (iotcs_policy_handle_t*)util_calloc(1, sizeof(iotcs_policy_handle_t));
    rv = json_get_bool_value_by_key(tok, "enabled", &enabled);
    GOTO_ERR(rv != IOTCS_RESULT_INVALID_ARGUMENT && rv != IOTCS_RESULT_OK);
    if (!enabled) {
        goto error;
    }
    CHECK_OOM(cur_policy->urn = util_safe_strcpy(device_model));
    GOTO_ERR(json_get_string_value_by_key(tok, "id", &value, &len) != IOTCS_RESULT_OK);
    CHECK_OOM(cur_policy->id = util_safe_strncpy(value, len));

    GOTO_ERR((rv = json_get_array_by_key(tok, "pipelines", &pipe_arr_token)) != IOTCS_RESULT_OK);
    GOTO_ERR_NO_OK(rv, json_get_array_iterator(&pipe_arr_token, &pipe_arr_iter));
    while (json_array_iterator_has_next(&pipe_arr_iter)) {
        GOTO_ERR_NO_OK(rv, json_array_iterator_next(&pipe_arr_iter, &pipe_arr_current_token));
        tmp_cur_pipe = util_calloc(1, sizeof(iotcs_pipeline_t));
        tmp_cur_pipe->next = NULL;
        if (cur_pipe) {
            cur_pipe->next = tmp_cur_pipe;
        } else {
            cur_policy->pipe_list = tmp_cur_pipe;
        }
        cur_pipe = tmp_cur_pipe;

        if ((rv = json_get_string_value_by_key(&pipe_arr_current_token, "attributeName", &value, &len)) != IOTCS_RESULT_OK) {
            GOTO_ERR(rv != IOTCS_RESULT_INVALID_ARGUMENT);
            tmp_cur_pipe->attribute = NULL;
        } else {
            CHECK_OOM(tmp_cur_pipe->attribute = util_safe_strncpy(value, len));
        }

        GOTO_ERR((rv = json_get_array_by_key(&pipe_arr_current_token, "pipeline", &pipe_token)) != IOTCS_RESULT_OK);
        GOTO_ERR_NO_OK(rv, json_get_array_iterator(&pipe_token, &pipe_iter));
        while (json_array_iterator_has_next(&pipe_iter)) {
            GOTO_ERR_NO_OK(rv, json_array_iterator_next(&pipe_iter, &function_token));
            tmp_cur_func_list = util_calloc(1, sizeof(iotcs_pipeline_function_list_t));
            tmp_cur_func_list->type = IOTCS_FUNCTION_TYPE_NONE;
            tmp_cur_func_list->next = NULL;
            if (cur_func_list) {
                cur_func_list->next = tmp_cur_func_list;
            } else {
                cur_pipe->function_list = tmp_cur_func_list;
            }
            cur_func_list = tmp_cur_func_list;
            GOTO_ERR(json_get_string_value_by_key(&function_token, "id", &value, &len) != IOTCS_RESULT_OK);
            CHECK_OOM(value = util_safe_strncpy(value, len));
            if (strcmp(value, "mean") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_MEAN;
                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "window", &cur_func_list->function.mean_function.window)) != IOTCS_RESULT_OK) {
                            goto error;
                        }
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "slide", &cur_func_list->function.mean_function.slide)) != IOTCS_RESULT_OK || cur_func_list->function.mean_function.slide < 0) {
                            cur_func_list->function.mean_function.slide = 0;
                        }
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "min") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_MIN;
                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "window", &cur_func_list->function.min_function.window)) != IOTCS_RESULT_OK) {
                            goto error;
                        }
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "slide", &cur_func_list->function.min_function.slide)) != IOTCS_RESULT_OK || cur_func_list->function.min_function.slide < 0) {
                            cur_func_list->function.min_function.slide = 0;
                        }
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "max") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_MAX;
                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "window", &cur_func_list->function.max_function.window)) != IOTCS_RESULT_OK) {
                            goto error;
                        }
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "slide", &cur_func_list->function.max_function.slide)) != IOTCS_RESULT_OK || cur_func_list->function.max_function.slide < 0) {
                            cur_func_list->function.max_function.slide = 0;
                        }
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "standardDeviation") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_STANDARD_DEVIATION;
                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "window", &cur_func_list->function.standard_deviation_function.window)) != IOTCS_RESULT_OK) {
                            goto error;
                        }
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "slide", &cur_func_list->function.standard_deviation_function.slide)) != IOTCS_RESULT_OK || cur_func_list->function.standard_deviation_function.slide < 0) {
                            cur_func_list->function.standard_deviation_function.slide = 0;
                        }
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "sampleQuality") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_SAMPLE_QUALITY;
                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        if (json_get_int_value_by_key(&parameter_list_token, "rate", &cur_func_list->function.sample_quality.rate) != IOTCS_RESULT_OK) {
                            if (json_get_string_value_by_key(&parameter_list_token, "rate", &value, &len) != IOTCS_RESULT_OK) {
                                cur_func_list->function.sample_quality.rate = IOTCS_SAMPLE_QUALITY_ALL;
                            } else {
                                if (strncmp(value, "all", len) == 0) {
                                    cur_func_list->function.sample_quality.rate = IOTCS_SAMPLE_QUALITY_ALL;
                                } else if (strncmp(value, "none", len) == 0) {
                                    cur_func_list->function.sample_quality.rate = IOTCS_SAMPLE_QUALITY_NONE;
                                } else if (strncmp(value, "random", len) == 0) {
                                    cur_func_list->function.sample_quality.rate = IOTCS_SAMPLE_QUALITY_RANDOM;
                                } else {
                                    cur_func_list->function.sample_quality.rate = IOTCS_SAMPLE_QUALITY_ALL;
                                }
                            }
                        }
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "eliminateDuplicates") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_ELIMINATE_DUBL;
                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "window", &cur_func_list->function.eliminate_dubl_function.window)) != IOTCS_RESULT_OK) {
                            goto error;
                        }
                        cur_func_list->function.eliminate_dubl_function.slide = 0;
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "detectDuplicates") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_DETECT_DUBL;
                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if ((rv = json_get_int_value_by_key(&parameter_list_token, "window", &cur_func_list->function.detect_dubl_function.window)) != IOTCS_RESULT_OK) {
                        goto error;
                    }
                    if ((rv = json_get_object_by_key(&parameter_list_token, "alert", &sub_object_token)) == IOTCS_RESULT_OK) {
                        GOTO_ERR(json_get_string_value_by_key(&sub_object_token, "urn", &value, &len) != IOTCS_RESULT_OK);
                        CHECK_OOM(cur_func_list->function.detect_dubl_function.format = util_safe_strncpy(value, len));
                        if (json_get_string_value_by_key(&sub_object_token, "severity", &value, &len) != IOTCS_RESULT_OK) {
                            CHECK_OOM(cur_func_list->function.detect_dubl_function.severity = util_safe_strcpy("SIGNIFICANT"));
                        } else {
                            CHECK_OOM(cur_func_list->function.detect_dubl_function.severity = util_safe_strncpy(value, len));
                        }

                        GOTO_ERR((rv = json_get_object_by_key(&sub_object_token, "fields", &parameter_list_current_token)) != IOTCS_RESULT_OK);
                        GOTO_ERR_NO_OK(rv, json_get_object_iterator(&parameter_list_current_token, &field_object_iter));
                        cur_func_list->function.detect_dubl_function.fields_len = json_child_items_count(&parameter_list_current_token);
                        CHECK_OOM(cur_func_list->function.detect_dubl_function.fields = (iotcs_key_value*)util_malloc(cur_func_list->function.detect_dubl_function.fields_len * sizeof (iotcs_key_value)));
                        params = cur_func_list->function.detect_dubl_function.fields;
                        while (json_object_iterator_has_next(&field_object_iter)) {
                            json_object_iterator_next(&field_object_iter, &sub_object_token);
                            GOTO_ERR_NO_OK(rv, json_get_string_and_move(&sub_object_token, &value, &len));
                            CHECK_OOM(params->key = util_safe_strncpy(value, len));
                            GOTO_ERR_NO_OK(rv, json_get_string_and_move(&sub_object_token, &value, &len));
                            CHECK_OOM(params->value = util_safe_strncpy(value, len));
                            params++;
                        }
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "alertCondition") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_ALERT_CONDITION;

                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        GOTO_ERR(json_get_string_value_by_key(&parameter_list_token, "condition", &value, &len) != IOTCS_RESULT_OK);
                        CHECK_OOM(cur_func_list->function.alert_condition.condition = util_safe_strncpy(value, len));

                        if ((rv = json_get_object_by_key(&parameter_list_token, "alert", &sub_object_token)) == IOTCS_RESULT_OK) {
                            GOTO_ERR(json_get_string_value_by_key(&sub_object_token, "urn", &value, &len) != IOTCS_RESULT_OK);
                            CHECK_OOM(cur_func_list->function.alert_condition.format = util_safe_strncpy(value, len));
                            GOTO_ERR(json_get_string_value_by_key(&sub_object_token, "severity", &value, &len) != IOTCS_RESULT_OK);
                            CHECK_OOM(cur_func_list->function.alert_condition.severity = util_safe_strncpy(value, len));

                            GOTO_ERR((rv = json_get_object_by_key(&sub_object_token, "fields", &parameter_list_current_token)) != IOTCS_RESULT_OK);
                            GOTO_ERR_NO_OK(rv, json_get_object_iterator(&parameter_list_current_token, &field_object_iter));
                            cur_func_list->function.alert_condition.fields_len = json_child_items_count(&parameter_list_current_token);
                            CHECK_OOM(cur_func_list->function.alert_condition.fields = (iotcs_key_value*)util_malloc(cur_func_list->function.alert_condition.fields_len * sizeof (iotcs_key_value)));
                            params = cur_func_list->function.alert_condition.fields;
                            while (json_object_iterator_has_next(&field_object_iter)) {
                                json_object_iterator_next(&field_object_iter, &sub_object_token);
                                GOTO_ERR_NO_OK(rv, json_get_string_and_move(&sub_object_token, &value, &len));
                                CHECK_OOM(params->key = util_safe_strncpy(value, len));
                                GOTO_ERR_NO_OK(rv, json_get_string_and_move(&sub_object_token, &value, &len));
                                CHECK_OOM(params->value = util_safe_strncpy(value, len));
                                params++;
                            }
                        }
                        GOTO_ERR(json_get_bool_value_by_key(&parameter_list_token, "filter", &cur_func_list->function.alert_condition.filter) != IOTCS_RESULT_OK);

                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "actionCondition") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_ACTION_CONDITION;

                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        GOTO_ERR(json_get_string_value_by_key(&parameter_list_token, "condition", &value, &len) != IOTCS_RESULT_OK);
                        CHECK_OOM(cur_func_list->function.action_condition.condition = util_safe_strncpy(value, len));

                        if ((rv = json_get_object_by_key(&parameter_list_token, "action", &sub_object_token)) == IOTCS_RESULT_OK) {
                            GOTO_ERR(json_get_string_value_by_key(&sub_object_token, "name", &value, &len) != IOTCS_RESULT_OK);
                            CHECK_OOM(cur_func_list->function.action_condition.action = util_safe_strncpy(value, len));

                            GOTO_ERR((rv = json_get_array_by_key(&sub_object_token, "arguments", &parameter_list_current_token)) != IOTCS_RESULT_OK);
                            GOTO_ERR_NO_OK(rv, json_get_array_iterator(&parameter_list_current_token, &param_arr_iter));
                            cur_func_list->function.action_condition.fields_len = json_child_items_count(&parameter_list_current_token);
                            if (cur_func_list->function.action_condition.fields_len) {
                                CHECK_OOM(cur_func_list->function.action_condition.arguments = util_malloc(cur_func_list->function.action_condition.fields_len * sizeof (iotcs_key_value)));
                                act_param = cur_func_list->function.action_condition.arguments;
                                while (json_array_iterator_has_next(&param_arr_iter)) {
                                    GOTO_ERR_NO_OK(rv, json_array_iterator_next(&param_arr_iter, &sub_object_token));
                                    GOTO_ERR_NO_OK(rv, json_get_string_and_move(&sub_object_token, &value, &len));
                                    CHECK_OOM(act_param->key = util_safe_strcpy("value"));
                                    CHECK_OOM(act_param->value = util_safe_strncpy(value, len));
                                    act_param++;
                                }
                            } else {
                                cur_func_list->function.action_condition.arguments = NULL;
                            }
                        }
                        rv = json_get_bool_value_by_key(&parameter_list_token, "filter", &cur_func_list->function.action_condition.filter);
                        GOTO_ERR(rv != IOTCS_RESULT_OK && rv != IOTCS_RESULT_INVALID_ARGUMENT);
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "filterCondition") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_FILTER_CONDITION;
                
                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        GOTO_ERR(json_get_string_value_by_key(&parameter_list_token, "condition", &value, &len) != IOTCS_RESULT_OK);
                        CHECK_OOM(cur_func_list->function.filter_condition.condition = util_safe_strncpy(value, len));
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "computedMetric") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_COMPUTED_METRIC;

                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        GOTO_ERR(json_get_string_value_by_key(&parameter_list_token, "formula", &value, &len) != IOTCS_RESULT_OK);
                        CHECK_OOM(cur_func_list->function.computed_metric.formula = util_safe_strncpy(value, len));
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "batchByTime") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_BATCH_TIME;

                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        GOTO_ERR(json_get_string_value_by_key(&parameter_list_token, "alertSeverity", &value, &len) != IOTCS_RESULT_OK);
                        CHECK_OOM(cur_func_list->function.batch_time.severity = util_safe_strncpy(value, len));
                        if (strcmp(cur_func_list->function.batch_time.severity, "CRITICAL") == 0) {
                        g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                        } else if (strcmp(cur_func_list->function.batch_time.severity, "SIGNIFICANT") == 0) {
                            g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                        } else if (strcmp(cur_func_list->function.batch_time.severity, "NORMAL") == 0) {
                            g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_NORMAL;
                        } else if (strcmp(cur_func_list->function.batch_time.severity, "LOW") == 0) {
                            g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_LOW;
                        } else {
                            g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                        }
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "delayLimit", &cur_func_list->function.batch_time.delay)) == IOTCS_RESULT_OK) {
                            g_batch_message_time = cur_func_list->function.batch_time.delay;
                        }
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            } else if (strcmp(value, "batchBySize") == 0) {
                cur_func_list->type = IOTCS_FUNCTION_TYPE_BATCH_SIZE;

                if ((rv = json_get_object_by_key(&function_token, "parameters", &parameter_list_token)) == IOTCS_RESULT_OK) {
                    len = json_child_items_count(&parameter_list_token);
                    if (len > 0) {
                        GOTO_ERR(json_get_string_value_by_key(&parameter_list_token, "alertSeverity", &value, &len) != IOTCS_RESULT_OK);
                        CHECK_OOM(cur_func_list->function.batch_size.severity = util_safe_strncpy(value, len));
                        if (strcmp(cur_func_list->function.batch_size.severity, "CRITICAL") == 0) {
                        g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                        } else if (strcmp(cur_func_list->function.batch_size.severity, "SIGNIFICANT") == 0) {
                            g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                        } else if (strcmp(cur_func_list->function.batch_size.severity, "NORMAL") == 0) {
                            g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_NORMAL;
                        } else if (strcmp(cur_func_list->function.batch_size.severity, "LOW") == 0) {
                            g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_LOW;
                        } else {
                            g_expected_alert_severity = IOTCS_MESSAGE_SEVERITY_SIGNIFICANT;
                        }
                        if ((rv = json_get_int_value_by_key(&parameter_list_token, "batchSize", &cur_func_list->function.batch_size.size)) == IOTCS_RESULT_OK) {
                            g_batch_message_size = cur_func_list->function.batch_size.size;
                        }
                    }
                } else if (rv != IOTCS_RESULT_INVALID_ARGUMENT) {
                    goto error;
                }
            }
        }
    }
    *policy = cur_policy;
    return IOTCS_RESULT_OK;
    error:
    // clear memory
    util_free(cur_func_list);
    util_free(cur_policy);
    util_free(cur_pipe);

    return IOTCS_RESULT_FAIL;
}

iotcs_result dp_add_policy_fromJSON(char *endpoint_id, char *device_model, json_tokens_t* tok) {
    iotcs_policy_handle_t *new_list = NULL;
    json_tokens_t policies_tok, token;
    json_array_iterator_t policy_iter;
    iotcs_policy_handle_t *policy = NULL, *cur_policy = NULL;
    iotcs_result rv;

    if ((rv = json_get_array_by_key(tok, "items", &policies_tok)) == IOTCS_RESULT_INVALID_ARGUMENT) {
        return IOTCS_RESULT_INVALID_ARGUMENT;
    } else {
        GOTO_ERR(rv != IOTCS_RESULT_OK);
    }
    GOTO_ERR_NO_OK(rv, json_get_array_iterator(&policies_tok, &policy_iter));
    if (!json_array_iterator_has_next(&policy_iter)) {
        return IOTCS_RESULT_INVALID_ARGUMENT;
    }
    while (json_array_iterator_has_next(&policy_iter)) {
        GOTO_ERR_NO_OK(rv, json_array_iterator_next(&policy_iter, &token));
        if (dp_parse_policy_fromJSON(device_model, &token, &policy) != IOTCS_RESULT_OK) {
            // Faile to parse
        } else {
            policy->endpoint_id = util_safe_strcpy(endpoint_id);
            put_policy_id_for_endpoint(endpoint_id, device_model, policy->id);
            // First success iteration
            if (!new_list) {
                new_list = policy;
                cur_policy = new_list;
            } else {
                cur_policy->next = policy;
                cur_policy = policy;
            }
        }
    }
    if (g_policy_list) {
        cur_policy = g_policy_list;
        while (cur_policy->next) {
            cur_policy = cur_policy->next;
        }
        cur_policy->next = new_list;
    } else {
        g_policy_list = new_list;
    }
    return IOTCS_RESULT_OK;
    error:
    return IOTCS_RESULT_FAIL;
}

iotcs_result dp_put_policy_fromJSON(char *endpoint_id, char *device_model, json_tokens_t* tok) {
    iotcs_policy_handle_t *new_list = NULL;
    json_tokens_t policies_tok, token;
    json_array_iterator_t policy_iter;
    char *value = NULL, *id = NULL;
    int len = 0;
    iotcs_policy_handle_t *policy = NULL, *cur_policy = NULL, *prev_policy = NULL, *tmp = NULL, *tail_new = NULL;
    iotcs_result rv;

    if ((rv = json_get_array_by_key(tok, "items", &policies_tok)) == IOTCS_RESULT_INVALID_ARGUMENT) {
        return IOTCS_RESULT_INVALID_ARGUMENT;
    } else {
        GOTO_ERR(rv != IOTCS_RESULT_OK);
    }
    GOTO_ERR_NO_OK(rv, json_get_array_iterator(&policies_tok, &policy_iter));
    if (!json_array_iterator_has_next(&policy_iter)) {
        return IOTCS_RESULT_INVALID_ARGUMENT;
    }
    while (json_array_iterator_has_next(&policy_iter)) {
        GOTO_ERR_NO_OK(rv, json_array_iterator_next(&policy_iter, &token));
        GOTO_ERR(json_get_string_value_by_key(&token, "id", &value, &len) != IOTCS_RESULT_OK);
        CHECK_OOM(id = util_safe_strncpy(value, len));
        if (g_policy_list) {
            cur_policy = g_policy_list;
            while (cur_policy) {
                if (strcmp(cur_policy->id, id) == 0 &&
                    strcmp(cur_policy->endpoint_id, endpoint_id) == 0) {
                    if (!prev_policy) {
                        tmp = g_policy_list;
                        g_policy_list = g_policy_list->next;
                    } else {
                        prev_policy->next = cur_policy->next;
                        tmp = cur_policy;
                    }
                } else {
                    prev_policy = cur_policy;
                }
                cur_policy = cur_policy->next;
                dp_release_policy(tmp);
                tmp = NULL;
            }
        }

        if (dp_parse_policy_fromJSON(device_model, &token, &policy) != IOTCS_RESULT_OK) {
            // Faile to parse
        } else {
            policy->endpoint_id = util_safe_strcpy(endpoint_id);
            put_policy_id_for_endpoint(endpoint_id, device_model, policy->id);
            // First success iteration
            if (!new_list) {
                new_list = policy;
                policy->next = NULL;
                tail_new = policy;
            } else {
                policy->next = new_list;
                new_list = policy;
            }
        }
    }
    if (new_list) {
        if (g_policy_list) {
            tail_new->next = g_policy_list;
            g_policy_list = new_list;
        } else {
            g_policy_list = new_list;
        }
    }
    return IOTCS_RESULT_OK;
    error:
    return IOTCS_RESULT_FAIL;
}

iotcs_result proceed_policy_request(char *url_buffer, char** device_policy_json, protocol_response *response){
    iotcs_result result = IOTCS_RESULT_FAIL;

    protocol_request request = {
        .body = "",
        .url = url_buffer,
        .method = PROTOCOL_REQUEST_METHOD_GET,
        .headers =
        {
            .connection = "close",
            .content_type = "application/json",
            .host = tam_get_server_host(),
            .x_endpointId = tam_get_endpoint_id()
        }
    };

    result = cl_proceed_request(request, response);

    if (result == IOTCS_RESULT_OK) {
        if (response->status_code == PROTOCOL_RESPONSE_CODE_OK) {
            *device_policy_json = response->body;
            result = IOTCS_RESULT_OK;
        }
    }

    return result;
}

iotcs_result dp_policies_from_server_by_endpoint(const char* endpoint_id, const char* device_model_urn, char** device_policy_json) {
    iotcs_result result = IOTCS_RESULT_FAIL;
    protocol_response response;
    char *query = NULL;
    char* url_buffer = util_get_url_buffer();
    int expected_len = strlen("{\"devices.id\":\"%s\"}&fields=id,pipelines,enabled,lastModified") + strlen(endpoint_id) + 1;
    int tmp_cx, cx = 0;

    CHECK_OOM(query = util_malloc(sizeof(char) * expected_len));
    GOTO_ERR(0 > util_safe_snprintf(query, expected_len, "{\"devices.id\":\"%s\"}&fields=id,pipelines,enabled,lastModified", endpoint_id));
    GOTO_ERR(0 > (tmp_cx = util_safe_snprintf(url_buffer, UTIL_REST_API_URL_MAX_LENGTH, "/iot/privateapi/v2/deviceModels/%s/devicePolicies?q=", device_model_urn)));
    cx += tmp_cx;
    GOTO_ERR(0 > (tmp_cx = util_urlencode(query, url_buffer + cx, UTIL_REST_API_URL_MAX_LENGTH - cx)));
    cx += tmp_cx;

    result = proceed_policy_request(url_buffer, device_policy_json, &response);

    util_free(query);
    return result;
    error:
    util_free(query);
    return result;
}


iotcs_result dp_policy_from_server_by_policy(const char* policy_id, const char* device_model_urn, char** device_policy_json) {
    iotcs_result result = IOTCS_RESULT_FAIL;
    protocol_response response;
    char *query = NULL;
    char* url_buffer = util_get_url_buffer();
    int expected_len = strlen("{\"id\":\"%s\"}&fields=id,pipelines,enabled,lastModified") + strlen(policy_id) + 1;
    int tmp_cx, cx = 0;

    CHECK_OOM(query = util_malloc(sizeof(char) * expected_len));
    GOTO_ERR(0 > util_safe_snprintf(query, expected_len, "{\"id\":\"%s\"}&fields=id,pipelines,enabled,lastModified", policy_id));
    GOTO_ERR(0 > (tmp_cx = util_safe_snprintf(url_buffer, UTIL_REST_API_URL_MAX_LENGTH, "/iot/privateapi/v2/deviceModels/%s/devicePolicies?q=", device_model_urn)));
    cx += tmp_cx;
    GOTO_ERR(0 > (tmp_cx = util_urlencode(query, url_buffer + cx, UTIL_REST_API_URL_MAX_LENGTH - cx)));
    cx += tmp_cx;

    result = proceed_policy_request(url_buffer, device_policy_json, &response);
    util_free(query);

    return result;
    error:
    util_free(query);
    return result;
}

iotcs_bool dp_check_is_assigned_policy_from_server(const char* device_model_urn, const char* policy_id, char* gw_endpoint_id) {
    protocol_response response;
    char *query = NULL;
    char* url_buffer = util_get_url_buffer();
    int expected_len = strlen("{\"devices.id\":\"%s\"}&fields=id") + strlen(gw_endpoint_id) + 1;
    int tmp_cx, cx = 0;
    json_tokens_t tokens, policies_tok;
    char* device_policy_json;

    CHECK_OOM(query = util_malloc(sizeof(char) * expected_len));
    GOTO_ERR(0 > util_safe_snprintf(query, expected_len, "{\"devices.id\":\"%s\"}&fields=id", gw_endpoint_id));
    GOTO_ERR(0 > (tmp_cx = util_safe_snprintf(url_buffer, UTIL_REST_API_URL_MAX_LENGTH, "/iot/privateapi/v2/deviceModels/%s/devicePolicies?q=", device_model_urn)));
    cx += tmp_cx;
    GOTO_ERR(0 > (tmp_cx = util_urlencode(query, url_buffer + cx, UTIL_REST_API_URL_MAX_LENGTH - cx)));
    cx += tmp_cx;

    GOTO_ERR(IOTCS_RESULT_OK != proceed_policy_request(url_buffer, &device_policy_json, &response));
    GOTO_ERR(IOTCS_RESULT_OK != json_parse(device_policy_json, strlen(device_policy_json), &tokens));
    GOTO_ERR(IOTCS_RESULT_OK != json_get_array_by_key(&tokens, "items", &policies_tok));
    tmp_cx = json_child_items_count(&policies_tok);

    util_free(query);
    return (tmp_cx > 0 ? IOTCS_TRUE : IOTCS_FALSE);
    error:
    util_free(query);
    return IOTCS_FALSE;
}

iotcs_bool dp_check_is_assigned_icd_policy_from_server(json_tokens_t *policies_tok, const char* device_model_urn, char* policy_id, char* gw_endpoint_id, 
    protocol_response *response) {
    iotcs_result result = IOTCS_RESULT_FAIL;
    char *query = NULL;
    char* url_buffer = util_get_url_buffer();
    int expected_len = strlen("{\"directlyConnectedOwner\":\"%s\"}&fields=id") + strlen(gw_endpoint_id) + 1;
    int tmp_cx, cx = 0;
    json_tokens_t tokens;
    char* device_policy_json;

    CHECK_OOM(query = util_malloc(sizeof(char) * expected_len));
    GOTO_ERR(0 > util_safe_snprintf(query, expected_len, "{\"directlyConnectedOwner\":\"%s\"}&fields=id", gw_endpoint_id));
    GOTO_ERR(0 > (tmp_cx = util_safe_snprintf(url_buffer, UTIL_REST_API_URL_MAX_LENGTH, "/iot/privateapi/v2/deviceModels/%s/devicePolicies/%s/devices?q=", device_model_urn, policy_id)));
    cx += tmp_cx;
    GOTO_ERR(0 > (tmp_cx = util_urlencode(query, url_buffer + cx, UTIL_REST_API_URL_MAX_LENGTH - cx)));
    cx += tmp_cx;

    GOTO_ERR_MSG(IOTCS_RESULT_OK != (result = proceed_policy_request(url_buffer, &device_policy_json, response)),
                "Handle policyChange request: can not get ICD list");

    GOTO_ERR_MSG(IOTCS_RESULT_OK != (result = json_parse(device_policy_json, strlen(device_policy_json), &tokens)),
            "Handle policyChange request: can not get ICD list");
    GOTO_ERR_MSG(IOTCS_RESULT_OK != (result = json_get_array_by_key(&tokens, "items", policies_tok)),
            "Handle policyChange request: can not get ICD list");


    util_free(query);
    return (result == IOTCS_RESULT_OK);
    error:
    util_free(query);
    return IOTCS_FALSE;
}

void dp_delete_policy(const char* policy_id, const char* dm_urn) {
    iotcs_policy_handle_t *policy = NULL, *cur_policy = NULL, *tmp = NULL;
#ifdef IOTCS_GATEWAY
    json_tokens_t policies_tok, token;
    json_array_iterator_t policy_iter;
    protocol_response response;
    iotcs_bool is_found = IOTCS_FALSE;
    iotcs_bool icd_check = IOTCS_FALSE;
    char *value;
    int len;
#endif
    char *endpoint_id = tam_get_endpoint_id();

    if (!g_policy_list) {
        return;
    }

#ifdef IOTCS_GATEWAY
    icd_check = dp_check_is_assigned_icd_policy_from_server(&policies_tok, dm_urn, policy_id, endpoint_id, &response);
#endif
    cur_policy = g_policy_list;
    while (cur_policy) {
        if (strcmp(cur_policy->id, policy_id) == 0) {
#ifdef IOTCS_GATEWAY
            if (strcmp(cur_policy->endpoint_id, endpoint_id) == 0) {
                is_found = dp_check_is_assigned_policy_from_server(dm_urn, policy_id, endpoint_id);
            } else if (icd_check) {
                if (json_get_array_iterator(&policies_tok, &policy_iter) != IOTCS_RESULT_OK) {
                    break;
                }
                is_found = IOTCS_FALSE;
                while (json_array_iterator_has_next(&policy_iter)) {
                    char* icd_endpoint_id = NULL;
                    if (json_array_iterator_next(&policy_iter, &token) != IOTCS_RESULT_OK) {
                        break;
                    }
                    if (json_get_string_value_by_key(&token, "id", &value, &len) == IOTCS_RESULT_OK) {
                        if (strncmp(cur_policy->endpoint_id, value, len) == 0) {
                            is_found = IOTCS_TRUE;
                            break;
                        }
                    }
                }
            } else {
                is_found = IOTCS_FALSE;
            }

            if (!is_found) {
#else
            if (strcmp(cur_policy->endpoint_id, endpoint_id) == 0) {
#endif
                if (policy) {
                    policy->next = cur_policy->next;
                } else {
                    g_policy_list = cur_policy->next;
                }
                tmp = cur_policy;
                put_policy_id_for_endpoint(cur_policy->endpoint_id, dm_urn, NULL);
                cur_policy = cur_policy->next;
                dp_release_policy(tmp);
                tmp = NULL;
                continue;
            }
        }
        policy = cur_policy;
        cur_policy = cur_policy->next;
    }
}

void dp_put_policy(const char* policy_id, const char* target_dm_urn) {
    char *policy_buffer = NULL;
    char *endpoint_id = tam_get_endpoint_id();
#ifdef IOTCS_GATEWAY
    char *icd_endpoint_id = NULL;
    json_tokens_t policies_tok, token;
    json_array_iterator_t policy_iter;
    protocol_response response;
    iotcs_bool icd_check = IOTCS_FALSE;
    char *value;
    int len;
#endif
    int i = 0;
    iotcs_result result;
    json_tokens_t tokens;

    if (dp_check_is_assigned_policy_from_server(target_dm_urn, policy_id, endpoint_id)) {
        put_policy_id_for_endpoint(endpoint_id, target_dm_urn, policy_id);
    }

#ifdef IOTCS_GATEWAY
    icd_check = dp_check_is_assigned_icd_policy_from_server(&policies_tok, target_dm_urn, policy_id, endpoint_id, &response);
    if (icd_check) {
        if (json_get_array_iterator(&policies_tok, &policy_iter) == IOTCS_RESULT_OK) {
            while (json_array_iterator_has_next(&policy_iter)) {
                icd_endpoint_id = NULL;
                if (json_array_iterator_next(&policy_iter, &token) != IOTCS_RESULT_OK) {
                    break;
                }
                if (json_get_string_value_by_key(&token, "id", &value, &len) == IOTCS_RESULT_OK) {
                    icd_endpoint_id = util_safe_strncpy(value, len);
                    put_policy_id_for_endpoint(icd_endpoint_id, target_dm_urn, policy_id);
                    util_free(icd_endpoint_id);
                }
            }
        }
    }
#endif

    i = 0;
    endpoint_id = get_next_endpoint_by_policy_id(policy_id, i);
    dp_policy_from_server_by_policy(policy_id, target_dm_urn, &policy_buffer);
    while (endpoint_id) {
        i++;
        if (!policy_buffer ||
            json_parse(policy_buffer, strlen(policy_buffer), &tokens) != IOTCS_RESULT_OK) {
            policy_buffer = NULL;
        }
        if (!policy_buffer ||
            dp_put_policy_fromJSON(endpoint_id, target_dm_urn, &tokens) != IOTCS_RESULT_OK) {
            LOG_INFO("Can not parse policy for id=\"%s\", urn=\"%s\"", endpoint_id, target_dm_urn);
        }
        endpoint_id = get_next_endpoint_by_policy_id(policy_id, i);
    }
}

iotcs_result dp_policy_changed_handle_unsafe(const char* endpoint_id, iotcs_request_message *request) {
    int i = 0;
    iotcs_result result;
    json_tokens_t tokens, token;
    json_array_iterator_t policy_iter;
    char *policy_id, *value, *dm_urn;
    int len;

    UTIL_POLICY_LOCK();
    if (request) {
        GOTO_ERR_MSG(IOTCS_RESULT_OK != (result = json_parse(request->body, strlen(request->body), &tokens)),
                "Handle policyChange request: json parse failed");
        GOTO_ERR_MSG(IOTCS_RESULT_OK != (result = json_get_array_iterator(&tokens, &policy_iter)),
                "Handle policyChange request: json parse failed");
        while (json_array_iterator_has_next(&policy_iter)) {
            GOTO_ERR_MSG(IOTCS_RESULT_OK != (result = json_array_iterator_next(&policy_iter, &token)),
                "Handle policyChange request: json parse failed");
            GOTO_ERR_MSG(IOTCS_RESULT_OK != (result = json_get_string_value_by_key(&token, "id", &value, &len) != IOTCS_RESULT_OK),
                "Handle policyChange request: json parse failed");
            CHECK_OOM(policy_id = util_safe_strncpy(value, len));
            GOTO_ERR_MSG(IOTCS_RESULT_OK != (result = json_get_string_value_by_key(&token, "deviceModelUrn", &value, &len) != IOTCS_RESULT_OK),
                "Handle policyChange request: json parse failed");
            CHECK_OOM(dm_urn = util_safe_strncpy(value, len));
            GOTO_ERR_MSG(IOTCS_RESULT_OK != (result = json_get_string_value_by_key(&token, "op", &value, &len) != IOTCS_RESULT_OK),
                "Handle policyChange request: json parse failed");
            if (strncmp(value, "unassigned", len) == 0) {
                dp_delete_policy(policy_id, dm_urn);
            } else if (strncmp(value, "assigned", len) == 0) {
                dp_put_policy(policy_id, dm_urn);
            } else if (strncmp(value, "changed", len) == 0) {
                dp_put_policy(policy_id, dm_urn);
            } else {
                LOG_WARNS("Handle policyChange request: Unknown op");
            }
        }
    }

    for (i = 0; i < WINDOWS_COUNT; i++) {
        util_free(window_array[i].attribute_name);
        window_array[i].attribute_name = NULL;
        window_array[i].device_handle = NULL;
        window_array[i].endpoint_id = NULL;

        window_array[i].value_index = 0;
        window_array[i].size = 0;
        util_free(window_array[i].typed_value);
        window_array[i].typed_value = NULL;
        util_free(window_array[i].count);
        window_array[i].count = NULL;

        window_array[i].args = NULL;
        window_array[i].sheduled_time = 0;
        window_array[i].time_delta = 0;
        window_array[i].cur_function = NULL;
        release_msg_copy(window_array[i].message);
        window_array[i].message = NULL;
    }
    UTIL_POLICY_UNLOCK();
    return IOTCS_RESULT_OK;
    error:
    UTIL_POLICY_UNLOCK();
    return result;
}

iotcs_result dp_policy_changed_handle(const char *endpoint_id, iotcs_request_message *request) {
    iotcs_result result;
    UTIL_LOCK_LL();
    result = dp_policy_changed_handle_unsafe(endpoint_id, request);
    UTIL_UNLOCK_LL();
    return result;
}

iotcs_result dp_get_policy_unsafe(const char* endpoint_id, const char* device_model_urn) {
    char *policy_buffer = NULL;
    json_tokens_t tokens;

    GOTO_ERR(dp_policies_from_server_by_endpoint(endpoint_id, device_model_urn, &policy_buffer) != IOTCS_RESULT_OK);
    GOTO_ERR(!policy_buffer);
    GOTO_ERR(json_parse(policy_buffer, strlen(policy_buffer), &tokens) != IOTCS_RESULT_OK);
    GOTO_ERR(dp_add_policy_fromJSON(endpoint_id, device_model_urn, &tokens) != IOTCS_RESULT_OK);
    return IOTCS_RESULT_OK;

    error:
    return IOTCS_RESULT_FAIL;
}

iotcs_result dp_init_policy() {
    int i;
    UTIL_INIT_POLICY_LOCK();
    if (!g_policy_thread) {
        g_policy_thread = iotcs_port_thread_create_small_stack(policy_thread);
    }
    for (i = 0; i < WINDOWS_COUNT; i++) {
        window_array[i].attribute_name = NULL;
        window_array[i].device_handle = NULL;
        window_array[i].endpoint_id = NULL;
        window_array[i].value_index = 0;
        window_array[i].size = 0;
        window_array[i].typed_value = NULL;
        window_array[i].count = NULL;
        window_array[i].last_offer_time = -0;
        window_array[i].args = NULL;
        window_array[i].sheduled_time = 0;
        window_array[i].time_delta = 0;
        window_array[i].cur_function = NULL;
        window_array[i].message = NULL;
    }

    return IOTCS_RESULT_OK;
}
