/*
 * 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 <pthread.h>
#include <errno.h>
#include <sys/time.h>
#include "util/util_array_queue.h"
#include "iotcs_port_queue.h"
#include "iotcs_port_time.h"

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

#define THREAD_WAIT_INFINITY -1    
#define THREAD_WAIT_ERROR -2

typedef struct {
    util_array_queue queue;
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    pthread_mutex_t mutex;
    pthread_cond_t cond_var;
#endif
} blocking_queue;

static void blocking_queue_finit(blocking_queue *queue);

int blocking_queue_init(blocking_queue *queue, int32_t items_count, int32_t item_size, int8_t *buffer) {
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    int i;

    if (0 != (i = pthread_mutex_init(&queue->mutex, NULL))) {
        blocking_queue_finit(queue);
        LOG_CRITS("blocking_queue_create - OOM error");
        return -1;
    }

    if (0 != (pthread_cond_init(&queue->cond_var, NULL))) {
        blocking_queue_finit(queue);
        LOG_CRITS("blocking_queue_create - OOM error");
        return -1;
    }
#endif

    util_array_queue_init(&queue->queue, items_count, item_size, buffer);

    return 0;
}

static void blocking_queue_finit(blocking_queue *queue) {

#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    pthread_mutex_destroy(&queue->mutex);
    pthread_cond_destroy(&queue->cond_var);
#endif
}

#ifdef IOTCS_MESSAGING_THREAD_SAFETY

/*
 * returns: 0 - timeout passed, THREAD_WAIT_INFINITY - timeout wasn't set,
 * THREAD_WAIT_ERROR - if errors occur, remaining timeout otherwise
 */
static int32_t cond_var_wait(pthread_mutex_t *mutex, pthread_cond_t *cond_var, int32_t timeout_ms) {
    struct timespec abstimeout;

    if (timeout_ms == 0)
        return 0;

    if (timeout_ms > 0) {
        struct timespec timeout;
        IOTCS_MILLISEC_TO_TIMESPEC(timeout_ms, &timeout);

        struct timeval now;
        gettimeofday(&now, NULL);
        abstimeout.tv_sec = now.tv_sec + timeout.tv_sec;
        abstimeout.tv_nsec = IOTCS_MICROSEC_TO_NANOSEC(now.tv_usec) + timeout.tv_nsec;
        if (abstimeout.tv_nsec >= IOTCS_NANOSEC_IN_SEC) {
            abstimeout.tv_sec++;
            abstimeout.tv_nsec -= IOTCS_NANOSEC_IN_SEC;
        }

        int ret = pthread_cond_timedwait(cond_var, mutex, &abstimeout);

        if (ret == ETIMEDOUT) {
            return 0;
        }

        if (ret != 0) {
            LOG_ERRS("cond_var_wait failed - pthread_cond_timedwait");
            return THREAD_WAIT_ERROR;
        }

        gettimeofday(&now, NULL);

        abstimeout.tv_sec -= now.tv_sec;
        abstimeout.tv_nsec -= IOTCS_MICROSEC_TO_NANOSEC(now.tv_usec);

        int32_t time_left = IOTCS_TIMESPEC_TO_MILLISEC(&abstimeout);
        return (time_left > 0) ? time_left : 0;
    } else {
        if (pthread_cond_wait(cond_var, mutex) != 0) {
            LOG_ERRS("cond_var_wait failed - pthread_cond_wait");
            return THREAD_WAIT_ERROR;
        }

        return THREAD_WAIT_INFINITY;
    }
}
#endif

static iotcs_result blocking_queue_put(blocking_queue *queue, void* item, int32_t timeout_ms) {
    iotcs_result rv;

#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    pthread_mutex_lock(&queue->mutex);

    while (IOTCS_RESULT_FAIL == (rv = util_array_queue_push(&queue->queue, item))) {
        if (timeout_ms == 0) {
            rv = IOTCS_RESULT_FAIL;
            break;
        }

        if (THREAD_WAIT_ERROR == (timeout_ms = cond_var_wait(&queue->mutex,
                &queue->cond_var, timeout_ms))) {
            LOG_ERRS("blocking_queue_put: cond_var_wait method failed");
            pthread_mutex_unlock(&queue->mutex);
            return IOTCS_RESULT_FAIL;
        }
    }

    if (rv == IOTCS_RESULT_OK) {
        pthread_cond_signal(&queue->cond_var);
    }

    pthread_mutex_unlock(&queue->mutex);
#else
    rv = util_array_queue_push(&queue->queue, item);
#endif

    return rv;
}

static iotcs_result blocking_queue_get(blocking_queue *queue, void* item, int32_t timeout_ms) {
    iotcs_result rv;

#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    pthread_mutex_lock(&queue->mutex);

    if (timeout_ms == 0) {
        if (IOTCS_RESULT_OK == (rv = util_array_queue_pop(&queue->queue, item))) {
            pthread_cond_signal(&queue->cond_var);
        }
        pthread_mutex_unlock(&queue->mutex);
        return rv;
    }

    while (IOTCS_RESULT_OK != (rv = util_array_queue_pop(&queue->queue, item))) {
        if (timeout_ms == 0) {
            break;
        }
        if (THREAD_WAIT_ERROR == (timeout_ms = cond_var_wait(&queue->mutex,
                &queue->cond_var, timeout_ms))) {
            LOG_ERRS("blocking_queue_get: cond_var_wait method failed");
            pthread_mutex_unlock(&queue->mutex);
            return IOTCS_RESULT_FAIL;
        }
    }

    if (rv == IOTCS_RESULT_OK) {
        pthread_cond_signal(&queue->cond_var);
    }

    pthread_mutex_unlock(&queue->mutex);
#else
    rv = util_array_queue_pop(&queue->queue, item);
#endif
    return rv;
}

static int blocking_queue_count(blocking_queue *queue) {
    int rv;

#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    pthread_mutex_lock(&queue->mutex);
#endif
    rv = UTIL_ARRAY_QUEUE_COUNT(&queue->queue);
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
    pthread_mutex_unlock(&queue->mutex);
#endif

    return rv;
}

static blocking_queue message_queue;
static int8_t message_queue_buf[IOTCS_PORT_MESSAGE_QUEUE_SIZE * sizeof (iotcs_message)];

iotcs_port_message_queue iotcs_port_message_queue_create() {
    blocking_queue_init(&message_queue, IOTCS_PORT_MESSAGE_QUEUE_SIZE, sizeof (iotcs_message), message_queue_buf);
    return (iotcs_port_message_queue)&message_queue;
}

void iotcs_port_message_queue_destroy(iotcs_port_message_queue queue) {
    blocking_queue_finit((blocking_queue*) queue);
}

iotcs_result iotcs_port_message_queue_put(iotcs_port_message_queue queue, iotcs_message *message, int32_t timeout_ms) {
    return blocking_queue_put((blocking_queue*) queue, (void*) message, timeout_ms);
}

iotcs_result iotcs_port_message_queue_get(iotcs_port_message_queue queue, iotcs_message *message /* OUT */, int32_t timeout_ms) {
    return blocking_queue_get((blocking_queue*) queue, (void*) message, timeout_ms);
}

int iotcs_port_message_queue_count(iotcs_port_message_queue queue) {
    return blocking_queue_count((blocking_queue*) queue);
}

int iotcs_port_request_queue_count(iotcs_port_request_queue queue) {
    return blocking_queue_count((blocking_queue*) queue);
}

static blocking_queue request_queue;
static int8_t request_queue_buf[IOTCS_PORT_REQUEST_QUEUE_SIZE * sizeof (iotcs_request_message*)];

iotcs_port_request_queue iotcs_port_request_queue_create() {
    blocking_queue_init(&request_queue, IOTCS_PORT_REQUEST_QUEUE_SIZE, sizeof (iotcs_request_message*), request_queue_buf);
    return (iotcs_port_message_queue)&request_queue;
}

void iotcs_port_request_queue_destroy(iotcs_port_request_queue queue) {
    blocking_queue_finit((blocking_queue*) queue);
}

iotcs_result iotcs_port_request_queue_put(iotcs_port_request_queue queue, iotcs_request_message *request, int32_t timeout_ms) {
    return blocking_queue_put((blocking_queue*) queue, (void*) &request, timeout_ms);
}

iotcs_result iotcs_port_request_queue_get(iotcs_port_request_queue queue, iotcs_request_message **request /* OUT */, int32_t timeout_ms) {
    return blocking_queue_get((blocking_queue*) queue, (void*) request, timeout_ms);
}

#ifdef IOTCS_STORAGE_SUPPORT
static blocking_queue scs_request_queue;
static int8_t scs_request_queue_buf[IOTCS_PORT_MEDIA_STORAGE_REQUEST_QUEUE_SIZE * sizeof (iotcs_storage_request)];

iotcs_port_storage_request_queue iotcs_port_storage_request_queue_create() {
    blocking_queue_init(&scs_request_queue, IOTCS_PORT_MEDIA_STORAGE_REQUEST_QUEUE_SIZE, sizeof (iotcs_storage_request), scs_request_queue_buf);
    return (iotcs_port_storage_request_queue)&scs_request_queue;
}

void iotcs_port_storage_request_queue_destroy(iotcs_port_storage_request_queue queue) {
    blocking_queue_finit((blocking_queue*) queue);
}

iotcs_result iotcs_port_storage_request_queue_put(iotcs_port_storage_request_queue queue, iotcs_storage_request *request, int32_t timeout_ms) {
    return blocking_queue_put((blocking_queue*) queue, (void*) request, timeout_ms);
}

iotcs_result iotcs_port_storage_request_queue_get(iotcs_port_storage_request_queue queue, iotcs_storage_request *request /* OUT */, int32_t timeout_ms) {
    return blocking_queue_get((blocking_queue*) queue, (void*) request, timeout_ms);
}
#endif

#if  defined(IOTCS_MESSAGE_DISPATCHER) && (IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE > 1)
static blocking_queue request_handler_queue;
static int8_t request_handler_queue_buf[IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE *sizeof(iotcs_request_handler_message)];

iotcs_port_request_handler_queue iotcs_port_request_handler_queue_create() {
	blocking_queue_init(&request_handler_queue, IOTCS_REQUEST_HANDLER_THREAD_POOL_SIZE, sizeof(iotcs_request_handler_message), request_handler_queue_buf);
	return (iotcs_port_message_queue)&request_handler_queue;
}

void iotcs_port_request_handler_queue_destroy(iotcs_port_request_handler_queue queue) {
	blocking_queue_finit((blocking_queue*)queue);
}

iotcs_result iotcs_port_request_handler_queue_put(iotcs_port_request_handler_queue queue, iotcs_request_handler_message *request, int32_t timeout_ms) {
	return blocking_queue_put((blocking_queue*)queue, (void*)request, timeout_ms);
}

iotcs_result iotcs_port_request_handler_queue_get(iotcs_port_request_handler_queue queue, iotcs_request_handler_message *request /* OUT */, int32_t timeout_ms) {
	return blocking_queue_get((blocking_queue*)queue, (void*)request, timeout_ms);
}

int iotcs_port_request_handler_queue_count(iotcs_port_request_handler_queue queue) {
    return blocking_queue_count((blocking_queue*) queue);
}
#endif
