#include <utility>
#include <queue>
#include <string>
#include "rust_internal.h"
#include "uv.h"

class rust_uvtmp_thread;

struct connect_data {
    rust_uvtmp_thread *thread;
    uv_connect_t connect;
    uv_tcp_t tcp;
    chan_handle chan;
};

const intptr_t connected_tag = 1;
const intptr_t wrote_tag = 2;
const intptr_t read_tag = 3;

struct iomsg {
    intptr_t tag;
    union {
	connect_data *connected_val;
	connect_data *wrote_val;
	struct {
	    connect_data *cd;
	    uint8_t *buf;
	    ssize_t nread;
	} read_val;
    } val;
};

struct write_data {
    connect_data *cd;
    uint8_t *buf;
    size_t len;
    chan_handle chan;
};

struct read_start_data {
    connect_data *cd;
    chan_handle chan;
};

// FIXME: Copied from rust_builtins.cpp. Could bitrot easily
static void
send(rust_task *task, chan_handle chan, void *data) {
    rust_task *target_task = task->kernel->get_task_by_id(chan.task);
    if(target_task) {
        rust_port *port = target_task->get_port_by_id(chan.port);
        if(port) {
            port->send(data);
            scoped_lock with(target_task->lock);
            port->deref();
        }
        target_task->deref();
    }
}

class rust_uvtmp_thread : public rust_thread {

private:
    rust_task *task;
    uv_loop_t *loop;
    uv_idle_t idle;
    lock_and_signal lock;
    bool stop_flag;
    std::queue<std::pair<std::string, chan_handle> > connect_queue;
    std::queue<connect_data*> close_connection_queue;
    std::queue<write_data*> write_queue;
    std::queue<read_start_data*> read_start_queue;

public:

    rust_uvtmp_thread() {
	task = rust_scheduler::get_task();
	stop_flag = false;
	loop = uv_loop_new();
	uv_idle_init(loop, &idle);
	idle.data = this;
	uv_idle_start(&idle, idle_cb);
    }

    ~rust_uvtmp_thread() {
	uv_loop_delete(loop);
    }

    void stop() {
	scoped_lock with(lock);
	stop_flag = true;
    }

    void connect(char *ip, chan_handle chan) {
	scoped_lock with(lock);
	connect_queue.push(std::pair<std::string, chan_handle>
			   (std::string(ip), chan));
    }

    void
    close_connection(connect_data *cd) {
	scoped_lock with(lock);
	close_connection_queue.push(cd);
    }

    void
    write(connect_data *cd, uint8_t *buf, size_t len, chan_handle chan) {
	scoped_lock with(lock);
	write_data *wd = new write_data();
	wd->cd = cd;
	wd->buf = new uint8_t[len];
	wd->len = len;
	wd->chan = chan;

	memcpy(wd->buf, buf, len);

	write_queue.push(wd);
    }

    void
    read_start(connect_data *cd, chan_handle chan) {
	scoped_lock with(lock);
	read_start_data *rd = new read_start_data();
	rd->cd = cd;
	rd->chan = chan;

	read_start_queue.push(rd);
    }

private:

    virtual void
    run() {
	uv_run(loop);
    }

    static void
    idle_cb(uv_idle_t* handle, int status) {
	rust_uvtmp_thread *self = (rust_uvtmp_thread*) handle->data;
	self->on_idle();
    }

    void
    on_idle() {
	scoped_lock with(lock);
	make_new_connections();
	close_connections();
	write_buffers();
	start_reads();
	close_idle_if_stop();
    }

    void
    make_new_connections() {
	assert(lock.lock_held_by_current_thread());
	while (!connect_queue.empty()) {
	    std::pair<std::string, chan_handle> pair = connect_queue.front();
	    connect_queue.pop();
	    struct sockaddr_in client_addr = uv_ip4_addr("0.0.0.0", 0);
	    struct sockaddr_in server_addr = uv_ip4_addr(pair.first.c_str(), 80);

	    connect_data *cd = new connect_data();
	    cd->thread = this;
	    cd->chan = pair.second;
	    cd->connect.data = cd;

	    uv_tcp_init(loop, &cd->tcp);
	    uv_tcp_bind(&cd->tcp, client_addr);

	    uv_tcp_connect(&cd->connect, &cd->tcp, server_addr, connect_cb);
	}
    }

    static void
    connect_cb(uv_connect_t *handle, int status) {
	connect_data *cd = (connect_data*)handle->data;
	cd->thread->on_connect(cd);
    }

    void
    on_connect(connect_data *cd) {
	iomsg msg;
	msg.tag = connected_tag;
	msg.val.connected_val = cd;

	send(task, cd->chan, &msg);
    }

    void
    close_connections() {
	assert(lock.lock_held_by_current_thread());
	while (!close_connection_queue.empty()) {
	    connect_data *cd = close_connection_queue.front();
	    close_connection_queue.pop();
	    
	    cd->tcp.data = cd;
	    
	    uv_close((uv_handle_t*)&cd->tcp, tcp_close_cb);
	}
    }

    static void
    tcp_close_cb(uv_handle_t *handle) {
	connect_data *cd = (connect_data*)handle->data;
	delete cd;
    }

    void
    write_buffers() {
	assert(lock.lock_held_by_current_thread());
	while (!write_queue.empty()) {
	    write_data *wd = write_queue.front();
	    write_queue.pop();

	    uv_write_t *write = new uv_write_t();

	    write->data = wd;

	    uv_buf_t buf;
	    buf.base = (char*)wd->buf;
	    buf.len = wd->len;

	    uv_write(write, (uv_stream_t*)&wd->cd->tcp, &buf, 1, write_cb);
	}
    }

    static void
    write_cb(uv_write_t *handle, int status) {
	write_data *wd = (write_data*)handle->data;
	rust_uvtmp_thread *self = wd->cd->thread;
	self->on_write(handle, wd);
    }

    void
    on_write(uv_write_t *handle, write_data *wd) {
	iomsg msg;
	msg.tag = wrote_tag;
	msg.val.wrote_val = wd->cd;

	send(task, wd->chan, &msg);

	delete [] wd->buf;
	delete wd;
	delete handle;
    }

    void
    start_reads() {
	assert (lock.lock_held_by_current_thread());
	while (!read_start_queue.empty()) {
	    read_start_data *rd = read_start_queue.front();
	    read_start_queue.pop();

	    connect_data *cd = rd->cd;
	    cd->tcp.data = rd;

	    uv_read_start((uv_stream_t*)&cd->tcp, alloc_cb, read_cb);
	}
    }

    static uv_buf_t
    alloc_cb(uv_handle_t* handle, size_t size) {
	uv_buf_t buf;
	buf.base = new char[size];
	buf.len = size;
	return buf;
    }

    static void
    read_cb(uv_stream_t *handle, ssize_t nread, uv_buf_t buf) {
	read_start_data *rd = (read_start_data*)handle->data;
	rust_uvtmp_thread *self = rd->cd->thread;
	self->on_read(rd, nread, buf);
    }

    void
    on_read(read_start_data *rd, ssize_t nread, uv_buf_t buf) {
	iomsg msg;
	msg.tag = read_tag;
	msg.val.read_val.cd = rd->cd;
	msg.val.read_val.buf = (uint8_t*)buf.base;
	msg.val.read_val.nread = nread;

	send(task, rd->chan, &msg);
	if (nread == -1) {
	    delete rd;
	}
    }

    void
    close_idle_if_stop() {
	assert(lock.lock_held_by_current_thread());
	if (stop_flag) {
	    uv_close((uv_handle_t*)&idle, NULL);
	}
    }

};

extern "C" rust_uvtmp_thread *
rust_uvtmp_create_thread() {
    rust_uvtmp_thread *thread = new rust_uvtmp_thread();
    return thread;
}

extern "C" void
rust_uvtmp_start_thread(rust_uvtmp_thread *thread) {
    thread->start();    
}

extern "C" void
rust_uvtmp_join_thread(rust_uvtmp_thread *thread) {
    thread->stop();
    thread->join();
}

extern "C" void
rust_uvtmp_delete_thread(rust_uvtmp_thread *thread) {
    delete thread;
}

extern "C" void
rust_uvtmp_connect(rust_uvtmp_thread *thread, char *ip, chan_handle *chan) {
    thread->connect(ip, *chan);
}

extern "C" void
rust_uvtmp_close_connection(rust_uvtmp_thread *thread, connect_data *cd) {
  thread->close_connection(cd);
}

extern "C" void
rust_uvtmp_write(rust_uvtmp_thread *thread, connect_data *cd,
		 uint8_t *buf, size_t len, chan_handle *chan) {
    thread->write(cd, buf, len, *chan);
}

extern "C" void
rust_uvtmp_read_start(rust_uvtmp_thread *thread, connect_data *cd,
		      chan_handle *chan) {
    thread->read_start(cd, *chan);
}

extern "C" void
rust_uvtmp_delete_buf(uint8_t *buf) {
    delete [] buf;
}