
#include <sstream>
#include <iostream>

#include "log.h"
#include "server.h"
#include "sockets.h"
#include "threads.h"
#include "datetime.h"
#include "error_handling.h"
#include "context.h"

#ifdef _DEBUG
#ifndef DEBUG_NEW
#define DEBUG_NEW new(_NORMAL_BLOCK, __FILE__, __LINE__)
#define new DEBUG_NEW
#endif
#endif

// ################################################################################################

acceptingThread :: acceptingThread(server *srv, sockets :: serverSocket* socket)
{
	srv_ = srv;
	socket_ = socket;
}

// ------------------------------------------------------------------------------------------------

void acceptingThread :: operator ()(threads::thread *thd)
{
	while (true)
	{
		sockets :: clientSocket* cs = NULL;
		try
		{
			// prijmeme klientsky socket
			cs = NULL;
			cs = socket_->acceptConnection();
	
			// vytvorime klienta
			client* c = new client(cs, srv_);
			srv_->addUnauthorizedClient(c);
			c->startClient();
		}
		catch (std :: exception e)
		{
			// v pripade naalokovaneho klienta ho odalokujeme
			if (cs != NULL)
			{
				try
				{
					cs->close();
					delete cs;
					cs = NULL;
				}
				catch (std :: exception f)
				{
					errExceptionError(f.what());
				}
			}

			// zjistili jsme chybu, nahlasime ji a skoncime
			// errExceptionError(e.what());
			break;
		}
	}
}

// ################################################################################################

client :: client(sockets::clientSocket *socket, server *srv)
{
	userId_               = make_pair(0, 0);
	socket_               = socket;
	lastActivity_         = dv_time:: now();
	srv_                  = srv;
	ownThread_            = new threads :: thread(this, true);
	sender_               = new sender(this);
	senderThread_         = new threads :: thread(sender_, true);
	errCounter_           = 0;
}

// ------------------------------------------------------------------------------------------------

void client :: startClient()
{
	ownThread_->start();
	senderThread_->start();
}

// ------------------------------------------------------------------------------------------------

client :: ~client()
{
	delete ownThread_;
	delete senderThread_;
	delete sender_;
	senderThread_ = NULL;
	sender_ = NULL;

	delete socket_;
	socket_ = NULL;
}

// ------------------------------------------------------------------------------------------------

void client :: operator ()(threads::thread *thd)
{
	ownThread_ = thd;

	// autorizujeme klienta 
	try
	{
		userId_ =  make_pair(0, 0);

        // prvne, nacteme zpravu
        char* authMsg = NULL;
        if (getMessage(&authMsg))
        {
            // prelozime zpravu
            protocolReader authReader;
            message* m = authReader.translateMessage(authMsg);
            delete authMsg;

            if (m != NULL)
            {
                // zpracujeme zpravu, ta musi byt nutne logovaci
                gctx.eng.beginWork();
                // NENASTAVUJEME kontext, protoze jde o prihlaseni
                userId_ = gctx.eng.processRequest(make_pair(0,0), m->getRequestName(), m->getValues());
                gctx.eng.endWork();

#if defined(_DEBUG) && defined(DV_SERVER_VERBOSE) && defined(DV_SERVER_VERBOSE_LOGON)
                stringstream ss;
                ss << "Client [" << userId_.first << "] logged on!";
                log(ss.str());
#endif

                // smazeme zpravu
                delete m;
            }
        }
	}
	catch (std :: exception e)
	{
		userId_ = make_pair(0, 0);
	}

	if (userId_.first != 0)
	{
		// zaregistrujeme klienta, pokud uz zaregistrovan je, ukonci spojeni
		if (!srv_->moveFromUCToMap(this, userId_))
		{
			close();
			srv_->moveFromUCToRemoving(this);
			send(NULL);
		}
		else
		{
            /* posleme zpravu o uspesnem loginu */
            protocolWriter w;
            w.beginMessage(MT_ANSWER, REQUEST_LOGIN);
            w.endMessage(true);
            send(w.copyMessage());

			// prebirame zpravy
			while (true)
			{
				char* currentMsg = NULL;

				if (!getMessage(&currentMsg))
				{
					// nastala kriticka chyba
					break;
				}

				// preloz zpravu do protokolu
				protocolReader reader;
				message* m = reader.translateMessage(currentMsg);
				delete currentMsg;
				currentMsg = NULL;

				// v pripade neuspechu pridej jednicku k pocitadlu chyb,
				// v pripade uspechu vynuluj pocitadlo a sniz zpravu
				if (m != NULL)
				{
					errCounter_ = 0;
					dispatchMessage(m);
				}
				else 
				{
					errCounter_++;
					if (MAX_BAD_REQUEST_BEFORE_CLOSE <= errCounter_) break;
				}							
			}

			// konec zprav, zavri spojeni a uloz ho do fronty na uvolneni
			close();
			srv_->moveFromMapToRemoving(userId_);
			send(NULL);
		}
	}
	else
	{
		// zavri spojeni a uloz ho do fronty na uvolneni
		close();
		srv_->moveFromUCToRemoving(this);
		send(NULL);
	}
}

// ------------------------------------------------------------------------------------------------

bool client :: getMessage(char** result)
{
	int length = 0;
	try
	{
		// nacteme ze socketu pocet znaku v nasledujici zprave
		std :: string len = "";

		while (true)
		{
			char c;
			if (socket_->recvData(&c, 1) > 0)
			{
				if (isdigit((unsigned char)c))
				{
					len += c;
				}
				else break;
			}
			else
			{
				return false;
			}
		}

		length = atoi(len.c_str());
	}
	catch (std :: exception e)
	{
		return false;
	}

	// zkontrolujeme, jestli zprava neni prilis velka
	if (length > MAX_MESSAGE_LENGTH || length <= 0)
	{
		return false;
	}

	// naalokujeme buffer pro zpravu
	char* msg = NULL;
	try
	{
		msg = new char[length + 1];
	}
	catch (std :: bad_alloc e)
	{
		errExceptionError("Not enough memory to allocate client buffer");
		return false;
	}

	// nacteme zpravu
	try
	{
		for (int i = 0; i < length; i++) 
		{
			char c;
			if (socket_->recvData(&c, 1) > 0)
			{
				msg[i] = c;
			}
			else
			{
				delete[] msg;
				return false;
			}
		}
	}
	catch (std :: exception e)
	{
		delete[] msg;
		return false;
	}
	msg[length] = 0;

	// nastavime zpravu
	*result = msg;
    lastActivity_ = dv_time:: now();

	return true;
}

// ------------------------------------------------------------------------------------------------

void client :: close()
{
	try
	{
		socket_->close();
	}
	catch (std :: exception e)
	{
		errExceptionError(e.what());
	}
}

// ------------------------------------------------------------------------------------------------

void client :: waitForThreadExit(void)
{
	ownThread_->join();
	senderThread_->join();
}

// ------------------------------------------------------------------------------------------------

void client :: dispatchMessage(message* msg)
{
	srv_->dispatchMessage(userId_, msg);
}

// ------------------------------------------------------------------------------------------------

void client :: send(char *msg)
{
	sender_->send(msg);
}

// ################################################################################################

void server :: dispatchMessage(id idUser, message* msg)
{
	main_->dispatchMessage(idUser, msg);
}

// ------------------------------------------------------------------------------------------------

void server :: addUnauthorizedClient(client *c)
{
	clientsLock.down();

	unauthorizedClients.insert(c);

	clientsLock.up();
}

// ------------------------------------------------------------------------------------------------

void server :: moveFromMapToRemoving(id userId)
{
	clientsLock.down();

	client* c = clients[userId];
	clients.erase(userId);
	removingClients.push_back(c);

	clientsLock.up();
    cleaner_->notify();
}

// ------------------------------------------------------------------------------------------------

bool server :: moveFromUCToMap(client *c, id userId)
{
	bool result;
	clientsLock.down();

	// pokud naleznes klienta, vrat false, jinak ho prerad do mapy z neautorizovanych klientu
	if (clients.find(userId) == clients.end())
	{
		unauthorizedClients.erase(c);
		clients[userId] = c;
		result = true;
	} 
	else
	{
		result = false;
	}

	clientsLock.up();
	return result;
}

// ------------------------------------------------------------------------------------------------

void server :: closeClient(id idUser)
{
    clientsLock.down();

    std :: map<id, client*> :: iterator iter;
    if ((iter = clients.find(idUser)) != clients.end())
    {
        iter->second->close();
    }

    clientsLock.up();
}

// ------------------------------------------------------------------------------------------------

void server :: moveFromUCToRemoving(client *c)
{
	clientsLock.down();

	unauthorizedClients.erase(c);
	removingClients.push_back(c); 

	clientsLock.up();
    cleaner_->notify();
}

// ------------------------------------------------------------------------------------------------

server :: server() : clientsLock(1, 1)
{
	socket_ = NULL;
	acceptor_ = NULL;
	acceptingThread_ = NULL;
	cleaner_ = NULL;
	cleaningThread_ = NULL;
	main_ = NULL;
	mainThread_ = NULL;
}

// ------------------------------------------------------------------------------------------------

bool server :: start(int port, int maxConnections)
{
	bool result = true;

	try
	{
		// vytvor instanci socketu
		socket_ = new sockets :: serverSocket();

		// zacni poslouchat na portu
		socket_->listenPort(port, maxConnections);

		// 
		main_ = new mainThread(this);
		mainThread_ = new threads :: thread(main_, false);

		// vytvor akceptujici vlakno
		acceptor_ = new acceptingThread(this, socket_);
		acceptingThread_ = new threads :: thread(acceptor_, false);

		// vytvor nicici vlakno
		cleaner_ = new cleaningThread(this, dv_time:: datetime :: second() * (dv_time:: datetime)30);
		cleaningThread_ = new threads :: thread(cleaner_, false);
	}
	catch (std :: exception e)
	{
		result = false;
		errExceptionError(e.what());
	}

	return result;
}

// ------------------------------------------------------------------------------------------------

void server :: stop()
{
	// uzavreme socket
	try
	{
		socket_->close();
	}
	catch (std :: exception e)
	{
		errExceptionError(e.what());
	}

	// joinujeme cekaci vlakno
	acceptingThread_->join();

	// uvolnime acceptacni thread a jeho runnable objekt
	delete acceptor_;			acceptor_			= NULL;
	delete acceptingThread_;	acceptingThread_	= NULL;

	// uvolnime socket;
	delete socket_;				socket_				= NULL;

	// pockame na konec uvolnovaciho threadu
	cleaner_->prepareFinish();
	cleaningThread_->join();
	delete cleaner_;			cleaner_			= NULL;
	delete cleaningThread_;		cleaningThread_		= NULL;

	main_->prepareFinish();
	main_->dispatchMessage(make_pair(0, 0), NULL);
	mainThread_->join();
	delete main_;				main_				= NULL;
	delete mainThread_;			mainThread_			= NULL;
}

// ------------------------------------------------------------------------------------------------

bool server :: sendToClient(id idUser, char *msg)
{
	bool result;
	clientsLock.down();

	// pokud naleznes klienta, vrat false, jinak ho prerad do mapy z neautorizovanych klientu
	if (clients.find(idUser) != clients.end())
	{
		clients[idUser]->send(msg);
		result = true;
	} 
	else
	{
		result = false;
	}

	clientsLock.up();
	return result;
}

// ################################################################################################

cleaningThread :: cleaningThread(server *srv, dv_time:: datetime delta) : sem_(0, 16777216)
{
	srv_ = srv;
	finishing_ = false;
	delta_ = delta;
}

// ------------------------------------------------------------------------------------------------

void cleaningThread :: notify()
{
    sem_.up();
}

// ------------------------------------------------------------------------------------------------

void cleaningThread :: operator ()(threads::thread *thd)
{
	while (true)
	{
		std :: list<client*> clientsToDelete;

		// uzamkni klienty
		srv_->clientsLock.down();

		// projdi klienty a jejich posledni aktivitu, zrus stare, popr. pri ukonceni ukonci vsechny
		for (std :: map<id, client*> :: iterator iter = srv_->clients.begin(); 
			iter != srv_->clients.end(); iter++)
		{
			if ((iter->second->lastActivity_ + delta_ < dv_time:: now()) || finishing_)
				iter->second->close();
		}

		// projdi neautorizovane klienty, zrus stare, popr. pri ukonceni ukonci vsechny
		for (stdext :: hash_set<client*> :: iterator iter = srv_->unauthorizedClients.begin();
			iter != srv_->unauthorizedClients.end(); iter++)
		{
			if (((*iter)->lastActivity_ + delta_ < dv_time:: now()) || finishing_)
				(*iter)->close();
		}

		// kdyz koncime, zjisti, zda jsme uvolnili vsechno a pripadne skonci
		if (finishing_ &&
			srv_->clients.size() == 0 &&
			srv_->unauthorizedClients.size() == 0 &&
			srv_->removingClients.size() == 0)
		{
			srv_->clientsLock.up();
			break;
		}

		// kvuli rycholsti zde klienty na uvolneni jen zkopirujeme, abychom mohli rychle uvolnit zamek
		clientsToDelete.assign(srv_->removingClients.begin(), srv_->removingClients.end());
		srv_->removingClients.clear();

		srv_->clientsLock.up();

		// uvolnime klienty ve fronte na uvolneni
		for (std :: list<client*> :: iterator iter = clientsToDelete.begin(); iter != clientsToDelete.end(); iter++)
		{
			client* c = *iter;
			c->waitForThreadExit();
			delete c;
			*iter = NULL;
		}

        // pouzijeme radsi semafor... bude to rychlejsi
        sem_.down(CLEANING_THREAD_INTERVAL_MSEC);
	}
}

// ------------------------------------------------------------------------------------------------

void cleaningThread :: prepareFinish()
{
	finishing_ = true;
    notify();
}

// ################################################################################################

/*static int clId = 0;

id authorizeUser(sockets :: clientSocket* socket)
{
	return make_pair(++clId, 0);
}*/

// ################################################################################################

mainThread :: mainThread(server *srv) : workerSemaphore_(0, MAX_WORKER_QUEUE_ITEMS), listLock_(1, 1)
{
	srv_ = srv;
	finishing_ = false;
}

// ------------------------------------------------------------------------------------------------

void mainThread :: dispatchMessage(id idUser, message *msg)
{
	listLock_.down();

	msgItem item = {msg, idUser};
	messages_.push_back(item);

	listLock_.up();
	workerSemaphore_.up();
}

// ------------------------------------------------------------------------------------------------

void mainThread :: prepareFinish()
{
	finishing_ = true;
}

// ------------------------------------------------------------------------------------------------

void mainThread :: operator ()(threads::thread *thd)
{
	while (true)
	{
		msgItem msg = {NULL, make_pair(0,0)};

		workerSemaphore_.down();
		listLock_.down();

		// jestlize koncime, vyber celou frontu a smaz ji
		if (finishing_)
		{
			while (messages_.size() > 0)
			{
				delete messages_.front().msg;
				messages_.pop_front();
			}
		}
		else
		{
			msg = messages_.front();
			messages_.pop_front();
		}

		listLock_.up();

		// pokud jsme dostali nulovou zpravu, znamena to, ze koncime
		if (msg.msg == NULL)
		{
			break;
		}

        // process request
        gctx.eng.addRequest(msg.idUser, msg.msg);
	}
}

// ################################################################################################

sender :: sender(client *c) : workerSemaphore_(0, MAX_WORKER_QUEUE_ITEMS), listLock_(1, 1)
{
	client_ = c;
}

// ------------------------------------------------------------------------------------------------

void sender :: send(char *msg)
{
	listLock_.down();

	messages_.push_back(msg);

	listLock_.up();
	workerSemaphore_.up();
}

// ------------------------------------------------------------------------------------------------

void sender :: operator ()(threads::thread *thd)
{
	while (true)
	{
		// cekame na zpravu pomoci semaforu
		workerSemaphore_.down();

		// vyjmeme zpravu z fronty
		listLock_.down();

		char* msg = messages_.front();
		messages_.pop_front();

		listLock_.up();

		// jestlize jsme dorazili nakonec, vrat null, jinak posli zpravu
		if (msg == NULL)
			break;

		// posli zpravu
		char* toSend = msg;
		int msgLength = (int)strlen(toSend);
		while (msgLength > 0)
		{
			try
			{
				// zkus poslat zpravu
				int sended = client_->socket_->sendData(toSend, msgLength);
				msgLength -= sended;
				toSend += sended;
			}
			catch (std :: exception e)
			{
				// vyjimka, pouze prerus cyklus
				break;
			}
		}

		// smaz zpravu
		delete msg;
	}
}

// ------------------------------------------------------------------------------------------------