/*
 * 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 <sys/time.h>
#include "util/util_array_queue.h"
#include "iotcs_port_queue.h"
#include "mbed.h"
#include "rtos.h"
#include "Mail.h"

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

uint32_t convert_to_mbed_timeout(int32_t timeout_ms) {
    if (timeout_ms < 0) {
        return osWaitForever;
    }
    return (uint32_t)timeout_ms;
}

static void inc_atomic(volatile int *val) {
    __disable_irq();
    ++(*val);
    __enable_irq();
}

static void dec_atomic(volatile int *val) {
    __disable_irq();
    --(*val);
    __enable_irq();
}

template<typename T, uint32_t queue_sz>
static iotcs_result push(Mail<T, queue_sz>* queue, T *message, int32_t timeout_ms) {
    iotcs_result result = IOTCS_RESULT_FAIL;
    osStatus os_result;
#ifndef IOTCS_MESSAGING_THREAD_SAFETY
    timeout_ms = 1;
#endif // IOTCS_MESSAGING_THREAD_SAFETY

    T *message_p = queue->alloc(convert_to_mbed_timeout(timeout_ms));
    if (message_p != NULL) {
        memcpy(message_p, message, sizeof(T));
        os_result = queue->put(message_p);
        if (os_result == osOK) {
            result = IOTCS_RESULT_OK;
        } else if (os_result == osErrorNoMemory) {
            LOG_CRITS("System is out of memory: it was impossible to allocate or reserve memory for the operation");
        }
    }
    return result;
}

template<typename T, uint32_t queue_sz>
static iotcs_result pop(Mail<T, queue_sz>* queue, T *message, int32_t timeout_ms) {
    iotcs_result result = IOTCS_RESULT_FAIL;
    osEvent os_evt;
#ifndef IOTCS_MESSAGING_THREAD_SAFETY
    timeout_ms = 1;
#endif // IOTCS_MESSAGING_THREAD_SAFETY

    os_evt = queue->get(convert_to_mbed_timeout(timeout_ms));
    if (os_evt.status == osEventMail) {
        memcpy(message, os_evt.value.p, sizeof(T));
        queue->free((T*)os_evt.value.p);
        result = IOTCS_RESULT_OK;
    }
    return result;
}

static Mail<iotcs_message, IOTCS_PORT_MESSAGE_QUEUE_SIZE> message_queue;
static volatile int message_queue_cnt;
iotcs_port_message_queue iotcs_port_message_queue_create() {
    message_queue_cnt = 0;
    return (iotcs_port_message_queue)&message_queue;
}

void iotcs_port_message_queue_destroy(iotcs_port_message_queue queue) {
    if (message_queue_cnt) {
        iotcs_message message;
        while (pop<iotcs_message, IOTCS_PORT_MESSAGE_QUEUE_SIZE>
                    ((Mail<iotcs_message, IOTCS_PORT_MESSAGE_QUEUE_SIZE>*)queue,
                     &message, 0) == IOTCS_RESULT_OK) {
            dec_atomic(&message_queue_cnt);
        }
    }
}

iotcs_result iotcs_port_message_queue_put(iotcs_port_message_queue queue,
                                    iotcs_message *message,
                                    int32_t timeout_ms) {
    if (push<iotcs_message, IOTCS_PORT_MESSAGE_QUEUE_SIZE>
                ((Mail<iotcs_message, IOTCS_PORT_MESSAGE_QUEUE_SIZE>*)queue,
                 message, timeout_ms) == IOTCS_RESULT_OK) {
        inc_atomic(&message_queue_cnt);
        return IOTCS_RESULT_OK;
    }
    return IOTCS_RESULT_FAIL;
}

iotcs_result iotcs_port_message_queue_get(iotcs_port_message_queue queue,
                                    iotcs_message *message /* OUT */,
                                    int32_t timeout_ms) {
    if (pop<iotcs_message, IOTCS_PORT_MESSAGE_QUEUE_SIZE>
                ((Mail<iotcs_message, IOTCS_PORT_MESSAGE_QUEUE_SIZE>*)queue,
                 message, timeout_ms) == IOTCS_RESULT_OK) {
        dec_atomic(&message_queue_cnt);
        return IOTCS_RESULT_OK;
    }
    return IOTCS_RESULT_FAIL;
}

int iotcs_port_message_queue_count(iotcs_port_message_queue queue) {
    (void) queue;
    return (int)message_queue_cnt;
}

static Mail<iotcs_request_message*, IOTCS_PORT_REQUEST_QUEUE_SIZE> request_queue;
static volatile int request_queue_cnt;

iotcs_port_request_queue iotcs_port_request_queue_create() {
    request_queue_cnt = 0;
    return (iotcs_port_request_queue)&request_queue;
}

void iotcs_port_request_queue_destroy(iotcs_port_request_queue queue) {
    if (request_queue_cnt) {
        iotcs_request_message *request;
        while (pop<iotcs_request_message*, IOTCS_PORT_REQUEST_QUEUE_SIZE>
                    ((Mail<iotcs_request_message*, IOTCS_PORT_REQUEST_QUEUE_SIZE>*)queue,
                     &request, 0) == IOTCS_RESULT_OK) {
            dec_atomic(&request_queue_cnt);
        }
    }
}

iotcs_result iotcs_port_request_queue_put(iotcs_port_request_queue queue,
                                    iotcs_request_message *request,
                                    int32_t timeout_ms) {
    if (push<iotcs_request_message*, IOTCS_PORT_REQUEST_QUEUE_SIZE>
                ((Mail<iotcs_request_message*, IOTCS_PORT_REQUEST_QUEUE_SIZE>*)queue,
                 &request, timeout_ms) == IOTCS_RESULT_OK) {
        inc_atomic(&request_queue_cnt);
        return IOTCS_RESULT_OK;
    }
    return IOTCS_RESULT_FAIL;
}

iotcs_result iotcs_port_request_queue_get(iotcs_port_request_queue queue,
                                    iotcs_request_message **request /* OUT */,
                                    int32_t timeout_ms) {
    if (pop<iotcs_request_message*, IOTCS_PORT_REQUEST_QUEUE_SIZE>
                ((Mail<iotcs_request_message*, IOTCS_PORT_REQUEST_QUEUE_SIZE>*)queue,
                 request, timeout_ms) == IOTCS_RESULT_OK) {
        dec_atomic(&request_queue_cnt);
        return IOTCS_RESULT_OK;
    }
    return IOTCS_RESULT_FAIL;
}

int iotcs_port_request_queue_count(iotcs_port_request_queue queue) {
    (void) queue;
    return (int) request_queue_cnt;
}

#ifdef IOTCS_STORAGE_SUPPORT
static Mail<iotcs_storage_request*, IOTCS_PORT_MEDIA_STORAGE_REQUEST_QUEUE_SIZE> scs_queue;
static volatile int scs_queue_cnt;

iotcs_port_storage_request_queue iotcs_port_storage_request_queue_create() {
    scs_queue_cnt = 0;
    return (iotcs_port_storage_request_queue)&scs_queue;
}

void iotcs_port_storage_request_queue_destroy(iotcs_port_storage_request_queue queue) {
    if (scs_queue_cnt) {
        iotcs_storage_request* scs;
        while (pop<iotcs_storage_request, IOTCS_PORT_REQUEST_QUEUE_SIZE>
                    ((Mail<iotcs_storage_request, IOTCS_PORT_REQUEST_QUEUE_SIZE>*)queue,
                     scs, 0) == IOTCS_RESULT_OK) {
            dec_atomic(&scs_queue_cnt);
        }
    }
}

iotcs_result iotcs_port_storage_request_queue_put(iotcs_port_storage_request_queue queue,
                                    iotcs_storage_request *scs,
                                    int32_t timeout_ms) {
    if (push<iotcs_storage_request, IOTCS_PORT_REQUEST_QUEUE_SIZE>
                ((Mail<iotcs_storage_request, IOTCS_PORT_REQUEST_QUEUE_SIZE>*)queue,
                 scs, timeout_ms) == IOTCS_RESULT_OK) {
        inc_atomic(&scs_queue_cnt);
        return IOTCS_RESULT_OK;
    }
    return IOTCS_RESULT_FAIL;
}

iotcs_result iotcs_port_storage_request_queue_get(iotcs_port_storage_request_queue queue,
                                    iotcs_storage_request *scs /* OUT */,
                                    int32_t timeout_ms) {
    if (pop<iotcs_storage_request, IOTCS_PORT_REQUEST_QUEUE_SIZE>
                ((Mail<iotcs_storage_request, IOTCS_PORT_REQUEST_QUEUE_SIZE>*)queue,
                 scs, timeout_ms) == IOTCS_RESULT_OK) {
        dec_atomic(&scs_queue_cnt);
        return IOTCS_RESULT_OK;
    }
    return IOTCS_RESULT_FAIL;
}
#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_PORT_REQUEST_HANDLER_QUEUE_SIZE *sizeof(iotcs_request_handler_message)];

iotcs_port_request_handler_queue iotcs_port_request_handler_queue_create() {
	blocking_queue_init(&request_handler_queue, IOTCS_PORT_REQUEST_HANDLER_QUEUE_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
