#include <assert.h>
#include <string.h>
#include "ucore/wqueue.h"
#include "ucore/iomux.h"

static int uc_wqueue_handle_write(struct UCWQueue *wqueue)
{
    int rc = UC_WQ_DONE;
    struct IOMuxFD *fd = &wqueue->fd;
    struct IOMux *mux  = wqueue->mux;

    while (!uc_tailq_empty(&wqueue->queue)) {
        struct TailQ *q;
        struct MBuf *mbuf;
        q = UC_TAILQ_FIRST(&wqueue->queue);

        mbuf = UC_TAILQ_CONTAINER(q, struct MBuf, entry);
        rc = wqueue->write_cb(mux, fd, mbuf);
        if (rc <= 0) {
            //NOTE for rc < 0, nothing can touch the wqueue or fd
            //any more, as it might been free'd
            break;
        }

        mbuf = uc_mbuf_dequeue(&wqueue->queue);
        assert(mbuf != NULL);
        uc_mbuf_free(mbuf);
        wqueue->queue_len--;
    }

    if (rc >= 0 && uc_tailq_empty(&wqueue->queue)
        && (wqueue->fd.what & MUX_EV_WRITE) != 0) {

        int urc;

        urc = iomux_clear_event(mux, 
                                  fd, 
                                  MUX_EV_WRITE);
        assert(urc == 0);

    }

    return rc;
}



static void uc_wqueue_cb(struct IOMux *mux, struct IOMuxFD *fd, unsigned int what)
{
    int rc = 0;
    struct UCWQueue *wqueue = UC_CONTAINER_OF(fd, struct UCWQueue, fd);

    if (what & MUX_EV_WRITE) {
        rc = uc_wqueue_handle_write(wqueue);
        //NOTE - can't touch the wqueue or fd if rc < 0
        //as it might been free'd
    }


    if (rc >= 0 && what & MUX_EV_READ) {
        rc = wqueue->read_cb(mux, fd);
        //NOTE - can't touch the wqueue or fd after this point,
        //as it might been free'd
    }
}

void uc_wqueue_init(struct UCWQueue *wqueue, struct IOMux *mux)
{
    assert(mux != NULL);

    memset(wqueue, 0, sizeof *wqueue);
    wqueue->mux = mux;
    wqueue->fd.callback = uc_wqueue_cb;
    
    uc_tailq_init(&wqueue->queue);
}

int uc_wqueue_clear(struct UCWQueue *wqueue)
{
    struct MBuf *buf;
    int rc = 0;

    while ((buf = uc_mbuf_dequeue(&wqueue->queue)) != NULL) {
        uc_mbuf_free(buf);
        wqueue->queue_len--;
    }

    if ((wqueue->fd.what & MUX_EV_WRITE) != 0) {
        rc = iomux_clear_event(wqueue->mux, 
                                &wqueue->fd, 
                                MUX_EV_WRITE);
    }

    assert(rc == 0);

    return rc;
}

int uc_wqueue_enqueue(struct UCWQueue *wqueue, struct MBuf *mbuf)
{
    int rc = 0;

    uc_mbuf_enqueue(mbuf, &wqueue->queue);
    wqueue->queue_len++;

    if ((wqueue->fd.what & MUX_EV_WRITE) == 0) {
        rc = iomux_add_event(wqueue->mux, 
                                &wqueue->fd, 
                                MUX_EV_WRITE);
        assert(rc == 0);
    }

    return 0;
}

int uc_wqueue_flush(struct UCWQueue *wqueue)
{
    if (uc_wqueue_handle_write(wqueue) < 0) {
        return -1;
    }

    return 0;
}