#include <cstring></cstring>

#include <climits></climits>

#include <chrono></chrono>

#include <thread></thread>

#include <mutex></mutex>

#include <unistd.h></unistd.h>

#include <sys types.h=""></sys>

#include <sys socket.h=""></sys>

#include <sys epoll.h=""></sys>

#include <netinet in.h=""></netinet>

#include "protocol.h"

#include "connection.h"

#define DebugMSG ShowDebugMSG

Connection::Connection():

	server(), srvPrev(), srvNext() { }

Connection::~Connection()

	{ destroy(); }

bool Connection::open(Protocol &protocol, void *address, size_t addressSize, int sockId, Server *server) {

	OpenLock lock(*this, protocol, address, addressSize);

	if (!lock.success)

		return false;

	this->sockId = sockId;

	this->server = server;

	return true;

}

bool Connection::open(Protocol &protocol, void *address, size_t addressSize)

	{ return open(protocol, address, addressSize, -1, nullptr); }

bool Connection::read(Task &task) {

	DebugMSG();

	ReadLock lock(stateProtector);

	State s = state;

	if (s < STATE_OPEN || s >= STATE_CLOSING)

		return false;

	DebugMSG();

	readQueue.push(task);

	updateEvents(readQueue.count(), writeQueue.count());

	return true;

}

bool Connection::write(Task &task) {

	DebugMSG();

	ReadLock lock(stateProtector);

	State s = state;

	if (s < STATE_OPEN || s >= STATE_CLOSING)

		return false;

	DebugMSG();

	writeQueue.push(task);

	updateEvents(readQueue.count(), writeQueue.count());

	return true;

}

void Connection::handleState() {

	if (stateLocal == STATE_NONE) {

		DebugMSG("none");

		bool connected = false;

		if (sockId >= 0) {

			DebugMSG("accepting");

			initSocket(sockId);

			connected = true;

		} else

		if (addressSize >= sizeof(sa_family_t)) {

			DebugMSG("connecting");

			struct sockaddr *addr = (struct sockaddr*)address;

			sockId = ::socket(addr->sa_family, SOCK_STREAM, 0);

			if (sockId >= 0) {

				initSocket(sockId);

				if (0 == ::connect(sockId, addr, (int)addressSize))

					connected = true;

			} else {

				DebugMSG("none -> cannot create socket");

			}

		}

		if (sockId >= 0) {

			DebugMSG("opening");

			prev = protocol->sockLast;

			(prev ? prev->next : protocol->sockFirst) = this;

			if (server) {

				srvPrev = server->connLast;

				(srvPrev ? srvPrev->srvNext : server->connFirst) = this;

			}

			updateEvents(false, !connected);

			if (connected) {

				DebugMSG("to open");

				state = stateLocal = STATE_OPEN;

				stateProtector.wait();

				onOpen(address, addressSize);

			} else {

				assert(!server);

				DebugMSG("to connecting");

				state = stateLocal = STATE_CONNECTING;

				stateProtector.wait();

			}

		} else {

			DebugMSG("none -> to closed");

			Server *srv = server;

			assert(connected || !server);

			server = nullptr;

			sockId = -1;

			onOpeningError();

			state = STATE_CLOSED;

			stateProtector.wait();

			while(closeWaiters > 0)

				{ closeCondition.notify_all(); std::this_thread::yield(); }

			state = STATE_FINISHED;

			DebugMSG("none -> to finished");

			if (srv) server->handleDisconnect(*this, true);

		}

	} else {

		State sw = stateWanted;

		if ( stateLocal == STATE_CONNECTING

		  && sw == STATE_CLOSING )

		{

			DebugMSG("connecting -> to closing");

			onOpeningError();

			state = stateLocal = STATE_CLOSING;

			stateProtector.wait();

		}

		if ( stateLocal == STATE_OPEN

		  && sw == STATE_CLOSE_REQ )

		{

			DebugMSG("to close req");

			state = stateLocal = STATE_CLOSE_REQ;

			stateProtector.wait();

			onCloseReqested();

		}

		if ( stateLocal >= STATE_OPEN

		  && stateLocal <= STATE_CLOSE_REQ

		  && sw == STATE_CLOSING )

		{

			DebugMSG("to closing");

			state = stateLocal = STATE_CLOSING;

			stateProtector.wait();

			while(Task *task = readQueue.peek())

				{ readQueue.pop(); onReadReady(*task, true); }

			while(Task *task = writeQueue.peek())

				{ writeQueue.pop(); onWriteReady(*task, true); }

			onClose(error);

		}

		if ( stateLocal == STATE_CLOSING && !enqueued ) {

			DebugMSG("to closed");

			Server *srv = server;

			bool err = error;

			if (server) {

				(srvPrev ? srvPrev->srvNext : server->connFirst) = srvNext;

				(srvNext ? srvNext->srvPrev : server->connLast ) = srvPrev;

				server = nullptr;

			}

			error = false;

			(prev ? prev->next : protocol->sockFirst) = next;

			(next ? next->prev : protocol->sockLast ) = prev;

			if (sockId >= 0) {

				struct epoll_event event = {};

				epoll_ctl(protocol->epollFd, EPOLL_CTL_DEL, sockId, &event);

				::close(sockId);

				sockId = -1;

			}

			events = 0xffffffff;

			protocol = nullptr;

			memset(address, 0, sizeof(address));

			addressSize = 0;

			stateLocal = STATE_NONE;

			state = STATE_CLOSED;

			stateWanted = STATE_NONE;

			while(closeWaiters > 0)

				{ closeCondition.notify_all(); std::this_thread::yield(); }

			state = STATE_FINISHED;

			DebugMSG("to finished");

			if (srv) srv->handleDisconnect(*this, err);

		}

	}

}

void Connection::handleEvents(unsigned int events) {

	bool needUpdate = false;

	DebugMSG("%08x", events);

	if (stateLocal == STATE_CONNECTING) {

		DebugMSG("connecting");

		if (events & (EPOLLHUP | EPOLLERR)) {

			close(true);

			return;

		}

		if (events & EPOLLOUT) {

			DebugMSG("connecting -> to open");

			state = stateLocal = STATE_OPEN;

			stateProtector.wait();

			onOpen(address, addressSize);

			needUpdate = true;

		}

	}

	if ( stateLocal < STATE_OPEN

	  || stateLocal >= STATE_CLOSING )

		return;

	bool closing = (bool)(events & (EPOLLHUP | EPOLLRDHUP | EPOLLERR));

	if (events & EPOLLIN) {

		DebugMSG("input");

		bool ready = true;

		int pops = 0;

		int count = (readQueue.count() + 2)*2;

		while(ready) {

			Task *task = readQueue.peek();

			if (!task) {

				if (pops) needUpdate = true;

				break;

			}

			while(task->size > task->completion) {

				size_t size = task->size - task->completion;

				if (size > INT_MAX) size = INT_MAX;

				int res = ::recv(

					sockId,

					(unsigned char*)task->data + task->completion,

					(int)size, MSG_DONTWAIT );

				if (res <= 0) {

					ready = false;

					break;

				} else {

					task->completion += res;

				}

			}

			if (task->size <= task->completion) {

				readQueue.pop();

				onReadReady(*task, false);

				++pops;

				if (!closing && --count < 0) ready = false;

			}

		}

	}

	if (closing) {

		DebugMSG("hup");

		close((bool)(events & EPOLLERR));

		return;

	}

	if (events & EPOLLOUT) {

		DebugMSG("output");

		bool ready = true;

		int pops = 0;

		int count = writeQueue.count() + 2;

		while(ready) {

			Task *task = writeQueue.peek();

			if (!task) {

				if (pops) needUpdate = true;

				break;

			}

			while(task->size > task->completion) {

				size_t size = task->size - task->completion;

				if (size > INT_MAX) size = INT_MAX;

				int res = ::send(

					sockId,

					(const unsigned char*)task->data + task->completion,

					(int)size, MSG_DONTWAIT );

				if (res <= 0) {

					ready = false;

					break;

				} else {

					task->completion += res;

				}

			}

			if (task->size <= task->completion) {

				writeQueue.pop();

				onWriteReady(*task, false);

				++pops;

				if (--count <= 0) ready = false;

			}

		}

	}

	if (needUpdate)

		updateEvents(readQueue.count(), writeQueue.count());

}

void Connection::onReadReady(Task&, bool) { }

void Connection::onWriteReady(Task&, bool) { }