#include <stdlib.h>
#include <string.h>

#include "dataserver_utils.h"
#include "dataserver_rb.h"

typedef struct {
    struct list_head head;

    size_t bs;
} ds_rb_bs_t;

static int _ds_rb_init(ds_rb_t *rb)
{
    if (rb->lock_g_ptr)
        rb->lock_g_ptr(rb->g, 1);

    if (!rb->initialized) {
        {
            pthread_condattr_t cattr;
            pthread_condattr_init(&cattr);
            pthread_condattr_setclock(&cattr, CLOCK_MONOTONIC);
            pthread_cond_init(&rb->cond, &cattr);
            pthread_condattr_destroy(&cattr);
        }
        pthread_mutex_init(&rb->lock, NULL);

        pthread_mutex_lock(&rb->lock);
        rb->initialized = 1;
        rb->flag = 0;
        pthread_mutex_unlock(&rb->lock);
    }

    if (rb->lock_g_ptr)
        rb->lock_g_ptr(rb->g, 0);

    return 0;
}

int ds_rb_init(ds_rb_t *rb, size_t size)
{
    _ds_rb_init(rb);

    pthread_mutex_lock(&rb->lock);

    rb->pread = rb->pwrite = 0;
    rb->data = (uint8_t*)malloc(size);
    if (!rb->data) {
        pthread_mutex_unlock(&rb->lock);
        return -1;
    }

    rb->size = size;
    INIT_LIST_HEAD(&rb->queue);
    pthread_mutex_init(&rb->queue_lock, NULL);

    pthread_mutex_unlock(&rb->lock);

    return 0;
}

int ds_rb_deinit(ds_rb_t *rb)
{
    pthread_mutex_lock(&rb->lock);

    if (rb->data) {

        free(rb->data);
        rb->data = NULL;

        rb->pread = rb->pwrite = 0;
        rb->size = 0;
        rb->flag = 0;
        //pthread_cond_destroy(&rb->cond);

        {
            ds_rb_bs_t *rbs, *tmp;
            pthread_mutex_lock(&rb->queue_lock);
            list_for_each_entry_safe(rbs, tmp, &rb->queue, head) {
                list_del(&rbs->head);
                free(rbs);
            }
            pthread_mutex_unlock(&rb->queue_lock);
            //pthread_mutex_destroy(&rb->queue_lock);
        }
    }

    pthread_mutex_unlock(&rb->lock);
    //pthread_mutex_destroy(&rb->lock);
    return 0;
}

int ds_rb_empty(ds_rb_t *rb)
{
    int empty;
    pthread_mutex_lock(&rb->lock);
    empty = (rb->pread == rb->pwrite)? 1 : 0;
    pthread_mutex_unlock(&rb->lock);
    return empty;
}

ssize_t ds_rb_free(ds_rb_t *rb)
{
    ssize_t free;
    pthread_mutex_lock(&rb->lock);
    free = rb->pread - rb->pwrite;
    if (free <= 0)
        free += rb->size;
    pthread_mutex_unlock(&rb->lock);
    return free - 1;
}

ssize_t ds_rb_avail(ds_rb_t *rb)
{
    ssize_t avail;
    pthread_mutex_lock(&rb->lock);
    avail = rb->pwrite - rb->pread;
    if (avail < 0)
        avail += rb->size;
    pthread_mutex_unlock(&rb->lock);
    return avail;
}

ssize_t ds_rb_read(ds_rb_t *rb, uint8_t *buf, size_t len)
{
    size_t todo = len;
    size_t split;

    pthread_mutex_lock(&rb->lock);
    if (!rb->data) {
        pthread_mutex_unlock(&rb->lock);
        return 0;
    }

    split = ((ssize_t)(rb->pread + len) > rb->size) ? rb->size - rb->pread : 0;
    if (split > 0) {
        memcpy(buf, rb->data + rb->pread, split);
        buf += split;
        todo -= split;
        rb->pread = 0;
    }

    memcpy(buf, rb->data + rb->pread, todo);

    rb->pread = (rb->pread + todo) % rb->size;

    pthread_mutex_unlock(&rb->lock);

    return len;
}

ssize_t ds_rb_write(ds_rb_t *rb, const uint8_t *buf, size_t len)
{
    size_t todo = len;
    size_t split;

    pthread_mutex_lock(&rb->lock);
    if (!rb->data) {
        pthread_mutex_unlock(&rb->lock);
        return 0;
    }

    split = ((ssize_t)(rb->pwrite + len) > rb->size) ? rb->size - rb->pwrite : 0;

    if (split > 0) {
        memcpy(rb->data + rb->pwrite, buf, split);
        buf += split;
        todo -= split;
        rb->pwrite = 0;
    }

    memcpy(rb->data + rb->pwrite, buf, todo);

    rb->pwrite = (rb->pwrite + todo) % rb->size;

    //if (len > 0)
    //    pthread_cond_signal(&rb->cond);

    pthread_mutex_unlock(&rb->lock);
    return len;
}

int ds_rb_set_flag(ds_rb_t *rb, int flag)
{
    _ds_rb_init(rb);
    pthread_mutex_lock(&rb->lock);
    rb->flag |= flag;
    pthread_cond_signal(&rb->cond);
    pthread_mutex_unlock(&rb->lock);
    return 0;
}

int ds_rb_get_flag_timeout(ds_rb_t *rb, int timeout_ms)
{
    int flag;
    struct timespec ts;

    _ds_rb_init(rb);

    pthread_mutex_lock(&rb->lock);

    if (timeout_ms) {
        ds_get_timespec_timeout(10, &ts);
        pthread_cond_timedwait(&rb->cond, &rb->lock, &ts);
    }

    flag = rb->flag;

    pthread_mutex_unlock(&rb->lock);
    return flag;
}

int ds_rb_queue_block(ds_rb_t *rb, size_t bs)
{
    ds_rb_bs_t *rbs = (ds_rb_bs_t *)malloc(sizeof(ds_rb_bs_t));
    if (!rbs)
        return -1;

    rbs->bs = bs;

    pthread_mutex_lock(&rb->queue_lock);
    list_add_tail(&rbs->head, &rb->queue);
    pthread_mutex_unlock(&rb->queue_lock);

    return 0;
}

int ds_rb_dequeue_block(ds_rb_t *rb, size_t *bs)
{
    int ret = 0;
    pthread_mutex_lock(&rb->queue_lock);
    if (list_empty(&rb->queue)) {
        *bs = 0;
        ret = -1;
    } else {
        ds_rb_bs_t *rbs = list_first_entry(&rb->queue, ds_rb_bs_t, head);
        *bs = rbs->bs;
        list_del(&rbs->head);
        free(rbs);
    }
    pthread_mutex_unlock(&rb->queue_lock);
    return ret;
}

int ds_rb_init_lock(ds_rb_t *rb, ds_rb_lock_ptr_t lock_g_ptr, void *g)
{
    rb->lock_g_ptr = lock_g_ptr;
    rb->g = g;
    return 0;
}