/*
*	Base server class that is listening on a socket.
*
*	©Copyright 2002-2004, Microsoft Corporation
*
*	Written by Sergey Vasilevskiy.
*/

#include "stdafx.h"

using namespace TPCE;

// Explicit template instantiation to generate code
template class CBaseServer<TCEServerLocals>;
template class CBaseServer<TMEEServerLocals>;

//Constructor
template <typename LocalVarsType>
CBaseServer<LocalVarsType>::CBaseServer(char *szServer, unsigned short shPort, int iMsgSize, 
										HANDLE hStopListen, bool bVerbose)
: m_server(NULL)
, m_hExitEvent(NULL)
, m_hProcessThreadExited(NULL)
, m_NumThreadsCreated(0)
, m_bRunProcessing(1)
, m_hStopListen(hStopListen)
, m_bVerbose(bVerbose)
, m_iDeadlockCount(0)
, m_OutputLock()
{	
	int			ret;
	int			iSendBufSize = 1024;	// socket send buffer size
	int			iRecvBufSize = 4096;	// socket receive buffer size

	m_sockfd = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, NULL, 0);
	if (m_sockfd == INVALID_SOCKET)
		ThrowError("WSASocket()");

	// Set buffer sizes for the socket (OS default is 8K).
	ret = setsockopt(m_sockfd, SOL_SOCKET, SO_SNDBUF, (char*)&iSendBufSize, sizeof(iSendBufSize));
	if (ret == SOCKET_ERROR)
		ThrowError("setsockopt(SO_SNDBUF)");

	ret = setsockopt(m_sockfd, SOL_SOCKET, SO_RCVBUF, (char*)&iRecvBufSize, sizeof(iRecvBufSize));
	if (ret == SOCKET_ERROR)
		ThrowError("setsockopt(SO_RCVBUF)");

	m_shPort = shPort;
	m_server = new char[strlen(szServer)+1];
	if (m_server != NULL)
		strcpy(m_server, szServer);

	//Create the WSA Event object
	if ( (m_hWSAEvent = WSACreateEvent()) == WSA_INVALID_EVENT)
	{
		ThrowError("WSACreateEvent");
	}

	if (bVerbose)
	{
		cout<<"WSA event created."<<endl;
	}
	//Register for the accept available event
	if ( WSAEventSelect(m_sockfd, m_hWSAEvent, FD_ACCEPT) == SOCKET_ERROR)
	{
		ThrowError("WSAEventSelect");
	}

	m_hWaitListenEvents[0] = m_hWSAEvent;
	m_hWaitListenEvents[1] = m_hStopListen;	

	m_iMsgSize = iMsgSize;

	OutputStdin("Max message size = %d\n", m_iMsgSize);

	if ( (m_hExitEvent=CreateEvent(NULL, TRUE, FALSE, NULL)) == NULL )	//manual reset, nonsignaled
		throw new CSystemErr(CSystemErr::eCreateEvent, "CBaseServer");
	
	if ( (m_hProcessThreadExited=CreateEvent(NULL, FALSE, TRUE, NULL)) == NULL )	//auto reset, signaled
		throw new CSystemErr(CSystemErr::eCreateEvent, "CBaseServer");
}

//Destructor
template <typename LocalVarsType>
CBaseServer<LocalVarsType>::~CBaseServer()
{
	if (m_hExitEvent != NULL)
		SetEvent(m_hExitEvent);

	if (m_server != NULL)
		delete m_server;

	WSACleanup();

	while (m_NumThreadsCreated > 0)	//should be protected probably
	{
		WaitForSingleObject(m_hProcessThreadExited, WSA_INFINITE);
		//--m_NumThreadsCreated;	//wait for all created threads to exit
	}

	if (m_sockfd != INVALID_SOCKET)
		closesocket(m_sockfd);

	//Close the events
	if (m_hExitEvent != NULL)
		CloseHandle(m_hExitEvent);

	if (m_hProcessThreadExited != NULL)
		CloseHandle(m_hProcessThreadExited);
}

//Throw a socket exception.
template <typename LocalVarsType>
void CBaseServer<LocalVarsType>::ThrowError(char *szText)
{
	throw new CSystemErr(WSAGetLastError(), CSystemErr::eNone, szText);
}

//Thread that listens on the port, accepts a new connection, and creates a new thread
//that handles that connection.
//The input parameter is the thread exit event.
template <typename LocalVarsType>
void CBaseServer<LocalVarsType>::ListenAndDispatchThread(void *ptr)
{	
	SOCKADDR_IN								addr, acc_addr;
	SOCKET									acc_sockfd;	//socket for accepted connections
	int										acc_addr_size;
	int										ret;
	uintptr_t								nret;
	TRunProcessThreadInput<LocalVarsType>*	pRunProcessInput;
	CDateTime								TimeNow;

	ResetEvent(m_hProcessThreadExited);		//indicate that we're running

	//zero the struct
	memset(&addr, 0, sizeof(addr));
	addr.sin_addr.S_un.S_addr = ADDR_ANY;
	addr.sin_family = AF_INET;
	//addr.sin_port = 0;	//let system assign a port
	addr.sin_port = htons(m_shPort);	//assign the port ourselves

	try 
	{
		//bind socket to a port
		ret = bind(m_sockfd, (const sockaddr*)&addr, sizeof(addr));
		if (ret == SOCKET_ERROR)
			ThrowError("bind()");

		//start listening
		ret = listen(m_sockfd, SOMAXCONN);
		if (ret != 0)
			ThrowError("listen()");

		OutputStdinTimed("Started listening on port %d...\n", m_shPort);

		while (true)
		{

			if ( (ret = WSAWaitForMultipleEvents(sizeof(m_hWaitListenEvents)/sizeof(m_hWaitListenEvents[0]), 
											&m_hWaitListenEvents[0], FALSE, WSA_INFINITE, TRUE)) == WSA_WAIT_FAILED)
				ThrowError("WSAWaitForMultipleEvents");

			WSAResetEvent(m_hWSAEvent);	//reset to allow waiting for the accept

			//test whether got the Stop event
			if (ret == WSA_WAIT_EVENT_0 + 1)
			{				
				break;	//exit the thread
			}

			acc_addr_size = sizeof(acc_addr);
			// accept will signal m_hWSAEvent if there are connections in the backlog
			acc_sockfd = accept(m_sockfd, (sockaddr*)&acc_addr, &acc_addr_size);
			if (acc_sockfd == INVALID_SOCKET)
				ThrowError("accept()");			

			OutputStdinTimed("New connection from %u.%u.%u.%u (total = %d)\n", 
							(UINT)acc_addr.sin_addr.S_un.S_un_b.s_b1,
							(UINT)acc_addr.sin_addr.S_un.S_un_b.s_b2,
							(UINT)acc_addr.sin_addr.S_un.S_un_b.s_b3,
							(UINT)acc_addr.sin_addr.S_un.S_un_b.s_b4,
							m_NumThreadsCreated + 1
							);

			//prepare parameters for the new thread
			pRunProcessInput = new TRunProcessThreadInput<LocalVarsType>;
			if (pRunProcessInput == NULL)
				throw new CSystemErr(CSystemErr::eNew, "CBaseServer::ListenAndDispatchThread (new)");
			pRunProcessInput->obj = this;	//run ProcessingThread on this object class
			pRunProcessInput->ProcessingInput.hSocket = acc_sockfd;
			pRunProcessInput->ProcessingInput.iThreadNum = m_NumThreadsCreated;			

			nret = _beginthread(RunProcessingThread<LocalVarsType>, 0, pRunProcessInput);
			if (nret == -1)
			{
				throw new CSystemErr(CSystemErr::ebeginthread, "CBaseServer::ListenAndDispatchThread (beginthread)");
				return;
			}

			//increment the number of threads created so know for how many to wait on destruction
			InterlockedIncrement(&m_NumThreadsCreated);
			//++m_NumThreadsCreated;	

		}
	}
	catch (CBaseErr *pErr)
	{
		OutputStdinTimed("Exception in ListenAndDispatch thread (error %d, location %s)\n", 
					pErr->ErrorNum(), pErr->ErrorLoc());
		//throw;	//rethrow the exception
		if (pErr != NULL)
			delete pErr;	//cleanup the thrown exception
	}

	//TimeNow.SetToCurrent();
	TimeNow.Set();
	
	OutputStdinTimed("Listening thread exited\n");
}

//
//Thread that reads a message from the network and processes it.
//
template <typename LocalVarsType>
void CBaseServer<LocalVarsType>::ProcessingThread(void *ptr)
{
	PProcessThreadInput	pInput = (PProcessThreadInput)ptr;
	SOCKET				sockfd = pInput->hSocket;	//connected socket passed in as the parameter
	int					iThreadNum = pInput->iThreadNum;
	char*				message_buffer = NULL;		//network input buffer
	WSABUF				RecvBuffer, SendBuffer;
	int					waitret, ret;
	WSAEVENT			hWaitProcessEvents[2] = { WSA_INVALID_EVENT, WSA_INVALID_EVENT };
	DWORD				dwNumberOfBytesRecvd;
	DWORD				dwNumberOfBytesSent;
	DWORD				Flags = 0;	//WSARecv flags
	WSANETWORKEVENTS	sNetworkEvents;
	LocalVarsType*		pLocal = NULL;	//implementation-specific local thread variables
	char				szMsg[2*1024];
	int					iRetStatus;
	DWORD				iTotalBytesRecvd;
	fd_set				fdSet;
	WSAOVERLAPPED		sOverlappedSend;	// for WSASend
	CDateTime			TimeNow;
	TSUTReplyMessage	SUTReply;	// structure sent to the driver after every transaction

	try
	{		
		pLocal = InitializeLocal();	//initialize local variables - should be defined in subclasses

		message_buffer = (char*)malloc(m_iMsgSize);	//allocate network input buffer

		//Set up WSABUF structs		
		SendBuffer.buf = (char *)&SUTReply;
		SendBuffer.len = sizeof(SUTReply);
		//for select
		FD_ZERO(&fdSet);
		FD_SET(sockfd, &fdSet);


		//Create the WSA Event object
		//if ( (hWaitProcessEvents[0] = WSACreateEvent()) == WSA_INVALID_EVENT)
		//	ThrowError("WSACreateEvent");
		if ( (hWaitProcessEvents[0] = CreateEvent(NULL, FALSE, FALSE, NULL)) == NULL)
			ThrowError("CreateEvent");

		//Register for network read event and for close connection event
		if ( (ret = WSAEventSelect(sockfd, hWaitProcessEvents[0], FD_READ | FD_CLOSE)) == SOCKET_ERROR)
			ThrowError("WSAEventSelect");

		//second event is the Stop event
		hWaitProcessEvents[1] = m_hStopListen;

		//set the event signaled on WSASend
		sOverlappedSend.hEvent = hWaitProcessEvents[0];

		while(TRUE)
		{
			// Make sure that there is a green light on processing requests
			//
			if (!m_bRunProcessing)
			{				
				// Wait for exit event set by the user.
				//
				WaitForSingleObject(m_hStopListen, INFINITE);
				//StopListen event => exit the thread
				break;
			}

			iTotalBytesRecvd = 0;
			RecvBuffer.buf = message_buffer;
			RecvBuffer.len = m_iMsgSize;
			//WSAResetEvent(hWaitProcessEvents[0]);	//did the read => reset the read notification event

			do
			{
				//Wait on network read available
				//OutputStdin("waiting for read event...");
				if ( (waitret = WSAWaitForMultipleEvents(2, &hWaitProcessEvents[0], FALSE, WSA_INFINITE, TRUE)) == WSA_WAIT_FAILED)
					ThrowError("WSAWaitForMultipleEvents");

				if (waitret == WSA_WAIT_EVENT_0 + 1)
				{
					//exit the thread
					break;
				}

				//test what events were signaled
				ret = WSAEnumNetworkEvents(sockfd, hWaitProcessEvents[0], &sNetworkEvents);
				if (ret == SOCKET_ERROR)
					ThrowError("WSAEnumNetworkEvents");

				if ((sNetworkEvents.lNetworkEvents & FD_CLOSE) != 0)
				{	//connection close event					
					break;
				}			

				ret = select(0, &fdSet, NULL, NULL, NULL);
				if (ret == SOCKET_ERROR)
					ThrowError("select");

				ret = WSARecv(sockfd, &RecvBuffer, 1, &dwNumberOfBytesRecvd, &Flags, NULL, NULL);
				if (ret != 0)
					ThrowError("WSARecv");
				
				iTotalBytesRecvd += dwNumberOfBytesRecvd;
				RecvBuffer.buf += dwNumberOfBytesRecvd;		//advance buffer start so don't overwrite recv'd data
				RecvBuffer.len -= dwNumberOfBytesRecvd;		//decrease the available buffer length
				
				if (m_bVerbose)
				{
					_snprintf(szMsg, sizeof(szMsg), "Thread %i: new request received (%d bytes)\n", 
							iThreadNum, dwNumberOfBytesRecvd);
					OutputStdinTimed(szMsg);
				}

				iRetStatus = ProcessMessage(message_buffer, iTotalBytesRecvd, pLocal, iThreadNum, &SUTReply);
											
			} while (iRetStatus == PARTIAL_MESSAGE);	//read the whole message			

			if (waitret == WSA_WAIT_EVENT_0 + 1)
			{
				//exit the thread
				break;
			}
			if ((sNetworkEvents.lNetworkEvents & FD_CLOSE) != 0)
			{	//connection close event

				if (m_bVerbose)
				{
					_snprintf(szMsg, sizeof(szMsg), "connection closed, thread %i exiting\n", 
						iThreadNum);

					OutputStdinTimed(szMsg);
				}
				break;
			}
			
			if (iRetStatus != SEND_NO_ACK)	//send only if ProcessMessage has not instructed otherwise
			{
                SendBuffer.len = TxnRespSize[SUTReply.eTxnType];    //  Send only the needed bytes
				ret = WSASend(sockfd, &SendBuffer, 1, &dwNumberOfBytesSent, 0, NULL, NULL);
				if (ret == SOCKET_ERROR)
				{	
					ThrowError("WSASend");
				}
			}	// if (iRetStatus != SEND_NO_ACK)
			//OutputStdin("Processing thread: sent %d bytes\n", dwNumberOfBytesSent);
		}	// while(TRUE)
					
	}
	catch (CBaseErr *pErr)
	{
		if (pErr->ErrorType() == ERR_TYPE_ODBC)
		{
			m_bRunProcessing = 0;	//pause all the other processing threads (easy deadlock debugging)
		}
		
		//char szMsg[2*1024];
		int cbRet;
		
		//TimeNow.SetToCurrent();	// capture current time for error output
		TimeNow.Set();	// capture current time for error output

		cbRet = _snprintf(szMsg, sizeof(szMsg), "%s ",TimeNow.ToStr(iDateTimeOutputStyle));
		if (pErr->ErrorType() == ERR_TYPE_ODBC)
		{
			cbRet += _snprintf(&szMsg[cbRet], sizeof(szMsg)-cbRet, "Processing stopped for all connections");
		}
		cbRet += _snprintf(&szMsg[cbRet], sizeof(szMsg)-cbRet, "\r\n%s", pErr->ErrorText());
		
		if (pErr->ErrorNum() != 0)
		{
			cbRet += _snprintf(&szMsg[cbRet], sizeof(szMsg)-cbRet, " error %d", pErr->ErrorNum());
		}

		// Add Action output for ODBC errors
		if (pErr->ErrorType() == ERR_TYPE_ODBC)
		{
			if (((CODBCERR*)pErr)->m_eAction != CODBCERR::eNone)
			{
				cbRet += _snprintf(&szMsg[cbRet], sizeof(szMsg)-cbRet, "\r\nAction: %i", 
					((CODBCERR*)pErr)->m_eAction);
			}
		}

		// add location if it exists
		if (pErr->ErrorLoc() != NULL)
		{
			cbRet += _snprintf(&szMsg[cbRet], sizeof(szMsg)-cbRet, " in %s", pErr->ErrorLoc());				
		}
		_snprintf(&szMsg[cbRet], sizeof(szMsg)-cbRet, "\n");
		szMsg[sizeof(szMsg)-1] = '\0';	//null-terminate if sprintf'ed on the last byte in szMsg

		OutputStdin(szMsg);

		if (pErr != NULL)
			delete pErr;	//cleanup the thrown exception
	}

	//cleanup
	if (message_buffer != NULL)
		free(message_buffer);
	if (pLocal != NULL)
		DeInitializeLocal(pLocal);
	closesocket(sockfd);
	if ( hWaitProcessEvents[0] != WSA_INVALID_EVENT )
		CloseHandle(hWaitProcessEvents[0]);

	InterlockedDecrement(&m_NumThreadsCreated);
	SetEvent(m_hProcessThreadExited);		//set on exit

	//TimeNow.SetToCurrent();
	TimeNow.Set();

	_snprintf(szMsg, sizeof(szMsg), "%s Processing thread %i exited.\n", 
			TimeNow.ToStr(iDateTimeOutputStyle), iThreadNum);
	OutputStdin(szMsg);
}

template <typename LocalVarsType>
LocalVarsType* CBaseServer<LocalVarsType>::InitializeLocal()
{
	//This function should be defined in subclasses.
	return NULL;
}

template <typename LocalVarsType>
void CBaseServer<LocalVarsType>::DeInitializeLocal(LocalVarsType* pLocal)
{
	//This function should be defined in subclasses.
	delete pLocal;
}


int WSARecvNBytes(
  SOCKET s,
  LPWSABUF lpBuffer,
  LPDWORD lpNumberOfBytesRecvd,
  LPDWORD lpFlags,
  DWORD	dwBytesToRecv	//block until this number of bytes has been received  
)
{
	int ret;
	DWORD dwBytesRecvd, dwTotalBytesRecvd = 0;
	char *szBuf = lpBuffer->buf;
	unsigned long lLen = lpBuffer->len;

	while (dwTotalBytesRecvd < dwBytesToRecv)
	{
		//OutputStdin("Receiving %d\n"), lpBuffer->len);
		ret = WSARecv(s, lpBuffer, 1 /*always*/, &dwBytesRecvd, lpFlags, NULL, NULL);
		dwTotalBytesRecvd += dwBytesRecvd;
		//OutputStdin("Recv'd %d, needed %d\n", dwBytesRecvd, dwBytesToRecv);

		if (ret == SOCKET_ERROR)
		{
			break;
		}
		else
		{
			if (dwBytesRecvd < lpBuffer->len)
			{
				lpBuffer->buf += dwBytesRecvd;
				lpBuffer->len -= dwBytesRecvd;
			}
		}
	}

	*lpNumberOfBytesRecvd = dwTotalBytesRecvd;	//output the total number of bytes recv'd
	lpBuffer->buf = szBuf;	//restore
	lpBuffer->len = lLen;

	return ret;
}

template <typename LocalVarsType>
void CBaseServer<LocalVarsType>::OutputStdin(char *szFmt, ...)
{
	int			len;	

	m_OutputLock.lock();

	va_list marker;
	va_start( marker, szFmt );
	len = vprintf(szFmt, marker);
	fflush(stdin);
	va_end( marker );	

	m_OutputLock.unlock();
}

// Same as OutputStdin, except prefixes the output with the current timestamp.
//
template <typename LocalVarsType>
void CBaseServer<LocalVarsType>::OutputStdinTimed(char *szFmt, ...)
{
	int			len;	

	m_OutputLock.lock();

	// Print the current timestamp.
	//
	CDateTime TimeNow;
	printf("%s ", TimeNow.ToStr(iDateTimeOutputStyle));
	
	// Output the rest.
	//
	va_list marker;
	va_start( marker, szFmt );
	len = vprintf(szFmt, marker);
	fflush(stdin);
	va_end( marker );	

	m_OutputLock.unlock();
}

