#include "line_processor.h"

// Read from a socket/fd and callback for each "line" determined by a delimiter string.
// This isn't meant to be used seriously and is currently just for testing.

#define LINE_PROCESSOR_CHUNK_SIZE 128
static str LINE_PROCESSOR_END = al_str_c("X");

static void remove_client(struct nn_line_processor *pro, struct nn_line_processor_client *client)
{
    ev_io_stop(pro->loop->ev, &client->event);
    if (pro->type == NNWT_LINE_PROCESSOR_SOCKET) {
        nn_socket_close(&client->sock);
    }
    nn_buffer_free(&client->buf);
    al_array_remove(pro->clients, client);
    al_free(client);
}

static void read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
{
    (void)loop;
    struct nn_line_processor_client *cl = (struct nn_line_processor_client *)w->data;
    (void)revents;
    struct nn_line_processor *pro = cl->pro;
    u8 type = pro->type;

    nn_buffer_ensure_space(&cl->buf, cl->index + LINE_PROCESSOR_CHUNK_SIZE);
    u8 *ptr = nn_buffer_get_ptr(&cl->buf, cl->index);

    ssize_t ret;
    switch (type) {
    case NNWT_LINE_PROCESSOR_SOCKET:
        ret = nn_socket_read(&cl->sock, ptr, LINE_PROCESSOR_CHUNK_SIZE);
        break;
    case NNWT_LINE_PROCESSOR_FD:
        ret = nn_read(pro->fd, ptr, LINE_PROCESSOR_CHUNK_SIZE);
        break;
    default:
        al_assert_and_return();
    }

    if (ret <= 0) {
        return;
    }

    cl->index += ret;
    ptr = nn_buffer_get_ptr(&cl->buf, 0);
    str s = al_str_w((char *)ptr, 0, cl->index);

    str *dl = &pro->delim;
    u32 offset = dl->length - 1;
    for (size_t i = cl->mark; i < cl->index - offset; i++) {
        if (al_str_cmp(&s, dl, i, dl->length) == 0) {
            s.length = i;

            if (al_str_eq(&s, &LINE_PROCESSOR_END)) {
                remove_client(pro, cl);
                return;
            }

            switch (pro->callback(pro->userdata, &s)) {
            case NNWT_LINE_PROCESSOR_STOP:
                nn_line_processor_stop(pro);
                return;
            case NNWT_LINE_PROCESSOR_CONTINUE:
                break;
            }

            size_t skip = i + offset + 1;
            cl->index = cl->index - skip;
            al_memmove(ptr, &ptr[skip], cl->index);
            s = al_str_w((char *)ptr, 0, cl->index);

            cl->mark = i = 0;
        }
    }

    cl->mark = cl->index;
}

static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 revents)
{
    (void)loop;
    struct nn_line_processor *pro = (struct nn_line_processor *)w->data;
    (void)revents;

    struct nn_line_processor_client *cl = al_alloc_object(struct nn_line_processor_client);
    if (!nn_socket_accept(pro->sock, &cl->sock, NNWT_SOCKET_NONBLOCKING)) {
        al_free(cl);
        return;
    }
    cl->index = 0;
    nn_buffer_init(&cl->buf);
    cl->mark = 0;
    cl->pro = pro;
    cl->event.data = cl;
    al_array_push(pro->clients, cl);

    ev_io_init_n(&cl->event, read_callback, nn_socket_get_fd(&cl->sock), EV_READ);
    ev_io_start(pro->loop->ev, &cl->event);
}

void nn_line_processor_init(struct nn_line_processor *pro, str *delim)
{
    pro->loop = NULL;
    al_str_clone(&pro->delim, delim);
    al_array_init(pro->clients);
}

void nn_line_processor_open_socket(struct nn_line_processor *pro, struct nn_socket *sock)
{
    pro->type = NNWT_LINE_PROCESSOR_SOCKET;
    pro->sock = sock;
}

void nn_line_processor_open_fd(struct nn_line_processor *pro, s32 fd)
{
    pro->type = NNWT_LINE_PROCESSOR_FD;
    pro->fd = fd;
}

void nn_line_processor_run(struct nn_line_processor *pro, struct nn_event_loop *loop)
{
    pro->loop = loop;
    switch (pro->type) {
    case NNWT_LINE_PROCESSOR_SOCKET:
        pro->event.data = pro;
        ev_io_init_n(&pro->event, socket_connection_callback, nn_socket_get_fd(pro->sock), EV_READ);
        ev_io_start(pro->loop->ev, &pro->event);
        break;
    case NNWT_LINE_PROCESSOR_FD: { // Not supported on windows.
        struct nn_line_processor_client *cl = al_alloc_object(struct nn_line_processor_client);
        cl->index = 0;
        nn_buffer_init(&cl->buf);
        cl->mark = 0;
        cl->pro = pro;
        cl->event.data = cl;
        al_array_push(pro->clients, cl);
        ev_io_init_n(&cl->event, read_callback, pro->fd, EV_READ);
        ev_io_start(pro->loop->ev, &cl->event);
        break;
    }
    }
}

void nn_line_processor_stop(struct nn_line_processor *pro)
{
    if (pro->loop) {
        switch (pro->type) {
        case NNWT_LINE_PROCESSOR_SOCKET:
            ev_io_stop(pro->loop->ev, &pro->event);
            break;
        case NNWT_LINE_PROCESSOR_FD:
            break;
        }
        struct nn_line_processor_client *cl;
        al_array_foreach(pro->clients, i, cl) {
            remove_client(cl->pro, cl);
        }
    }
    al_array_free(pro->clients);
    al_str_free(&pro->delim);
}