/*******************************************************************************
 * Copyright (c) 2005, 2009 Intel Corporation, IBM.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * Contributors:
 *    Vishnu K Naikawadi, Intel - Initial API and Implementation (Based on the Java
 *                                API implementation by IBM)
 *    Spundun Bhatt (spundun@gmail.com), with the support and encouragement of the University of Southern California Information Sciences Institute Distributed Scalable Systems Division.
 *
 * $Id: TCPDataServer.cpp,v 1.29 2009/08/07 01:35:53 jwest Exp $
 *
 *******************************************************************************/ 



#include "TCPDataServer.h"


#include <stdio.h>
//#include <winsock2.h> 
#include <iostream>
#include <exception>
#include "tptp/TPTPCommon.h"

#include "tptp/TransportSupport.h"
#include "tptp/TPTPMessageHeader.h"
#include "tptp/TPTPSupportUtils.h"
#include "tptp/TPTPOSCalls.h"
#include "tptp/NoLog.h"


using namespace TPTP::Client;



const int TCPDataServer::BINARY_DATA = 0;
const int TCPDataServer::UTF8_STRING_DATA = 1;
const int TCPDataServer::UNICODE_STRING_DATA = 2;

void TCPDataServer::TCPDataProcessor::setSocket(SOCKET sock)
{
	_socket = sock;
}

void TCPDataServer::TCPDataProcessor::setDataProcessor(IDataProcessor* processor)
{
	_processor = processor;
}

void TCPDataServer::TCPDataProcessor::addDataProcessor(IDataProcessor* processor)
{	
	_processors.push_back(processor);
}

void TCPDataServer::TCPDataProcessor::removeDataProcessor(IDataProcessor* processor)
{
	vector<IDataProcessor*>::iterator processorIterator = _processors.begin();

	while (processorIterator != _processors.end())
	{
		if (*processorIterator == processor)
		{
			_processors.erase(processorIterator);
			break;
		}
	}

	return;
}


void TCPDataServer::TCPDataProcessor::resumeProcessing() 
{
	//synchronized(this) {
		_processing = true;
	//}
}

void TCPDataServer::TCPDataProcessor::pauseProcessing() 
{
	//synchronized(this) {
		_processing = false;
	//}
}

bool TCPDataServer::TCPDataProcessor::isProcessing() 
{
	//synchronized(this) {
		return _processing;
	//}
}

void TCPDataServer::TCPDataProcessor::shutdown() 
{
	//synchronized(this) {
		_shutdown = true;
	//}
}


void TCPDataServer::TCPDataProcessor::run() 
{
	while(!_shutdown)
	{
		int rc = 0;
		int  bufferLength = 1026 ;
		int  bytesRead = 0 ;
		char buffer[1026] ;
		char processBuffer[1026];

		do
		{
			if (_shutdown) break;
			
			rc = readFromSocket(_socket, buffer, bufferLength, &bytesRead);
			if (rc <= 0) Sleep(100);
		} while (rc <= 0);

		if (rc>0)
		{
			//this->_processor->incommingData(buffer, bytesRead, NULL);
			memcpy(processBuffer, buffer, bytesRead);			
			this->processData(processBuffer, 0, bytesRead);
		}
	}

		/* Run forever */
/*
outer:	while(!_shutdown || (_tcpServer->_currentFullBuffers != 0)) 
		{
			if(_tcpServer->isProcessing()) 
			{
				try 
				{
					bytesRead = 0;
					do
					{
						rc = readFromSocket(_socket, buffer, bufferLength, &bytesRead);
						if (rc <= 0) Sleep(100);
					} while (rc <= 0);
				}
				catch(exception e) 
				{
					// The server socket is toast, stop processing 
					pauseProcessing();
					continue;
				}

	
				while(true) 
				{
					// If all the buffers are full wait for the first one to be emptied 
					while(_tcpServer->_currentFullBuffers == NUM_BUFFERS) 
					{

						SLEEP(1000);
					}

					// Fill the next buffer 
					strcpy(_tcpServer->_bufferArray[_currentFillerBuffer]->data, buffer);
					_tcpServer->_bufferArray[_currentFillerBuffer]->length = bytesRead;


					if(_tcpServer->_bufferArray[_currentFillerBuffer]->length>0) 
					{
						// Move on to the next buffer 
						_currentFillerBuffer++;
						if(_currentFillerBuffer == NUM_BUFFERS) 
						{
							_currentFillerBuffer = 0;
						}
						_tcpServer->_currentFullBuffers++;

						// Is this the first buffer filled? 
						if(_tcpServer->_currentFullBuffers==1) 
						{
							//TODO: _tcpServer->_bufferArray[0].notifyAll();
						}
					}
				}
			}
			else {
				try 
				{
					// Monitoring is stopped, keep this thread in sleep state 
					SLEEP(1000); 
				} catch (exception e) 
				{

				}
			}
		}
*/
	//}

}

int TCPDataServer::TCPDataProcessor::loadMessageHeader(char data[], int offset, int limit) 
{
	/* Load all we can into the header */
	while (offset<limit && _currentHeaderOffset < Constants::DATA_MESSAGE_HEADER_LENGTH)
	{
		_messageHeader[_currentHeaderOffset++]=data[offset++];
	}
	if (_currentHeaderOffset == Constants::DATA_MESSAGE_HEADER_LENGTH)
	{
		MAKE_DIME_HEADER(_messageHeader);
		_dimeMessageHeader = (DIME_HEADER_PTR_T)_messageHeader;
	}	

	return offset;
}


int TCPDataServer::TCPDataProcessor::loadMessageHeaderDetails(char data[], int offset, int limit) 
{
	/* Load all we can into the header */
	while (offset<limit && _currentHeaderOffset < (int)(Constants::DATA_MESSAGE_HEADER_LENGTH+_dimeMessageHeader->id_length+_dimeMessageHeader->options_length+_dimeMessageHeader->type_length))
	{
		_messageHeader[_currentHeaderOffset++]=data[offset++];
	}
	if (_currentHeaderOffset == (int)(Constants::DATA_MESSAGE_HEADER_LENGTH+_dimeMessageHeader->id_length+_dimeMessageHeader->options_length+_dimeMessageHeader->type_length))
	{
		_dimeMessageHeader = (DIME_HEADER_PTR_T)_messageHeader;
	}	

	return offset;
}

long TCPDataServer::TCPDataProcessor::getMessageLength() 
{	
	//printf("The data length is - %d", _dimeMessageHeader->data_length);
	return _dimeMessageHeader->data_length;
}

int TCPDataServer::TCPDataProcessor::processData(char data[], int offset, int limit)
{
	long messageLength=0;
	int current;

	current=offset;

	/* Is there data to process */
	if(offset>=limit) {
		return limit;
	}

	/* Is this a new message? */
	if(_currentHeaderOffset < Constants::DATA_MESSAGE_HEADER_LENGTH) {
		/* Load the message header */
		current=this->loadMessageHeader(data, current, limit);

		/* Did we get the entire header, if not return */
		if(current==limit) {
			return current;
		}		
	}

	if(_currentHeaderOffset < (int)(Constants::DATA_MESSAGE_HEADER_LENGTH+_dimeMessageHeader->id_length+_dimeMessageHeader->options_length+_dimeMessageHeader->type_length)) {
		/* Load the message header */
		current=this->loadMessageHeaderDetails(data, current, limit);

		/* Did we get the entire header, if not return */
		if (current==limit && _currentHeaderOffset < (int)(sizeof(DIME_HEADER_T)+_dimeMessageHeader->id_length+_dimeMessageHeader->options_length+_dimeMessageHeader->type_length))
		{
			return current;
		}
		if(_dimeMessageHeader->data_length > 0 && current==limit) {
			return current;
		}		
	}

	//TODO - Check if the message is valid	

	/* How long is the current message */
	messageLength=getMessageLength();

	if (messageLength == 0)
	{
		//_processor->incomingData(_binaryForwardBuffer, 0, _dimeMessageHeader);
		for(int processorCount=0;processorCount < _processors.size(); processorCount++)
		{
			IDataProcessor* currentProcessor  = (IDataProcessor*)_processors[processorCount];
			currentProcessor->incomingData(_binaryForwardBuffer, _bytesWritten, _dimeMessageHeader);
		}
	}

	/* Process the entire buffer */
	while(current<limit) 
	{

		/* Copy as many bytes as possible into the forwarding buffer */
		while(current<limit && _bytesWritten<messageLength) {
			_binaryForwardBuffer[_bytesWritten++]=data[current++];
		}
		/* Are we at the end of the message? If so forward to the handler */
		if(_bytesWritten==messageLength) 
		{			
			//memcpy(sendBuffer, _dimeMessageHeader, Constants::DATA_MESSAGE_HEADER_LENGTH);
			//memcpy(sendBuffer+Constants::DATA_MESSAGE_HEADER_LENGTH, _binaryForwardBuffer, _bytesWritten);			
			//_processor->incomingData(sendBuffer, Constants::DATA_MESSAGE_HEADER_LENGTH+_bytesWritten, NULL);
			for(int processorCount=0;processorCount < _processors.size(); processorCount++)
			{
				IDataProcessor* currentProcessor  = (IDataProcessor*)_processors[processorCount];
				currentProcessor->incomingData(_binaryForwardBuffer, _bytesWritten, _dimeMessageHeader);
			}
			
			_bytesWritten=0;
			_currentHeaderOffset=0;
			/* Continue processing this data buffer */
			current=this->processData(data, current, limit);
		}	
	}

	return current;
}


THREAD_USER_FUNC_RET_TYPE TCPDataServer::TCPDataProcessor::startThread(void* pRequestBlock)
{
	TCPDataProcessor* tcpDataProcessor;

	if (pRequestBlock)
	{
		tcpDataProcessor = (TCPDataProcessor*)pRequestBlock;
		tcpDataProcessor->run();
		TPTP_LOG_DEBUG_MSG("The TCPDataProcessor thread started...");
	}

	return 0;
}







void TCPDataServer::BufferFlusher::setProcessor(IDataProcessor* processor)
{
	this->_processor = processor;
}



int TCPDataServer::BufferFlusher::loadMessageHeader(char data[], int offset, int limit) 
{
	/* Load all we can into the header */
	while(offset<limit && _currentHeaderOffset<(int)Constants::MESSAGE_HEADER_LENGTH) 
	{
		_messageHeader[_currentHeaderOffset++]=data[offset++];
	}
	return offset;
}

long TCPDataServer::BufferFlusher::getMessageLength() 
{
	//TODO: should use long
	unsigned int messageLength = 0;
	readUINTFromBuffer((unsigned char*)_messageHeader, &messageLength);
	return messageLength;
}


int TCPDataServer::BufferFlusher::getMessageType() 
{
	return _messageHeader[9];

}


bool TCPDataServer::BufferFlusher::checkMessageMagic() 
{
	//TODO: implement the Magic check
	return true;
}


int TCPDataServer::BufferFlusher::processData(char data[], int offset, int limit, char* addr)
{


	long messageLength;
	int current;

	current=offset;

	//char sendBuffer[MAX_MESSAGE_LENGTH];

	/* Is there data to process */
	if(offset>=limit) {
		return limit;
	}

	/* Is this a new message? */
	if(_currentHeaderOffset<Constants::DATA_MESSAGE_HEADER_LENGTH) {
		/* Load the message header */
		current=this->loadMessageHeader(data, current, limit);

		/* Did we get the entire header, if not return */
		if(current==limit) {
			return current;
		}


		/* Resize and compress the forward buffer if nessesary */
		if(getMessageLength() >= _currentBufferSize) 
		{						
			//int length = getMessageLength();
			char replacement[1024];
			/* Shift the available data to the front of the buffer */
			//TODO :
			//System.arraycopy(data, current, replacement, 0,(limit - current));
			_bytesWritten=limit-current;
			strcpy(_binaryForwardBuffer, replacement) ;
			return limit;
		}
	}

	/* Validate the message header, if we are in recovery
	   mode try and look at the next offset */
	if(!checkMessageMagic()) {
		TPTP_LOG_ERROR_MSG("Corrupt data");
		_currentHeaderOffset=0;
		return processData(data, offset+1, limit, addr);

	}


	/* How long is the current message */
	messageLength=getMessageLength();

	/* Process the entire buffer */
	while(current<limit) 
	{

		/* Copy as many bytes as possible into the forwarding buffer */
		while(current<limit && _bytesWritten<messageLength) {
			_binaryForwardBuffer[_bytesWritten++]=data[current++];
		}
		/* Are we at the end of the message? If so forward to the handler */
		if(_bytesWritten==messageLength) {			
			//memcpy(sendBuffer, _messageHeader, Constants::DATA_MESSAGE_HEADER_LENGTH);
			//memcpy(sendBuffer+Constants::DATA_MESSAGE_HEADER_LENGTH, _binaryForwardBuffer, _bytesWritten);
			//_processor->incomingData(sendBuffer, _bytesWritten, _dimeMessageHeader);
			_bytesWritten=0;
			_currentHeaderOffset=0;
			/* Continue processing this data buffer */
			current=this->processData(data, current, limit, addr);
		}	
	}

	return current;

}



void TCPDataServer::BufferFlusher::run()
{

outer:	while(_tcpServer->isProcessing() || (_tcpServer->_currentFullBuffers != 0)) 
		{ 
			/* 237169 make sure buffer is empty before exiting */
			/* If there are no current buffers to empty wait */
			if(_tcpServer->_currentFullBuffers==0) {
				_processor->waitingForData();
				do {
					//synchronized(_bufferArray[0]) {
						try {
							//_bufferArray[0]->wait(1000);
							SLEEP(1000);
						}
						catch(exception e) {
						}
					//}
					if(!_tcpServer->isProcessing() && _tcpServer->_currentFullBuffers==0) {
						goto outer;
					}
				}
				while(_tcpServer->_currentFullBuffers==0);
			}

			/* Empty the current buffer */
			if(_tcpServer->_bufferArray[_currentFlusherBuffer]->length>0) {
				TPTP_LOG_DEBUG_MSG1("Flushing --- %d bytes", _tcpServer->_bufferArray[_currentFlusherBuffer]->length);
				processData(_tcpServer->_bufferArray[_currentFlusherBuffer]->data,
							0,
							_tcpServer->_bufferArray[_currentFlusherBuffer]->length,
							_tcpServer->_bufferArray[_currentFlusherBuffer]->addr);
				/* Mark the buffer as empty */
				_tcpServer->_bufferArray[_currentFlusherBuffer]->length=0;

			}

			//synchronized(_bufferArray[0]) {

				_tcpServer->_currentFullBuffers--;

				/* Increment the flusher to the next buffer */
				_currentFlusherBuffer++;
				if(_currentFlusherBuffer==NUM_BUFFERS) {
					_currentFlusherBuffer=0;
				}

				/* If the buffers were half full before this flush notify the
				   filler it can continue. Generally this could be NUMBUFFERS
				   but not as efficient as more thread switches happen
				*/
				if(_tcpServer->_currentFullBuffers==0) {
					// TODO ?
					//_bufferArray[0].notifyAll();
				}
			//}
		}


		/* Notify that the flusher is exiting */
		//
		//if(_processor instanceof DataServerListener) {
		//	((DataServerListener)_processor).dataServerExited();
		//}

}



THREAD_USER_FUNC_RET_TYPE TCPDataServer::BufferFlusher::startThread(void* pRequestBlock)
{
	BufferFlusher* bufferFlusher;

	if (pRequestBlock)
	{
		bufferFlusher = (BufferFlusher*)pRequestBlock;
		(bufferFlusher)->run();
		TPTP_LOG_DEBUG_MSG("The BufferFlusher thread started");
	}

	return 0;
}







TCPDataServer::TCPDataServer()
{
	
}

TCPDataServer::~TCPDataServer()
{

}






int TCPDataServer::getPort() 
{
	return _port;
}

/*
char* TCPDataServer::getServerAddress() 
{
	return this->_sock;
}
*/




/**
 *********************************************************
 *                                                       
 * @brief                                                
 *    Given a hostname, get the current host IP address. An array of sockaddr_storages are stored in outResult, and the size
 *	  of that array is stored in outNumResults.
 * @return
 *    the IP address number
 *
 *********************************************************/
/*
void TCPDataServer::getIPAddresses(char *hostname, struct sockaddr_storage *** outResult, int *outNumResults) {
	int RetVal;
	int i;

	struct addrinfo Hints, *AddrInfo, *AI;

    int Family = PF_UNSPEC; // Give us both the IPv4 and IPv6 addresses
    int SocketType = SOCK_STREAM; // TCP only

	struct sockaddr_storage ** result;

	int    rc = 0 ;

	rc = initForSocketCalls();
	if(rc != 0) {
		*outNumResults = 0;
	}
	
	result = (struct sockaddr_storage **)tptp_malloc(sizeof(struct sockaddr_storage *) * 128);
	
	memset(&Hints, 0, sizeof (Hints));
    Hints.ai_family = Family;
    Hints.ai_socktype = SocketType;
	
	// This was originally in the code, but it prevents lookups of host names, e.g. www.google.com. It only works with numeric hosts.
    // Hints.ai_flags = AI_NUMERICHOST;  

    RetVal = getaddrinfo(hostname, NULL, &Hints, &AddrInfo);

	for (i = 0, AI = AddrInfo; AI != NULL; AI = AI->ai_next, i++) {
		result[i] = cloneSockAddr((struct sockaddr_storage *)AI->ai_addr);
	}
	freeaddrinfo(AddrInfo);

	*outResult = result;
	*outNumResults = i;

	return RetVal;
}*/


/**
 *********************************************************
 *                                                       
 * @brief                                                
 *    get the current localhost IP address
 *
 * @return
 *    the IP address number
 *
 *********************************************************/
/*
unsigned long TCPDataServer::getLocalIPAddress() {
	struct hostent *hpLocal = NULL;
	unsigned long   result = 0 ;
	int    rc = 0 ;

	rc = initForSocketCalls() ;
	if (rc == 0)
	{
		hpLocal = getHostInfoIPv4M() ;

		if (hpLocal != NULL)
			memcpy(&result, hpLocal->h_addr_list[0], sizeof(long));
		else
			result = -1;
	}

	return result;
}
*/

#define MAX_HOST_NAME_LENGTH	128

void TCPDataServer::connect()
{
	int rc = 0;
	char buffer[1024] ;
	int  bufferLength = 1024 ;
	int  bytesRead = 0 ;
	
	char hostname[MAX_HOST_NAME_LENGTH];
	
	initForSocketCalls();

	// Get the local host informantion 
	if (gethostname(hostname,MAX_HOST_NAME_LENGTH) < 0) {
		TPTP_LOG_DEBUG_MSG("Error: Unable to resolve local host name.");
		return;
	}

	/* go make connection */
	rc = connectToTCPServer(hostname, _port, &_sock) ;

	if (rc != 0)
	{
		TPTP_LOG_DEBUG_MSG("Error: unable to connect to the server.");
	}
	else
	{
		if (rc > 0)
		{
			/* pause a little to receive the reply */
			Sleep(1000) ;

			//init();

			do
			{
				rc = readFromSocket(_sock, buffer, bufferLength, &bytesRead);
				if (rc==0)Sleep(100);
			}
			while (rc==0);
			if (rc >= 0)
			{
				tptp_basic_msg_header_ptr_t pMsg = (tptp_basic_msg_header_ptr_t) buffer ;
				TPTP_LOG_DEBUG_MSG1("TCPDataServer::connect() read %d bytes from socket.", bytesRead) ;
				printThisEnvelope(pMsg) ;
				//_connectionId = getConnectionId() ;
			}
		}	
	}

}


void TCPDataServer::startServer(IDataProcessor* processor, int sockid)
{
		for(int i=0; i<NUM_BUFFERS; i++)
		{
			_bufferArray[i]=new SingleBuffer();
		}

		//_port = port;
		//this->connect();

		_sock = sockid;

		_server=new TCPDataProcessor(this);
		_server->setSocket(_sock);
		_server->addDataProcessor(processor);
		// Set the thread name - TODO
		//_server->setName("TCPDataFiller");

		/* Set up the data flusher */
		//_flusher=new BufferFlusher(this);
		//_flusher->setProcessor(processor);
		// Set the thread name - TODO
		//_flusher->setName("TCPDataFlusher");

		//Start the threads
		HANDLE th1;
		TID tid1;
		tptpStartThread(TCPDataServer::TCPDataProcessor::startThread, (void *) _server, &tid1, &th1);
		CLOSE_THREAD_HANDLE(th1);

		//HANDLE th2;
		//DWORD tid2;
		//rc = tptpStartThread(TCPDataServer::BufferFlusher::startThread, (LPVOID) _flusher, (unsigned *)&tid2, &th2);
}


void TCPDataServer::startServer(int sockid)
{

	_sock = sockid;

	_server=new TCPDataProcessor(this);
	_server->setSocket(_sock);

	/* Set up the data flusher */
	_flusher=new BufferFlusher(this);

	//Start the threads
	HANDLE th1;
	TID tid1;
	tptpStartThread(TCPDataServer::TCPDataProcessor::startThread, (void *) _server, &tid1, &th1);

	CLOSE_THREAD_HANDLE(th1);
}


bool TCPDataServer::isProcessing() 
{
	return _server->isProcessing();
}


void TCPDataServer::addDataProcessor(IDataProcessor* processor)
{
	_server->addDataProcessor(processor);

}

void TCPDataServer::removeDataProcessor(IDataProcessor* processor)
{
	_server->removeDataProcessor(processor);

}


void TCPDataServer::stopServer() 
{
	_server->pauseProcessing();
}


void TCPDataServer::resumeServer() 
{
	_server->resumeProcessing();
}



void TCPDataServer::shutdownServer() 
{
	_server->shutdown();
}


