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

#include "device_model/device_model_capability.h"
#include "advanced/iotcs_message.h"
#include "advanced/iotcs_messaging.h"
#include "messaging/msg_private.h"
#include "device_model/device_model_private.h"
#include "iotcs/iotcs_private.h"
#include "util/util_thread_private.h"
#include "trusted_assets_manager/iotcs_tam.h"
#include "util/util_memory.h"
#include "iotcs_port_system.h"
#include "iotcs_port_thread.h"
#include "iotcs_port_diagnostic.h"
#include "util/util.h"
#include "protocol/http/http_wrapper.h"

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

#ifndef ARRAY_SIZE
#define ARRAY_SIZE(a) (sizeof(a) / sizeof((a)[0]))
#endif

#ifndef IOTCS_CAPABILITY_RESPONSE_BUFFER_SIZE
#define IOTCS_CAPABILITY_RESPONSE_BUFFER_SIZE 1024
#endif

#ifndef IOTCSP_CAPABILITY_TMP_BUF_SZ
#define IOTCSP_CAPABILITY_TMP_BUF_SZ 32
#endif

#ifdef IOTCS_MESSAGE_DISPATCHER
#define DM_CAPABILITY_URN_MESSAGE_DISPATCHER_COUNTERS "deviceModels/urn:oracle:iot:dcd:capability:message_dispatcher/counters"
#define DM_CAPABILITY_URN_MESSAGE_DISPATCHER_POLLING_INTERVAL "deviceModels/urn:oracle:iot:dcd:capability:message_dispatcher/pollingInterval"
#define DM_CAPABILITY_URN_MESSAGE_DISPATCHER_RESET "deviceModels/urn:oracle:iot:dcd:capability:message_dispatcher/reset"
#endif

#define DM_CAPABILITY_URN_DIAGNOSTIC_INFO "deviceModels/urn:oracle:iot:dcd:capability:diagnostics/info"

#ifdef IOTCS_MESSAGING_THREAD_SAFETY
#define DM_CAPABILITY_URN_DIAGNOSTIC_TEST_CONNECTIVITY "deviceModels/urn:oracle:iot:dcd:capability:diagnostics/testConnectivity"
#endif

#define DM_TEST_CONNECTIVITY_MAX_TIME_TO_SLEEP_MS 100

typedef iotcs_result(*capability_handler)(iotcs_request_message *request);

typedef struct {
    const char *urn;
    capability_handler handler;
} dm_capability_handler_desc;

#ifdef IOTCS_MESSAGE_DISPATCHER
static iotcs_result md_counters_request_handler(iotcs_request_message *request);
static iotcs_result md_polling_interval_request_handler(iotcs_request_message *request);
static iotcs_result md_counter_reset_request_handler(iotcs_request_message *request);
#endif
static iotcs_result dgn_info_handler(iotcs_request_message *request);
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
static iotcs_result dgn_test_connectivity_handler(iotcs_request_message *request);
static iotcs_result dgn_start_stop_test_handler(iotcs_request_message *request);
static iotcs_result dgn_status_handler(iotcs_request_message *request);
static void reset_connectivity_test_fields(void);
static void connectivity_test_finalize(void);
#endif

static iotcs_result register_capabilities_resources(void);
static void unregister_capabilities_resources(void);

static volatile char g_resource_registered = 0x0;

static dm_capability_handler_desc g_capability_handler_descs[] = {
#ifdef IOTCS_MESSAGE_DISPATCHER
    {DM_CAPABILITY_URN_MESSAGE_DISPATCHER_COUNTERS, md_counters_request_handler},
    {DM_CAPABILITY_URN_MESSAGE_DISPATCHER_POLLING_INTERVAL, md_polling_interval_request_handler},
    {DM_CAPABILITY_URN_MESSAGE_DISPATCHER_RESET, md_counter_reset_request_handler},
#endif
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    {DM_CAPABILITY_URN_DIAGNOSTIC_TEST_CONNECTIVITY, dgn_test_connectivity_handler},
#endif
    {DM_CAPABILITY_URN_DIAGNOSTIC_INFO, dgn_info_handler}
};

iotcs_result device_model_process_capability_request(iotcs_request_message *request) {
    int i;
    dm_capability_handler_desc *desc;

    if (!request || !g_resource_registered) {
        return IOTCS_RESULT_FAIL;
    }

    for (i = 0; i < (int) ARRAY_SIZE(g_capability_handler_descs); i++) {
        desc = g_capability_handler_descs + i;
        if (strcmp(desc->urn, request->url) == 0) {
            (*desc->handler)(request);
            return IOTCS_RESULT_OK;
        }
    }

    return IOTCS_RESULT_FAIL;
}

void device_model_capability_initilize(void) {
    register_capabilities_resources();
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    reset_connectivity_test_fields();
#endif
}

void device_model_capability_finalize(void) {
    UTIL_LOCK_LL();
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    connectivity_test_finalize();
#endif
    unregister_capabilities_resources();
    UTIL_UNLOCK_LL();
}

static iotcs_resource_message_base capabilities_resources[] = {
#ifdef IOTCS_MESSAGE_DISPATCHER
    {
        .path = DM_CAPABILITY_URN_MESSAGE_DISPATCHER_COUNTERS,
        .name = "counters",
        .methods = IOTCS_REQUEST_METHOD_GET,
        .description = "Retrieve all counters from the gateway."
    },
    {
        .path = DM_CAPABILITY_URN_MESSAGE_DISPATCHER_POLLING_INTERVAL,
        .name = "polling_interval",
        .methods = IOTCS_REQUEST_METHOD_GET | IOTCS_REQUEST_METHOD_PUT,
        .description = "Gets/sets the incoming message polling interval in seconds on the directly connected device."
    },
    {
        .path = DM_CAPABILITY_URN_MESSAGE_DISPATCHER_RESET,
        .name = "reset",
        .methods = IOTCS_REQUEST_METHOD_PUT,
        .description = "Resets the counters to 0 in the gateway software"
    },
#endif
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    {
        .path = DM_CAPABILITY_URN_DIAGNOSTIC_TEST_CONNECTIVITY,
        .name = "diagnostic_test_connectivity",
        .methods = IOTCS_REQUEST_METHOD_GET | IOTCS_REQUEST_METHOD_PUT,
        .description = "Test connectivity."
    },
#endif
    {
        .path = DM_CAPABILITY_URN_DIAGNOSTIC_INFO,
        .name = "diagnostic_info",
        .methods = IOTCS_REQUEST_METHOD_GET,
        .description = "Retrieve all diagnostic information from the device."
    }
};

static iotcs_result register_capabilities_resources(void) {
    iotcs_result status;
    const char *endpoint_id = tam_get_endpoint_id();
    iotcs_message_base resource_msg_base = {
        .priority = IOTCS_MESSAGE_PRIORITY_HIGHEST,
        .reliability = IOTCS_MESSAGE_RELIABILITY_GUARANTED_DELIVERY,
        .type = IOTCS_MESSAGE_RESOURCE
    };

    iotcs_message resources = {
        .base = &resource_msg_base,
        .u.resource.base = capabilities_resources,
        .u.resource.resource_len = (int) ARRAY_SIZE(capabilities_resources),
        .u.resource.endpointName = endpoint_id,
        .u.resource.report_type = IOTCS_RESOURCE_MESSAGE_UPDATE
    };

    status = cl_internal_send_unsafe(&resources, 1, NULL, NULL);
    if (status == IOTCS_RESULT_OK)
        g_resource_registered = 0x1;

    return status;
}

static void unregister_capabilities_resources(void) {
    const char* endpoint_id;
    if (!g_resource_registered)
        return;
    endpoint_id = tam_get_endpoint_id();
    iotcs_message_base resource_msg_base = {
        .priority = IOTCS_MESSAGE_PRIORITY_HIGHEST,
        .reliability = IOTCS_MESSAGE_RELIABILITY_GUARANTED_DELIVERY,
        .type = IOTCS_MESSAGE_RESOURCE
    };

    iotcs_message resources = {
        .base = &resource_msg_base,
        .u.resource.base = capabilities_resources,
        .u.resource.resource_len = (sizeof (capabilities_resources)
        / sizeof (capabilities_resources[0])),
        .u.resource.endpointName = endpoint_id,
        .u.resource.report_type = IOTCS_RESOURCE_MESSAGE_DELETE
    };

    if (cl_internal_send_unsafe(&resources, 1, NULL, NULL) != IOTCS_RESULT_OK) {
        LOG_ERRS("Can not unregister capabilities resources!");
    }
    g_resource_registered = 0x0;
}

static iotcs_message_base response_base = {
    .priority = IOTCS_MESSAGE_PRIORITY_HIGHEST,
    .reliability = IOTCS_MESSAGE_RELIABILITY_GUARANTED_DELIVERY,
    .type = IOTCS_MESSAGE_RESPONSE
};

#ifdef IOTCS_MESSAGE_DISPATCHER

static int md_counters_json_callback(json_buf *buf, int pos, void* data) {
    int rv;
    int original_pos = pos;
    cl_message_dispatcher_counters *counters = cl_message_dispatcher_counters_get();
    (void) data;

    rv = json_write_float_field(buf, pos, "load", counters->load);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "totalMessagesSent", counters->total_messages_sent);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "totalMessagesReceived", counters->total_messages_received);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "totalMessagesRetried", counters->total_messages_retried);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "totalBytesSent", counters->total_bytes_sent);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "totalBytesReceived", counters->total_bytes_received);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "totalProtocolErrors", counters->total_protocols_errors);
    JSON_MOVE_POS(pos, rv);

    return pos - original_pos;

error:
    return -1;
}

iotcs_result md_counters_request_handler(iotcs_request_message *request) {
    int rv, status_code = PROTOCOL_RESPONSE_CODE_OK;
    json_buf buf_desc = {NULL, 0};
    iotcs_message response = {0};
    iotcs_result result = IOTCS_RESULT_FAIL;

    if (request->method != IOTCS_REQUEST_METHOD_GET) {
        LOG_ERRS("Unexpected HTTP method for message dispatcher /counters");
        return IOTCS_RESULT_FAIL;
    }

    buf_desc.buf = (char*) util_malloc(IOTCS_CAPABILITY_RESPONSE_BUFFER_SIZE);
    GOTO_ERR_CRIT(!buf_desc.buf);
    buf_desc.len = IOTCS_CAPABILITY_RESPONSE_BUFFER_SIZE;
    rv = json_write_object_null_terminated(&buf_desc, 0, md_counters_json_callback, NULL);
    result = IOTCS_RESULT_OUT_OF_MEMORY;
    GOTO_ERR(rv < 0);

    response.base = &response_base;
    response.u.response.status_code = status_code;
    response.u.response.request = request;
    request = NULL;
    response.u.response.body = buf_desc.buf;

    result = cl_internal_send(&response, 1, NULL, NULL);

error:
    util_free(buf_desc.buf);
    cl_internal_request_message_free(request);
    return result;
}

iotcs_result md_polling_interval_request_handler(iotcs_request_message *request) {
    int status_code = PROTOCOL_RESPONSE_CODE_OK;
    iotcs_message response = {0};
    char tmp_buf[IOTCSP_CAPABILITY_TMP_BUF_SZ];
    iotcs_result result = IOTCS_RESULT_FAIL;

    if (request->method != IOTCS_REQUEST_METHOD_GET && request->method != IOTCS_REQUEST_METHOD_PUT) {
        LOG_ERRS("Unexpected HTTP method for message dispatcher /pollingInterval");
        return IOTCS_RESULT_FAIL;
    }

    int64_t polling_interval = cl_message_dispatcher_get_polling_interval();

    if (request->method == IOTCS_REQUEST_METHOD_GET) {
        GOTO_ERR(0 > util_safe_snprintf(tmp_buf, IOTCSP_CAPABILITY_TMP_BUF_SZ, "{\"value\":%" PRId64 "}", polling_interval));
    } else {
        json_tokens_t _tokens;

        UTIL_LOCK_LL(); /* needed to json parser */
        if (IOTCS_RESULT_OK == (result = json_parse(request->body, strlen(request->body), &_tokens))) {
            result = json_get_int_value_by_key(&_tokens, "value", &polling_interval);
        }
        UTIL_UNLOCK_LL();

        GOTO_ERR(result != IOTCS_RESULT_OK);

        if (IOTCS_RESULT_OK != result) {
            status_code = PROTOCOL_RESPONSE_CODE_BAD_REQUEST;
        }
        UTIL_UNLOCK_LL();

        if (polling_interval >= 0) {
            cl_message_dispatcher_set_polling_interval(polling_interval);
        } else {
            status_code = PROTOCOL_RESPONSE_CODE_BAD_REQUEST;
        }

        tmp_buf[0] = '\0';
    }

    response.base = &response_base;
    response.u.response.status_code = status_code;
    response.u.response.request = request;
    request = NULL;
    response.u.response.body = tmp_buf;

    result = cl_internal_send(&response, 1, NULL, NULL);

error:

    cl_internal_request_message_free(request);
    return result;
}

iotcs_result md_counter_reset_request_handler(iotcs_request_message *request) {
    int status_code = PROTOCOL_RESPONSE_CODE_OK;
    iotcs_message response = {0};

    if (request->method != IOTCS_REQUEST_METHOD_PUT) {
        LOG_ERRS("Unexpected HTTP method for message dispatcher /reset");
        return IOTCS_RESULT_FAIL;
    }

    cl_message_dispatcher_counters_reset();

    response.base = &response_base;
    response.u.response.status_code = status_code;
    response.u.response.request = request;
    request = NULL;
    response.u.response.body = "{}";

    return cl_internal_send(&response, 1, NULL, NULL);
}
#endif

static int dgn_info_json_callback(json_buf *buf, int pos, void* data) {
    int rv;
    int original_pos = pos;
    (void) data;

#ifdef IOTCS_USE_DIAGNOSTIC_CAPABILITY
    char tmp_buf[IOTCSP_CAPABILITY_TMP_BUF_SZ];
    rv = json_write_int_field(buf, pos, "freeDiskSpace", iotcs_port_get_free_disk_space());
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_string_field(buf, pos, "ipAddress",
            iotcs_port_get_ip_address(tmp_buf, IOTCSP_CAPABILITY_TMP_BUF_SZ) <= 0 ? "N/A" : tmp_buf);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_string_field(buf, pos, "macAddress",
            iotcs_port_get_mac_address(tmp_buf, IOTCSP_CAPABILITY_TMP_BUF_SZ) <= 0 ? "N/A" : tmp_buf);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "startTime", iotcs_port_get_start_time());
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "totalDiskSpace", iotcs_port_get_total_disk_space());
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_string_field(buf, pos, "version", iotcs_port_get_version());
    JSON_MOVE_POS(pos, rv);
#else
    rv = json_write_int_field(buf, pos, "freeDiskSpace", -1);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_string_field(buf, pos, "ipAddress", "NOT_IMPLEMENTED");
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_string_field(buf, pos, "macAddress", "NOT_IMPLEMENTED");
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "startTime", -1);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "totalDiskSpace", -1);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_string_field(buf, pos, "version", "NOT_IMPLEMENTED");
    JSON_MOVE_POS(pos, rv);
#endif
    return pos - original_pos;

error:
    return -1;
}

iotcs_result dgn_info_handler(iotcs_request_message *request) {
    int rv, status_code = PROTOCOL_RESPONSE_CODE_OK;
    json_buf buf_desc = {NULL, 0};
    iotcs_message response = {0};
    iotcs_result result = IOTCS_RESULT_FAIL;

    if (request->method != IOTCS_REQUEST_METHOD_GET) {
        LOG_ERRS("Unexpected HTTP method for message dispatcher /info");
        return IOTCS_RESULT_FAIL;
    }

    buf_desc.buf = (char*) util_malloc(IOTCS_CAPABILITY_RESPONSE_BUFFER_SIZE);
    GOTO_ERR_CRIT(!buf_desc.buf);
    buf_desc.len = IOTCS_CAPABILITY_RESPONSE_BUFFER_SIZE;
    rv = json_write_object_null_terminated(&buf_desc, 0, dgn_info_json_callback, NULL);
    result = IOTCS_RESULT_OUT_OF_MEMORY;
    GOTO_ERR(rv < 0);

    response.base = &response_base;
    response.u.response.status_code = status_code;
    response.u.response.request = request;
    request = NULL;
    response.u.response.body = buf_desc.buf;

    result = cl_internal_send(&response, 1, NULL, NULL);

error:
    util_free(buf_desc.buf);
    cl_internal_request_message_free(request);
    return result;
}

#ifdef IOTCS_MESSAGING_THREAD_SAFETY
/* protected by LL lock */
static volatile iotcs_bool g_active;
static int64_t g_count;
static int64_t g_interval;
static size_t g_size;
static iotcs_port_thread g_connectivity_test_thread;
static int g_connectivity_test_stop = 0;

static iotcs_message_base data_msg_base = {
    .type = IOTCS_MESSAGE_DATA,
    .reliability = IOTCS_MESSAGE_RELIABILITY_BEST_EFFORT,
    .priority = IOTCS_MESSAGE_PRIORITY_HIGH,
};

static iotcs_data_message_base data_msg_props = {
    .format = "urn:oracle:iot:dcd:capability:diagnostics:test_message"
};

static iotcs_data_item_desc desc[] = {
    { .type = IOTCS_VALUE_TYPE_INT, .key = "current"},
    { .type = IOTCS_VALUE_TYPE_INT, .key = "count"},
    { .type = IOTCS_VALUE_TYPE_STRING, .key = "payload"},
    { .type = IOTCS_VALUE_TYPE_NONE, .key = NULL}
};

iotcs_result dgn_test_connectivity_handler(iotcs_request_message *request) {
    if (request->method == IOTCS_REQUEST_METHOD_GET) {
        return dgn_status_handler(request);
    } else if (request->method == IOTCS_REQUEST_METHOD_PUT) {
        return dgn_start_stop_test_handler(request);
    }

    LOG_ERRS("Unexpected HTTP method for DM_CAPABILITY_URN_DIAGNOSTIC_TEST_CONNECTIVITY handler.");
    return IOTCS_RESULT_FAIL;
}

static void reset_connectivity_test_fields(void) {
    g_active = IOTCS_FALSE;
    g_count = 0;
    g_interval = 0;
    g_size = 0;
}

static void* connectivity_test_thread(void *arg) {
    int rv = 0; /* 0 - no error, non zero - error */
    char* payload = NULL;
    iotcs_value data_msg_items[3];
    iotcs_message data_msg = {0};
    iotcs_bool is_stopped = IOTCS_FALSE;

    int64_t count, interval, time_to_wait, current_count;
    size_t size; //unsigned int is better for positive size
    UTIL_LOCK_LL();
    count = g_count;
    interval = g_interval;
    size = g_size;
    is_stopped = (g_connectivity_test_stop) ? IOTCS_TRUE : IOTCS_FALSE;
    UTIL_UNLOCK_LL();

    GOTO_ERR(is_stopped == IOTCS_TRUE);

    if (NULL == (payload = util_malloc(size + 1 /* for '\0' */))) {
        rv = 0x1;
        goto error;
    }

    memset(payload, 'A', size);
    payload[size] = '\0';

    data_msg.u.data.items_value = data_msg_items;

    data_msg.base = &data_msg_base;
    data_msg.u.data.items_desc = desc;
    data_msg.u.data.base = &data_msg_props;

    if (size > INT_MAX) {
        rv = 0x2;
        LOG_ERRS("Loss of data was found");
        goto error;
    }

    data_msg.u.data.items_value[1].int_value = (int) size;
    data_msg.u.data.items_value[2].string_value = payload;

    for (current_count = 0; current_count < count - 1; current_count++) {
		if (current_count > INT_MAX) {
			rv = 0x2;
			LOG_ERRS("Loss of data was found");
			goto error;
		}
        data_msg.u.data.items_value[0].int_value = (int) current_count;
        if (IOTCS_RESULT_OK != cl_internal_send(&data_msg, 1, NULL, NULL)) {
            rv = 0x2;
            goto error;
        }

        UTIL_LOCK_LL();
        is_stopped = (g_connectivity_test_stop) ? IOTCS_TRUE : IOTCS_FALSE;
        UTIL_UNLOCK_LL();

        GOTO_ERR(is_stopped == IOTCS_TRUE);

        time_to_wait = interval;

        /* Don't sleep too much, periodically wake up and check g_connectivity_test_stop flag. */
        while (time_to_wait > 0) {
            int64_t time_to_sleep = time_to_wait > DM_TEST_CONNECTIVITY_MAX_TIME_TO_SLEEP_MS ?
                    DM_TEST_CONNECTIVITY_MAX_TIME_TO_SLEEP_MS : time_to_wait;

			iotcs_port_sleep_millis(time_to_sleep < INT32_MAX ? (int) time_to_sleep : INT32_MAX);
            time_to_wait -= time_to_sleep;

            UTIL_LOCK_LL();
            is_stopped = (g_connectivity_test_stop) ? IOTCS_TRUE : IOTCS_FALSE;
            UTIL_UNLOCK_LL();

            GOTO_ERR(is_stopped == IOTCS_TRUE);
        }
    }

    /*last send should be made without sleeping*/
	if (current_count > INT_MAX) {
		rv = 0x2;
		LOG_ERRS("Loss of data was found");
		goto error;
	}
	data_msg.u.data.items_value[0].int_value = (int) current_count;
    if (IOTCS_RESULT_OK != cl_internal_send(&data_msg, 1, NULL, NULL)) {
        rv = 0x2;
        goto error;
    }

error:
    util_free(payload);
    iotcs_port_thread_cleanup();

    UTIL_LOCK_LL();
    reset_connectivity_test_fields();
    UTIL_UNLOCK_LL();
    return (void*) (intptr_t) rv;
}

static void destroy_connectivity_test_thread(void) {
    if (g_connectivity_test_thread) {
        iotcs_port_thread test_thread = g_connectivity_test_thread;
        g_connectivity_test_thread = NULL;
        UTIL_UNLOCK_LL();
        iotcs_port_thread_join(test_thread);
        UTIL_LOCK_LL();
    }
}

static iotcs_result recreate_connectivity_test_thread(int64_t count, int64_t interval, size_t size) {
    if (!g_resource_registered) {
        return IOTCS_RESULT_FAIL;
    }
    destroy_connectivity_test_thread();
    g_connectivity_test_stop = 0;
    g_count = count;
    g_interval = interval;
    g_size = size;
    if (NULL == (g_connectivity_test_thread = iotcs_port_thread_create(connectivity_test_thread))) {
        reset_connectivity_test_fields();
        return IOTCS_RESULT_FAIL;
    }
    return IOTCS_RESULT_OK;
}

static PROTOCOL_RESPONSE_CODE start_connectivity_test(int64_t count, int64_t interval, size_t size) {
    UTIL_LOCK_LL();
    if (g_active == IOTCS_FALSE) {
        g_active = IOTCS_TRUE;
        iotcs_result res = recreate_connectivity_test_thread(count, interval, size);
        UTIL_UNLOCK_LL();
        return IOTCS_RESULT_OK != res ? PROTOCOL_RESPONSE_CODE_INTERNAL_SERVER_ERROR : PROTOCOL_RESPONSE_CODE_OK;
    } else {
        UTIL_UNLOCK_LL();
        return PROTOCOL_RESPONSE_CODE_CONFLICT;
    }
}

static void stop_connectivity_test_unsafe(void) {
    g_connectivity_test_stop = 1;
}

static PROTOCOL_RESPONSE_CODE stop_connectivity_test(void) {
    UTIL_LOCK_LL();
    stop_connectivity_test_unsafe();
    UTIL_UNLOCK_LL();
    return PROTOCOL_RESPONSE_CODE_OK;
}

static void connectivity_test_finalize(void) {
    stop_connectivity_test_unsafe();
    destroy_connectivity_test_thread();
}

iotcs_result dgn_start_stop_test_handler(iotcs_request_message *request) {
    int status_code = PROTOCOL_RESPONSE_CODE_OK;
    iotcs_message response = {0};
    iotcs_bool should_be_started = IOTCS_FALSE;
    iotcs_result result = IOTCS_RESULT_FAIL;
    json_tokens_t _tokens;

    UTIL_LOCK_LL(); /* needed to json parser */
    if (request->method != IOTCS_REQUEST_METHOD_PUT) {
        LOG_ERRS("Unexpected HTTP method");
        status_code = PROTOCOL_RESPONSE_CODE_INTERNAL_SERVER_ERROR;
        UTIL_UNLOCK_LL();
    } else if (IOTCS_RESULT_OK != json_parse(request->body, strlen(request->body), &_tokens)) {
        status_code = PROTOCOL_RESPONSE_CODE_INTERNAL_SERVER_ERROR;
        UTIL_UNLOCK_LL();
    } else if (IOTCS_RESULT_OK != json_get_bool_value_by_key(&_tokens, "active", &should_be_started)) {
        status_code = PROTOCOL_RESPONSE_CODE_BAD_REQUEST;
        UTIL_UNLOCK_LL();
    } else {
        if (should_be_started) {
            int64_t count = 0, interval = 0, size = 0;
            if (IOTCS_RESULT_OK != json_get_int_value_by_key(&_tokens, "count", &count) ||
                    IOTCS_RESULT_OK != json_get_int_value_by_key(&_tokens, "interval", &interval) ||
                    IOTCS_RESULT_OK != json_get_int_value_by_key(&_tokens, "size", &size)) {
                UTIL_UNLOCK_LL();
                status_code = PROTOCOL_RESPONSE_CODE_BAD_REQUEST;
            } else {
                UTIL_UNLOCK_LL();
                status_code = start_connectivity_test(count, interval, (size_t) size);
            }
        } else {
            UTIL_UNLOCK_LL();
            status_code = stop_connectivity_test();
        }
    }

    response.base = &response_base;
    response.u.response.status_code = status_code;
    response.u.response.request = request;
    request = NULL;
    response.u.response.body = "{}";

    result = cl_internal_send(&response, 1, NULL, NULL);

    cl_internal_request_message_free(request);
    return result;
}

static int dgn_status_json_callback(json_buf *buf, int pos, void* data) {
    int rv;
    int original_pos = pos;
    int64_t count, interval;
    size_t size;
    iotcs_bool active;
    iotcs_bool is_stopped;
    (void) data;
    UTIL_LOCK_LL();
    count = g_count;
    interval = g_interval;
    size = g_size;
    active = g_active;
    is_stopped = g_connectivity_test_stop;
    UTIL_UNLOCK_LL();

    active = active && !is_stopped ? IOTCS_TRUE : IOTCS_FALSE;

    rv = json_write_bool_field(buf, pos, "active", active);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "count", count);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "interval", interval);
    JSON_MOVE_POS(pos, rv);
    JSON_PUT_SYMBOL(buf, pos, ',');

    rv = json_write_int_field(buf, pos, "size", (int64_t) size);
    JSON_MOVE_POS(pos, rv);

    return pos - original_pos;

error:
    return -1;
}

iotcs_result dgn_status_handler(iotcs_request_message *request) {
    int rv, status_code = PROTOCOL_RESPONSE_CODE_OK;
    json_buf buf_desc = {NULL, 0};
    iotcs_message response = {0};
    iotcs_result result = IOTCS_RESULT_FAIL;

    if (request->method != IOTCS_REQUEST_METHOD_GET) {
        LOG_ERRS("Unexpected HTTP method for diagnostic /testConnectivity/status");
        return IOTCS_RESULT_FAIL;
    }

    buf_desc.buf = (char*) util_malloc(IOTCS_CAPABILITY_RESPONSE_BUFFER_SIZE);
    GOTO_ERR_CRIT(!buf_desc.buf);
    buf_desc.len = IOTCS_CAPABILITY_RESPONSE_BUFFER_SIZE;
    rv = json_write_object_null_terminated(&buf_desc, 0, dgn_status_json_callback, NULL);
    result = IOTCS_RESULT_OUT_OF_MEMORY;
    GOTO_ERR(rv < 0);

    response.base = &response_base;
    response.u.response.status_code = status_code;
    response.u.response.request = request;
    request = NULL;
    response.u.response.body = buf_desc.buf;

    result = cl_internal_send(&response, 1, NULL, NULL);

error:
    util_free(buf_desc.buf);
    cl_internal_request_message_free(request);
    return result;
}
#endif
