//  include headers for processes and sockets
#ifdef WIN32
	#include <winsock2.h>
	#include <Ws2tcpip.h>
	#include <process.h>
#else
	#include <pthread.h>
	#include <arpa/inet.h>
	#include <sys/types.h>
	#include <sys/socket.h>
	#include <netinet/in.h>
	#include <netdb.h>
	#include <stdio.h>
	#define SOCKET int
#endif

#include "string_socket.h"
#include "../common/util_classes.h"
#include <errno.h>

#ifdef WIN32
	#ifdef _DEBUG
	#define new DEBUG_NEW
	#undef THIS_FILE
	static char THIS_FILE[] = __FILE__;
	#endif
#endif





typedef struct tag_error_struct{     int     errnum;     char*   errstr; } ERROR_STRUCT;  
const char msg_no_error[19] =      "No error occurred!"; 
const char msg_unknown_error[14] = "Unknown error";  


const ERROR_STRUCT errlist[] = {     
	{EINTR,           "EINTR - Interrupted"},     
	{EBADF, "EBADF - Bad file number"},                
	{EFAULT,   "EFAULT - Bad address"},               
	{EINVAL,          "EINVAL - Invalid argument"},               
	{EMFILE,          "EMFILE - Too many open files"},            
};


bool CloseSocket(int l_socket)
{
	#ifdef WIN32
			return closesocket(l_socket) == 0;
	#else
			return close(l_socket) == 0;
	#endif
};


const int FLAGS = 0;

inline bool send_buffer(int client_fd, const char* Buffer,int Length, string& Error)
{
	

	int count;
	#ifdef WIN32
		count = send(client_fd, Buffer, Length, 0);
	#else
		count = send(client_fd, Buffer, Length, MSG_NOSIGNAL);
		/*if (count == -1)
		{
			char buffer[1000];
			int err = errno;
			int result = strerror_r(err, buffer, 1000);
			if (result == -1)
				Error =  Format( "unknown error code: %i", err);
			else
				Error =  buffer;
			return false;
		}*/
	#endif
	if (count == -1)
	{
		int err = errno;
		Error = Format("%s; errno=%i", strerror(err), err);;
		return false;
	}	

	Error = "";
	return true;
};



bool SendString (int client_fd, const char* Str, size_t  PacketLength, string& ErrorStr)
{
	try {
		if (!send_buffer(client_fd, (const char*)&PacketLength, sizeof(PacketLength), ErrorStr) )
		{
			return false;
		}

		const int buffer_len = 1000;
		if (PacketLength <= buffer_len)
		{
			if (!send_buffer(client_fd, Str, PacketLength, ErrorStr))
				return false;
		}
		else
		{
			int i = 0;
			for (; i< PacketLength/buffer_len;  i++)
			{
				if (!send_buffer(client_fd, (Str + i*buffer_len), buffer_len, ErrorStr))
				{
					return false;
				}
			};

			size_t rest=PacketLength % buffer_len;
			if (!send_buffer(client_fd, (Str + i*buffer_len), rest, ErrorStr))
			{
				return false;
			}
		};
		
		return true;
	}
	catch (...)
	{
		return false;
	};
};


inline NetworkErrorsEnum recieve_buffer(int client_fd, char* Buffer,int Length, int& have_read, int TimeOut )
{
	fd_set fds;
	struct timeval timeout;
	timeout.tv_sec = (TimeOut == -1) ? 1000: TimeOut;
	timeout.tv_usec = 0;
	

	FD_ZERO(&fds);
	FD_SET(client_fd,&fds);

	if (!select(client_fd+1,&fds, 0,0, &timeout)) return neTimeout;
	
	if (!FD_ISSET(client_fd,&fds)) return neNetworkError;
	errno = 0;

	#ifdef WIN32
		have_read = recv (client_fd, Buffer, Length, 0);
	#else
		have_read = recv (client_fd, Buffer, Length, MSG_NOSIGNAL);
	#endif

	//MSG_OOB


	if (have_read <= 0) 
	{
		
		return neNetworkError;
	}

	return neSuccess;
};



NetworkErrorsEnum RecieveString (int  client_fd, string& Result, int TimeOut)
{

	try {
		int PacketLength;
		int have_read;

		NetworkErrorsEnum Res = recieve_buffer (client_fd, (char*)&PacketLength, sizeof(PacketLength), have_read, TimeOut);
		if (Res != neSuccess )
		{
			return Res;
		}

		
		const int buffer_len = 1000;
		char buffer[buffer_len+1];

		Result = "";

		size_t  AllRead = 0;

		while (AllRead < PacketLength)
		{
			int to_read = buffer_len;
			if (PacketLength - AllRead < buffer_len)
					to_read = PacketLength - AllRead;

			
			NetworkErrorsEnum Res = recieve_buffer (client_fd, (char*)buffer, to_read, have_read, TimeOut);
			if (Res != neSuccess )
					return Res;

			AllRead += have_read;
			buffer[have_read] = 0;
			Result += buffer;
		};
		return neSuccess;
	}
	catch (...)
	{
		return neNetworkError;
	};
};

bool SetSocketRecieveBuffer(int socket_fd, int BufferSize)
{

	if (setsockopt( socket_fd, SOL_SOCKET, SO_RCVBUF, (const char*)&BufferSize, sizeof(int) ))
	{
		return false;
	};
	return true;
};

bool SetSocketSendBuffer(int socket_fd, int BufferSize)
{
	if (setsockopt( socket_fd, SOL_SOCKET, SO_SNDBUF, (const char*)&BufferSize, sizeof(int) ))
	{
		return false;
	};
	return true;
};

SOCKET create_socket (const char* HostName, int Port, bool bBind, string& strError)
{
	strError = "";
	struct addrinfo hints, *res, *res0; 
	int error;
	SOCKET l_socket;
	const char *cause = NULL; 

	memset(&hints, 0, sizeof(hints));
	if (bBind)
		hints.ai_flags = AI_PASSIVE;
	hints.ai_family = PF_UNSPEC;
	hints.ai_socktype = SOCK_STREAM;
	error = getaddrinfo(HostName, Format("%i",Port).c_str(), &hints, &res0); 
	if (error) {
		strError = gai_strerror(error); /*NOTREACHED*/
		return -1;
	}
	l_socket = -1;
	cause = "no addresses";
	//errno = EADDRNOTAVAIL;
	for (res = res0; res; res = res->ai_next) 
	{ 
		l_socket = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
		if (l_socket < 0) 
		{
			cause = "socket";
			continue;
		} 
		int Res;

		if (bBind)
			Res = bind(l_socket, res->ai_addr, res->ai_addrlen);
		else
			Res = connect(l_socket, res->ai_addr, res->ai_addrlen);

		

		if (Res < 0) 
		{
			#ifdef WIN32
				int error = WSAGetLastError();
			#endif
			cause = (bBind? "bind":"connect");
			CloseSocket(l_socket);
			l_socket = -1;
			continue;
		} 
		break; /* okay we got one */ 
	}
	if (l_socket < 0) 
	{
		strError = Format("Could not %s (Cause  = %s) for %s:%i\n",(bBind? "bind":"connect"), cause, HostName, Port);
		/*NOTREACHED*/
	}
	freeaddrinfo(res0); 
	return l_socket;

};

SOCKET create_and_bind (const char* HostName, int Port, string& strError)
{
	return create_socket(HostName, Port, true, strError);
};


SOCKET create_and_connect (const char* HostName, int Port, string& strError)
{
	return create_socket(HostName, Port, false, strError);
};


bool SocketInitialize(bool bReadFromLocalFile)
{
	#ifdef WIN32
		WORD    wVersionRequested;
		WSADATA wsaData;
		int     err;
		wVersionRequested = MAKEWORD(1,1);
		err = WSAStartup(wVersionRequested,&wsaData);
		if (err != 0)

		{
			//MessageBox(NULL, "Service could invoke WSAStartup()", "Concordance Socket", MB_OK);
			return false;
		}
	#endif
	
	
	return true;
};

bool SocketDeinitialize()
{
	#ifdef WIN32
		WSACleanup ();
	#endif

	return true;
};

string GetNetworkErrorString(const NetworkErrorsEnum& t)
{
	switch (t) {
		case neSuccess: return "No error";
		case neNetworkError: return "Network error";
		case neTimeout: return "Timeout elapsed";
		case neProtocolError: return "Different protocols are used";
		case neSubcorpusNotFound: return "Unknown subcorpus";
		case neInternalReadAccessError: return "Internal index access error";
	};
	return "Unknown error code";
};


CHost::CHost()
{
	m_bThreadIsRunning = false;
	m_LogFunction = 0;
	m_pdProtocolDensity = pdMediumDensity;
	m_bDoubleMessage = false;
};



bool CHost::ReadFromString (string S)
{
	StringTokenizer tok(S.c_str(), " \t\r\n");
	int ItemNo = 0;
	while (tok()) 
	{
			string Item = tok.val();
			if (ItemNo == 0)
				m_CorporaName = Item;
			else
			if (ItemNo == 1)
				m_HostName = Item;
			else
			if (ItemNo == 2)
			{
				m_Port = atoi(Item.c_str());
				m_LocalPath = tok.get_rest();
				Trim (m_LocalPath);
				break;
			};
			ItemNo++;
	};

	if (m_HostName.empty()) return false;

	return true;
};


bool LoadHosts (string Path, vector<CHost>& Hosts)
{
	if (!FileExists(Path.c_str()))
	{
		fprintf (stderr, "No access to %s in LoadHosts\n", Path.c_str());
		return false;
	};
	FILE * fp = fopen (Path.c_str(), "r");
	if (!fp) 
	{
		fprintf (stderr, "Could not open file %s\n", Path.c_str());
		return false;
	};
	char buffer[1000];
	int LineNo = 0;
	while (fgets(buffer, 1000, fp))
	{
		LineNo++;
		string Q = buffer;
		Trim(Q);
		if (Q.empty()) continue;
		if (Q[0] == '/') continue;
		CHost T;
		if (!T.ReadFromString(Q)) 
		{
			fprintf (stderr, "Could not read line No %i\n", LineNo);
			fclose(fp);
			return false;
		};
		
		Hosts.push_back(T);
	};
	
	fclose(fp);

	return true;
};

void AnswerByRequest (SOCKET rConnectedSocket, CHost* pHost, const sockaddr_in& client_addr, int TimeOut )
{
	string S;

	if (pHost->m_pdProtocolDensity  == pdHeavyDensity)
	{
		pHost->m_LogFunction(Format("%s recieves a string (Timeout=%i)", pHost->m_CorporaName.c_str(), TimeOut ));
	}

	
	NetworkErrorsEnum Res = RecieveString(rConnectedSocket, S, TimeOut);
							
	if (Res != neSuccess)
	{
			pHost->m_LogFunction(Format("Error: Cannot recieve string for corpus \"%s\" (error:%s)\n",pHost->m_CorporaName.c_str(), GetNetworkErrorString(Res).c_str()));
			return;
	};

	if (pHost->m_pdProtocolDensity  == pdHeavyDensity)
	{
		const char* src_addr = inet_ntoa(client_addr.sin_addr);
		pHost->m_LogFunction(Format("%s recv \"%s\" from %s", pHost->m_CorporaName.c_str(), S.c_str(),src_addr ));
	}

	string Result = pHost->ProcessSocketString(S, rConnectedSocket);

	if (pHost->m_pdProtocolDensity  == pdHeavyDensity)
	{
		pHost->m_LogFunction(Format("%s has finished processing", pHost->m_CorporaName.c_str() ));
	}

	string ErrorStr;
	if (!SendString(rConnectedSocket, Result.c_str(), Result.length(), ErrorStr))
	{
		pHost->m_LogFunction(Format("Error! Cannot send string: %s", ErrorStr.c_str()));
		return;
	};

	if (pHost->m_pdProtocolDensity  == pdHeavyDensity)
	{
		const char* src_addr = inet_ntoa(client_addr.sin_addr);
		pHost->m_LogFunction(Format("%s send back the result to %s (%i bytes)", pHost->m_CorporaName.c_str(), src_addr, Result.size() ));
	}
}

string	CHost::ProcessSocketString(const string& S, SOCKET rConnectedSocket) 
{
	assert(false);
	return "";
};

#ifdef WIN32
	void  _cdecl ListenerThread( void* pParam )
#else
	void*   ListenerThread( void* pParam )
#endif
{

	int listen_val;
	SOCKET T;
	size_t QueryCount = 0;
	CHost* pHost = (CHost* )pParam;
	string strError; 

	if (! SocketInitialize(false) )
	{
		pHost->m_LogFunction("Service could invoke AfxSocketInit()");
		goto exit_label;
	};

	pHost->InitHostData();

	T = create_and_bind(pHost->m_HostName.c_str(), pHost->m_Port, strError);
	if (T == -1)
	{
		pHost->m_LogFunction ( Format("Cannot create or bind socket for %s (Error = %s)\n", pHost->m_CorporaName.c_str(), strError.c_str()) );
		goto exit_label;
	};
	SetSocketRecieveBuffer (T, 100000);


	listen_val = listen(T, 5);
	if (listen_val == -1) {
		pHost->m_LogFunction ("Could not create the socket queue");
		goto exit_label;
	}


	pHost->m_LogFunction ("waiting for accept");

	for (;;)
	{
		QueryCount++;

		if (!(QueryCount%100))
			pHost->m_LogFunction (Format("QueryCount = %i", QueryCount));

		
		struct sockaddr_in client_addr;
		#ifdef WIN32
			int client_addr_len = sizeof(sockaddr_in);
		#else
			size_t client_addr_len = sizeof(sockaddr_in);
		#endif


		SOCKET rConnectedSocket = accept(T, (struct sockaddr *)&client_addr, &client_addr_len);


		
		if (rConnectedSocket == -1) 
		{
			pHost->m_LogFunction ("Error:Could not accept a  socket ");
			goto exit_label;
		};

		try
		{
			//  if we have passed "accept", then Timeout can be very short
			AnswerByRequest(rConnectedSocket, pHost, client_addr, 10);

			if (pHost->m_bDoubleMessage)
				AnswerByRequest(rConnectedSocket, pHost, client_addr, 30);

			if (!CloseSocket(rConnectedSocket))
				pHost->m_LogFunction ( Format("cannot close socket for %s", pHost->m_CorporaName.c_str())  );

			if (!pHost->m_bThreadIsRunning) break;
		}
		catch(...)
		{
			pHost->m_LogFunction("Error: An exception occurred");
			pHost->m_LogFunction("Closing accepting socket");
			if (!CloseSocket(rConnectedSocket))
				pHost->m_LogFunction ( Format("cannot close socket for %s", pHost->m_CorporaName.c_str())  );;
		};

			
	}

	pHost->m_LogFunction("exit from Listener");

exit_label:
		pHost->FreeHostData();
		SocketDeinitialize();
		#ifdef WIN32
			_endthread();
			return;
		#else
			return NULL;
		#endif

};

void CHost::CreateListener() 
{

	m_LogFunction( Format ("Create Listener %s (%s:%i)", m_CorporaName.c_str(), m_HostName.c_str(), m_Port) );
	#ifdef WIN32
		_beginthread( ListenerThread, 0, (void*)this );
	#else
		pthread_t aThread;
		pthread_create(&aThread, NULL, ListenerThread, (void*)this);
	#endif

	m_bThreadIsRunning = true;
};


void CHost::CopyFrom(const CHost& X)
{
	m_HostName = X.m_HostName;
	m_Port = X.m_Port;
	m_LocalPath = X.m_LocalPath;
	m_CorporaName = X.m_CorporaName;
	m_LogFunction = X.m_LogFunction;
	m_pdProtocolDensity = X.m_pdProtocolDensity;
};