
#include <string.h>
#include "ice_streamer.h"
#include "frost_queue.h"
#include "frost_mem.h"
#include "frost/frost_condvar.h"
#include "frost/frost_mutex.h"

/**
 * The usable size of the ringbuffer.
 */
frost_uint
ice_streamer_ringbuf_size (ice_streamer_ringbuf_t *ringbuf)
{
    frost_return_val_if_fail (ringbuf, 0);

    return ringbuf->buffer_size_usable;
}

/**
 * Allocate a new ringbuffer
 *
 * @param size The total size of the new ringbuffer
 * @returns a new #ice_streamer_ringbuf_t
 */
ice_streamer_ringbuf_t *
ice_streamer_ringbuf_new (frost_uint size)
{
    ice_streamer_ringbuf_t *ringbuf = frost_new0 (ice_streamer_ringbuf_t, 1);

    frost_return_val_if_fail (size > 0, NULL);
    frost_return_val_if_fail (size < FROST_MAXUINT, NULL);

    /* we need to allocate one byte more than requested, cause the
     * final byte cannot be used.
     * if we used it, it might lead to the situation where
     * read_index == write_index, which is used for the "empty"
     * condition.
     */
    ringbuf->buffer_size_usable = size;
    ringbuf->buffer_size = size + 1;
    ringbuf->buffer = (frost_uint8 *)frost_malloc (ringbuf->buffer_size);

    ringbuf->free_cond = FROST_CondVarCreate();
    ringbuf->used_cond = FROST_CondVarCreate ();
    ringbuf->eos_cond = FROST_CondVarCreate ();

    ringbuf->hotspots = frost_queue_new ();

    return ringbuf;
}

/**
 * Free all memory used by the ringbuffer
 */
void
ice_streamer_ringbuf_destroy (ice_streamer_ringbuf_t *ringbuf)
{
    frost_return_if_fail (ringbuf);

    FROST_CondVarDestroy(ringbuf->eos_cond);
    FROST_CondVarDestroy (ringbuf->used_cond);
    FROST_CondVarDestroy (ringbuf->free_cond);

    frost_queue_free (ringbuf->hotspots);
    frost_free (ringbuf->buffer);
    frost_free (ringbuf);
}

/**
 * Clear the ringbuffers data
 */
void
ice_streamer_ringbuf_clear (ice_streamer_ringbuf_t *ringbuf)
{
    frost_return_if_fail (ringbuf);

    ringbuf->rd_index = 0;
    ringbuf->wr_index = 0;

    while (!frost_queue_is_empty (ringbuf->hotspots)) {
        ice_streamer_ringbuf_hotspot_t *hs;
        hs = (ice_streamer_ringbuf_hotspot_t *)frost_queue_pop_head (ringbuf->hotspots);
        if (hs->destroy)
            hs->destroy (hs->arg);
        frost_free (hs);
    }
    FROST_CondVarSignal(ringbuf->free_cond);
}

/**
 * Number of bytes free in the ringbuffer
 */
frost_uint
ice_streamer_ringbuf_bytes_free (const ice_streamer_ringbuf_t *ringbuf)
{
    frost_return_val_if_fail (ringbuf, 0);

    return ringbuf->buffer_size_usable -
           ice_streamer_ringbuf_bytes_used (ringbuf);
}

/**
 * Number of bytes used in the buffer
 */
frost_uint
ice_streamer_ringbuf_bytes_used (const ice_streamer_ringbuf_t *ringbuf)
{
    frost_return_val_if_fail (ringbuf, 0);

    if (ringbuf->wr_index >= ringbuf->rd_index) {
        return ringbuf->wr_index - ringbuf->rd_index;
    }

    return ringbuf->buffer_size - (ringbuf->rd_index - ringbuf->wr_index);
}

static frost_uint
read_bytes (ice_streamer_ringbuf_t *ringbuf, frost_uint8 *data, frost_uint len)
{
    frost_uint to_read, r = 0, cnt, tmp;
    frost_bool ok;

    to_read = MIN (len, ice_streamer_ringbuf_bytes_used (ringbuf));

    while (!frost_queue_is_empty (ringbuf->hotspots)) {
        ice_streamer_ringbuf_hotspot_t *hs = (ice_streamer_ringbuf_hotspot_t *)frost_queue_peek_head (ringbuf->hotspots);
        if (hs->pos != ringbuf->rd_index) {
            /* make sure we don't cross a hotspot */
            to_read = MIN (to_read,
                           (hs->pos - ringbuf->rd_index + ringbuf->buffer_size)
                           % ringbuf->buffer_size);
            break;
        }

        (void) frost_queue_pop_head (ringbuf->hotspots);
        ok = hs->callback (hs->arg);
        if (hs->destroy)
            hs->destroy (hs->arg);
        frost_free (hs);

        if (!ok) {
            return 0;
        }

        /* we loop here, to see if there are multiple
           hotspots in same position */
    }

    tmp = ringbuf->rd_index;

    while (to_read > 0) {
        cnt = MIN (to_read, ringbuf->buffer_size - tmp);
        memcpy (data, ringbuf->buffer + tmp, cnt);
        tmp = (tmp + cnt) % ringbuf->buffer_size;
        to_read -= cnt;
        r += cnt;
        data += cnt;
    }

    return r;
}

/**
 * Reads data from the ringbuffer. This is a non-blocking call and can
 * return less data than you wanted. Use #ice_streamer_ringbuf_wait_used to
 * ensure that you get as much data as you want.
 *
 * @param ringbuf Buffer to read from
 * @param data Allocated buffer where the readed data will end up
 * @param len number of bytes to read
 * @returns number of bytes that acutally was read.
 */
frost_uint
ice_streamer_ringbuf_read (ice_streamer_ringbuf_t *ringbuf, frost_pointer data, frost_uint len)
{
    frost_uint r;

    frost_return_val_if_fail (ringbuf, 0);
    frost_return_val_if_fail (data, 0);
    frost_return_val_if_fail (len > 0, 0);

    r = read_bytes (ringbuf, (frost_uint8 *) data, len);

    ringbuf->rd_index += r;
    ringbuf->rd_index %= ringbuf->buffer_size;

    if (r) {
        FROST_CondVarBroadcast(ringbuf->free_cond);
    }

    return r;
}

/**
 * Same as #ice_streamer_ringbuf_read but does not advance in the buffer after
 * the data has been read.
 *
 * @sa ice_streamer_ringbuf_read
 */
frost_uint
ice_streamer_ringbuf_peek (ice_streamer_ringbuf_t *ringbuf, frost_pointer data, frost_uint len)
{
    frost_return_val_if_fail (ringbuf, 0);
    frost_return_val_if_fail (data, 0);
    frost_return_val_if_fail (len > 0, 0);
    frost_return_val_if_fail (len <= ringbuf->buffer_size_usable, 0);

    return read_bytes (ringbuf, (frost_uint8 *) data, len);
}

/**
 * Same as #ice_streamer_ringbuf_read but blocks until you have all the data you want.
 *
 * @sa ice_streamer_ringbuf_read
 */
frost_uint
ice_streamer_ringbuf_read_wait (ice_streamer_ringbuf_t *ringbuf, frost_pointer data,
                        frost_uint len, FROST_Mutex *mtx)
{
    frost_uint r = 0, res;
    frost_uint8 *dest = (frost_uint8 *)data;

    frost_return_val_if_fail (ringbuf, 0);
    frost_return_val_if_fail (data, 0);
    frost_return_val_if_fail (len > 0, 0);
    frost_return_val_if_fail (mtx, 0);

    while (r < len) {
        res = ice_streamer_ringbuf_read (ringbuf, dest + r, len - r);
        r += res;
        if (r == len || ringbuf->eos) {
            break;
        }
        if (!res)
            FROST_CondVarWait (ringbuf->used_cond, mtx);
    }

    return r;
}

/**
 * Same as #ice_streamer_ringbuf_peek but blocks until you have all the data you want.
 *
 * @sa ice_streamer_ringbuf_peek
 */
frost_uint
ice_streamer_ringbuf_peek_wait (ice_streamer_ringbuf_t *ringbuf, frost_pointer data,
                        frost_uint len, FROST_Mutex *mtx)
{
    frost_return_val_if_fail (ringbuf, 0);
    frost_return_val_if_fail (data, 0);
    frost_return_val_if_fail (len > 0, 0);
    frost_return_val_if_fail (len <= ringbuf->buffer_size_usable, 0);
    frost_return_val_if_fail (mtx, 0);

    ice_streamer_ringbuf_wait_used (ringbuf, len, mtx);

    return ice_streamer_ringbuf_peek (ringbuf, data, len);
}

/**
 * Write data to the ringbuffer. If not all data can be written
 * to the buffer the function will not block.
 *
 * @sa ice_streamer_ringbuf_write_wait
 *
 * @param ringbuf Ringbuffer to put data in.
 * @param data Data to put in ringbuffer
 * @param len Length of data
 * @returns Number of bytes that was written
 */
frost_uint
ice_streamer_ringbuf_write (ice_streamer_ringbuf_t *ringbuf, frost_constpointer data,
                    frost_uint len)
{
    frost_uint to_write, w = 0, cnt;
    const frost_uint8 *src = (const frost_uint8 *)data;

    frost_return_val_if_fail (ringbuf, 0);
    frost_return_val_if_fail (data, 0);
    frost_return_val_if_fail (len > 0, 0);

    to_write = MIN (len, ice_streamer_ringbuf_bytes_free (ringbuf));

    while (to_write > 0) {
        cnt = MIN (to_write, ringbuf->buffer_size - ringbuf->wr_index);
        memcpy (ringbuf->buffer + ringbuf->wr_index, src + w, cnt);
        ringbuf->wr_index = (ringbuf->wr_index + cnt) % ringbuf->buffer_size;
        to_write -= cnt;
        w += cnt;
    }

    if (w) {
        FROST_CondVarBroadcast (ringbuf->used_cond);
    }

    return w;
}

/**
 * Same as #ice_streamer_ringbuf_write but blocks until there is enough free space.
 */

frost_uint
ice_streamer_ringbuf_write_wait (ice_streamer_ringbuf_t *ringbuf, frost_constpointer data,
                         frost_uint len, FROST_Mutex *mtx)
{
    frost_uint w = 0;
    const frost_uint8 *src = (const frost_uint8 *)data;

    frost_return_val_if_fail (ringbuf, 0);
    frost_return_val_if_fail (data, 0);
    frost_return_val_if_fail (len > 0, 0);
    frost_return_val_if_fail (mtx, 0);

    while (w < len) {
        w += ice_streamer_ringbuf_write (ringbuf, src + w, len - w);
        if (w == len || ringbuf->eos) {
            break;
        }

        FROST_CondVarWait (ringbuf->free_cond, mtx);
    }

    return w;
}

/**
 * Block until we have free space in the ringbuffer.
 */
void
ice_streamer_ringbuf_wait_free (const ice_streamer_ringbuf_t *ringbuf, frost_uint len, FROST_Mutex *mtx)
{
    frost_return_if_fail (ringbuf);
    frost_return_if_fail (len > 0);
    frost_return_if_fail (len <= ringbuf->buffer_size_usable);
    frost_return_if_fail (mtx);

    while ((ice_streamer_ringbuf_bytes_free (ringbuf) < len) && !ringbuf->eos) {
        FROST_CondVarWait(ringbuf->free_cond, mtx);
    }
}

/**
 * Block until we have used space in the buffer
 */

void
ice_streamer_ringbuf_wait_used (const ice_streamer_ringbuf_t *ringbuf, frost_uint len, FROST_Mutex *mtx)
{
    frost_return_if_fail (ringbuf);
    frost_return_if_fail (len > 0);
    frost_return_if_fail (len <= ringbuf->buffer_size_usable);
    frost_return_if_fail (mtx);

    while ((ice_streamer_ringbuf_bytes_used (ringbuf) < len) && !ringbuf->eos) {
        FROST_CondVarWait (ringbuf->used_cond, mtx);
    }
}

/**
 * Tell if the ringbuffer is EOS
 *
 * @returns frost_true if the ringbuffer is EOSed.
 */

frost_bool
ice_streamer_ringbuf_iseos (const ice_streamer_ringbuf_t *ringbuf)
{
    frost_return_val_if_fail (ringbuf, frost_true);

    return (!ice_streamer_ringbuf_bytes_used (ringbuf) && ringbuf->eos) ? frost_true : frost_false;
}

/**
 * Set EOS flag on ringbuffer.
 */
void
ice_streamer_ringbuf_set_eos (ice_streamer_ringbuf_t *ringbuf, frost_bool eos)
{
    frost_return_if_fail (ringbuf);

    ringbuf->eos = eos;

    if (eos) {
        FROST_CondVarBroadcast (ringbuf->eos_cond);
        FROST_CondVarBroadcast (ringbuf->used_cond);
        FROST_CondVarBroadcast (ringbuf->free_cond);
    }
}


/**
 * Block until we are EOSed
 */
void
ice_streamer_ringbuf_wait_eos (const ice_streamer_ringbuf_t *ringbuf, FROST_Mutex *mtx)
{
    frost_return_if_fail (ringbuf);
    frost_return_if_fail (mtx);

    while (!ice_streamer_ringbuf_iseos (ringbuf)) {
        FROST_CondVarWait (ringbuf->eos_cond, mtx);
    }

}
/** @} */

/**
 * @internal
 * Unused
 */
void
ice_streamer_ringbuf_hotspot_set (ice_streamer_ringbuf_t *ringbuf, frost_bool (*cb) (void *), void (*destroy) (void *), void *arg)
{
    ice_streamer_ringbuf_hotspot_t *hs;
    frost_return_if_fail (ringbuf);

    hs = frost_new0 (ice_streamer_ringbuf_hotspot_t, 1);
    hs->pos = ringbuf->wr_index;
    hs->callback = cb;
    hs->destroy = destroy;
    hs->arg = arg;

    frost_queue_push_tail (ringbuf->hotspots,(frost_pointer) hs);
}


