/*
 * 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_messaging.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"

#if  IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE > 1

static iotcs_port_request_handler_queue g_request_handler_dispatch_queue;
static iotcs_port_thread g_request_handler_dispatch_threads[IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE];

static int  stop_request_received = 0;

static iotcs_message_base default_base = {
    .priority = IOTCS_MESSAGE_PRIORITY_HIGH,
    .reliability = IOTCS_MESSAGE_RELIABILITY_BEST_EFFORT,
    .type = IOTCS_MESSAGE_RESPONSE
};

void* request_handler_dispatch_thread(void *arg) {
    (void) arg;
    iotcs_request_handler_message *message = (iotcs_request_handler_message *) util_malloc(sizeof(iotcs_request_handler_message));
    iotcs_result result;
    iotcs_message response_message;

    if (message != NULL) {
        while (!stop_request_received) {
            result = iotcs_port_request_handler_queue_get(g_request_handler_dispatch_queue, message, -1);
            if (result == IOTCS_RESULT_OK) {
                if (message->handler != NULL && message->request_message != NULL) {
                    // initialize the response message.
                    memset(&response_message, 0, sizeof (iotcs_message));
                    response_message.base = &default_base;
                    // call the request handler with the request message received.
                    message->handler(message->request_message, &response_message);
                    // associate the response with request message.
                    response_message.u.response.request = message->request_message;
                    // call this handler for sending the response message.
                    cl_internal_async_message_dispatcher_process_response_message(message->request_message, &response_message);
                }
            }
        }
        util_free(message);
    }
    return NULL;
}

iotcs_result cl_internal_request_handler_thread_pool_init() {
    if (NULL == (g_request_handler_dispatch_queue = iotcs_port_request_handler_queue_create())) {
        return IOTCS_RESULT_FAIL;
    }

    for (int i=0; i< IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE ; i++) {
        if (NULL == (g_request_handler_dispatch_threads[i] = iotcs_port_thread_create(request_handler_dispatch_thread))) {
            iotcs_port_request_handler_queue_destroy(g_request_handler_dispatch_queue);
            return IOTCS_RESULT_FAIL;
        }
    }

    return IOTCS_RESULT_OK;
}

iotcs_result cl_internal_request_handler_pool_dispatcher_dispatch(iotcs_request_handler handler, iotcs_request_message* request_message) {
    iotcs_result result;
    
    iotcs_request_handler_message handler_message;
    handler_message.handler = handler;
    handler_message.request_message = request_message;

    result = iotcs_port_request_handler_queue_put(g_request_handler_dispatch_queue, &handler_message, -1);

    return result;
}

void cl_internal_request_handler_dispatcher_finalize(void) {
    stop_request_received = 1;
    // add empty request handler messages to stop threads in thread pool.
    for (int i=0; i< IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE ; i++) {
        cl_internal_request_handler_pool_dispatcher_dispatch(NULL, NULL);
    }
    // wait for thread pool threads to exit.
    for (int i=0; i< IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE ; i++) {
        iotcs_port_thread_join(g_request_handler_dispatch_threads[i]);
    }
    //destory the request handler queue.
    iotcs_port_storage_request_queue_destroy(g_request_handler_dispatch_queue);
}


#endif
