/*
* 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 "util/util_array_queue.h"
#include "iotcs_port_queue.h"

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

#include <windows.h>
#include <sys/timeb.h>

#define THREAD_WAIT_INFINITY -1    
#define THREAD_WAIT_ERROR -2

typedef struct {
	util_array_queue queue;
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
	CRITICAL_SECTION   mutex;
	CONDITION_VARIABLE cond_var;
#endif
} blocking_queue;

int blocking_queue_init(blocking_queue *queue, int32_t items_count, int32_t item_size, int8_t *buffer) {
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
	InitializeCriticalSection(&queue->mutex);
	InitializeConditionVariable(&queue->cond_var);
#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
	DeleteCriticalSection(&queue->mutex);
#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(CRITICAL_SECTION* mutex, CONDITION_VARIABLE* cond_var, int32_t timeout_ms) {
	if (timeout_ms == 0)
		return 0;

	if (timeout_ms > 0) {
		struct timeb start, end;
		ftime(&start);

		if (!SleepConditionVariableCS(cond_var, mutex, timeout_ms)) {
			if (GetLastError() == ERROR_TIMEOUT) {
				return 0;
			}
			else {
				LOG_ERRS("cond_var_wait failed - SleepConditionVariableCS");
				return THREAD_WAIT_ERROR;
			}
		}

		ftime(&end);
		int32_t time_left = (int)(1000.0 * (end.time - start.time) + (end.millitm - start.millitm));
		return (time_left >= timeout_ms) ? 0 : time_left - timeout_ms;
	}
	else {
		if (!SleepConditionVariableCS(cond_var, mutex, INFINITE)) {
			LOG_ERRS("cond_var_wait failed - SleepConditionVariableCS INFINITE");
			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
	EnterCriticalSection(&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");
			LeaveCriticalSection(&queue->mutex);
			return IOTCS_RESULT_FAIL;
		}
	}

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

	LeaveCriticalSection(&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
	EnterCriticalSection(&queue->mutex);

	if (timeout_ms == 0){
		if (IOTCS_RESULT_OK == (rv = util_array_queue_pop(&queue->queue, item))) {
			WakeConditionVariable(&queue->cond_var);
		}
		LeaveCriticalSection(&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");
			LeaveCriticalSection(&queue->mutex);
			return IOTCS_RESULT_FAIL;
		}
	}

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

	LeaveCriticalSection(&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
	EnterCriticalSection(&queue->mutex);
#endif
	rv = UTIL_ARRAY_QUEUE_COUNT(&queue->queue);
#ifdef IOTCS_MESSAGING_THREAD_SAFETY
	LeaveCriticalSection(&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);
}

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);
}

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

#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
