#include "ice_streamer.h"

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

#include "utilities_debug/cabot_assert.h"
#include "utilities_debug/trace.h"
#include "frost_mem.h"

TRACE_IMPLEMENT("ice/ice_streamer_ringbuf_handler");

/*
   - producer:
     want_buffer -> buffering
     want_seek -> seeked
     want_stop -> stopped

   - consumer:
     * -> want_seek
          wait until state == seek_done
      seek_done -> want_buffer

     * -> want_stop
          wait until state == is_stopped

*/


static ice_streamer_handler_plugin_t *ringbuf_plugin;

void ice_streamer_ringbuf_handler_thread (void* data);

frost_bool
ice_streamer_ringbuf_plugin_init (ice_streamer_handler_t * handle)
{
    ice_streamer_ringbuf_priv_t *priv;

    priv = frost_new0 (ice_streamer_ringbuf_priv_t, 1);

    ice_streamer_private_data_set (handle, priv);

    priv->buffer = ice_streamer_ringbuf_new (CURRENT_MAX_PACKET_SIZE);
    priv->buffer_lock = FROST_MutexCreate() ;
    priv->state = STATE_WANT_BUFFER;
    priv->state_cond = FROST_CondVarCreate();
    priv->state_lock = FROST_MutexCreate ();
    priv->thread = FROST_ThreadCreate(ice_streamer_ringbuf_handler_thread, (void*)handle, 2048 + 2*CURRENT_MAX_PACKET_SIZE, FROST_medium_priority, "Ring Buffer");

    ice_streamer_handler_outdata_type_copy (handle);

    return frost_true;
}

void
ice_streamer_ringbuf_plugin_destroy (ice_streamer_handler_t * handle)
{
    ice_streamer_ringbuf_priv_t *priv;
    priv = (ice_streamer_ringbuf_priv_t *)ice_streamer_private_data_get (handle);

    FROST_MutexLock (priv->state_lock);
    ice_streamer_ringbuf_clear (priv->buffer);
    while (priv->state != STATE_IS_STOPPED) {
        priv->state = STATE_WANT_STOP;
        FROST_CondVarWait (priv->state_cond, priv->state_lock);
    }
    FROST_MutexUnlock(priv->state_lock);

    FROST_ThreadDelete (priv->thread);

    TRACE_INFO(("Ringbuf destroyed!\n"));
}

frost_int
ice_streamer_ringbuf_plugin_read (ice_streamer_handler_t * handle, void *buffer, frost_int len, ice_error_code_t *error)
{
    ice_streamer_ringbuf_priv_t *priv;
    priv = (ice_streamer_ringbuf_priv_t *)ice_streamer_private_data_get (handle);

    return ice_streamer_ringbuf_read_wait (priv->buffer, buffer, len, priv->buffer_lock);
}


frost_bool
ice_streamer_ringbuf_plugin_setup (ice_streamer_handler_plugin_t *plugin)
{
    ice_streamer_handler_methods_t methods;

    memset (&methods,0,sizeof(ice_streamer_handler_methods_t));
    methods.init = ice_streamer_ringbuf_plugin_init;
    methods.destroy = ice_streamer_ringbuf_plugin_destroy;
    methods.read = ice_streamer_ringbuf_plugin_read;
    /*
      methods.seek
    */

    ice_streamer_handler_plugin_methods_set(plugin, &methods);

    ringbuf_plugin = plugin;

    return frost_true;
}

ice_streamer_handler_t *
ice_streamer_ringbuf_xform_new (ice_streamer_handler_t *prev, frost_int32 entry, FROST_List *gt)
{
    ice_streamer_handler_t * handle;

    handle = ice_streamer_handler_new (ringbuf_plugin, prev, entry, gt);

    return handle;
}

void
fill (ice_streamer_handler_t * handle, ice_streamer_ringbuf_priv_t *priv)
{
    ice_error_code_t err;
    char buf[CURRENT_MAX_PACKET_SIZE];
    int res;

    res = ice_streamer_handler_read (handle, buf, CURRENT_PACKET_SIZE, &err);
    if (res > 0) {
        ice_streamer_ringbuf_write_wait (priv->buffer, buf, res, priv->buffer_lock);
    } else if (res == -1) {
        /* XXX copy error */
        FROST_MutexLock (priv->state_lock);
        priv->state = STATE_WANT_STOP;
    } else {
        ice_streamer_ringbuf_set_eos (priv->buffer, frost_true);
        priv->state = STATE_WANT_STOP;
    }
}

void
ice_streamer_ringbuf_handler_thread (void* data)
{
    ice_streamer_handler_t * handle = (ice_streamer_handler_t *)data;
    ice_streamer_ringbuf_priv_t *priv;

    priv = (ice_streamer_ringbuf_priv_t *)ice_streamer_private_data_get (handle);

    FROST_MutexLock (priv->state_lock);
    while (priv->state != STATE_WANT_STOP) {
        if (priv->state == STATE_WANT_BUFFER) {
            priv->state = STATE_BUFFERING;
            FROST_CondVarSignal (priv->state_cond);
            while (priv->state == STATE_BUFFERING) {
                FROST_MutexUnlock (priv->state_lock);
                TRACE_INFO(("ice_streamer_ringbuf_handler_thread Running!\n"));
                fill (handle, priv);
                FROST_MutexLock (priv->state_lock);
            }
        } else if (priv->state == STATE_WANT_SEEK) {
            /** **/
            priv->state = STATE_SEEK_DONE;
            FROST_CondVarSignal (priv->state_cond);
            while (priv->state == STATE_SEEK_DONE) {
                FROST_CondVarWait(priv->state_cond, priv->state_lock);
            }
        }
        TRACE_INFO(("thread: state: %d \n", priv->state));
    }
    priv->state = STATE_IS_STOPPED;
    FROST_CondVarSignal (priv->state_cond);
    FROST_MutexUnlock(priv->state_lock);

}


ICE_STREAMER_XFORM_BUILTIN (ringbuf,
                            "Ringbuffer",
                            ICE_STREAMER_VERSION,
                            "Buffer",
                            ice_streamer_ringbuf_plugin_setup);

const ice_streamer_plugin_desc_t * getRingBufPluginDesc()
{
    return &ice_streamer_builtin_ringbuf ;
}

