/*
 * 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 <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include "advanced/iotcs_storage_object.h"
#include "scs/storage_dispatcher_private.h"
#include "iotcs_port_queue.h"
#include "iotcs/iotcs_private.h"
#include "util/util_thread_private.h"
#include "util/util_memory.h"
#include "util/util.h"
#include "iotcs_port_thread.h"

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

#define STOP_STORAGE_OBJECT 1
#define AUTH_SLEEP_INTERVAL_MS 5000

static iotcs_port_storage_request_queue g_storage_queue;
static iotcs_port_thread g_storage_thread;

void* process_thread(void *arg) {
    (void) arg;
    int stop_request_received = 0;
    iotcs_storage_request *node = (iotcs_storage_request *) util_malloc(sizeof(iotcs_storage_request));
    iotcs_result result;
    struct sd_storage_object_t *storage_object = NULL;

    if (node != NULL) {
        while (!stop_request_received) {
            result = iotcs_port_storage_request_queue_get(g_storage_queue, node, -1);

            if (result == IOTCS_RESULT_OK) {
                if (node->storage == (iotcs_storage_object_handle) STOP_STORAGE_OBJECT) {
                    stop_request_received = 1;
                    break;
                }

                storage_object = (struct sd_storage_object_t *)node->storage->object;
                if (SCS_IS_SYNCING(storage_object->progress_state)) {
                    if (storage_object->input_path) {
                        result = cl_internal_storage_file_upload(node);
                    } else {
                        result = cl_internal_storage_file_download(node);
                    }
                } else {
                    cl_storage_progress_cb_h(node->storage, storage_object->progress_state, -1, NULL);
                }
                dm_free_uri_object(node->storage);
            }
        }
        util_free(node);
    }
    return NULL;
}

iotcs_result cl_internal_storage_async_dispatcher_init() {
    if (NULL == (g_storage_queue = iotcs_port_storage_request_queue_create())) {
        return IOTCS_RESULT_FAIL;
    }

    if (NULL == (g_storage_thread = iotcs_port_thread_create(process_thread))) {
        iotcs_port_storage_request_queue_destroy(g_storage_queue);
        return IOTCS_RESULT_FAIL;
    }

    return IOTCS_RESULT_OK;
}

iotcs_result cl_internal_storage_dispatcher_queue_safe(iotcs_storage_request* request) {
    iotcs_result result;
    
    UTIL_LOCK_STORAGE();
    result = iotcs_port_storage_request_queue_put(g_storage_queue, request, -1);
    UTIL_UNLOCK_STORAGE();

    return result;
}

void cl_internal_storage_async_dispatcher_finalize(void) {
    iotcs_storage_request stop = {(iotcs_storage_object_handle) STOP_STORAGE_OBJECT};
    iotcs_port_storage_request_queue_put(g_storage_queue, &stop, -1);
    iotcs_port_thread_join(g_storage_thread);
    iotcs_port_storage_request_queue_destroy(g_storage_queue);
}
