/*******************************************************************************
 * 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: ConnectionImpl.cpp,v 1.43 2009/09/11 19:16:13 jwest Exp $
 *
 *******************************************************************************/ 


#include "ConnectionImpl.h"
#include "tptp/TransportSupport.h"
#include "tptp/TPTPMessageHeader.h"
#include "tptp/TPTPSupportUtils.h"
#include "tptp/client/Constants.h"
#include "tptp/TPTPSupportTypes.h"
#include "tptp/client/INode.h"
#include "tptp/TPTPOSCalls.h"
#include "tptp/NoLog.h"
#include "tptp/TPTPErrorCode.h"

#include <stdlib.h>
#include <stdio.h>
#include <iostream>

using namespace std;
using namespace TPTP::Client;

#ifdef _WIN32
// TODO It's a workaround to get compiled on WIN IA32, it needs to be resolved.
	#define _WSPIAPI_COUNTOF(_Array) (sizeof(_Array) / sizeof(_Array[0]))
	#include <Ws2tcpip.h>
#endif

ConnectionImpl::ConnectionImpl()
{
	sock = -1;
	_connectionId = -1;
	_port = -1;
	_contextId = 1;

	_node = NULL;
	_contextMapper = NULL;
	_cmdHandler = NULL;

	_isComplete = false;
	_isInitialized = false;;
	_loginPending = false;
}


ConnectionImpl::~ConnectionImpl()
{
	
}

ConnectionImpl::ConnectionHandler::ConnectionHandler()
{
	_mapper = NULL;
}


ConnectionImpl::ConnectionHandler::ConnectionHandler(ContextMapper* ctxMapper)
{
	_mapper = ctxMapper;
}


ConnectionImpl::ConnectionHandler::~ConnectionHandler()
{


}



void ConnectionImpl::ConnectionHandler::incomingCommand(INode* node, CommandElement* command) 
{
	
	long contextId=command->getContext();

	TPTP_LOG_DEBUG_MSG1("The context of the returned command: %ld", contextId);


	// Find the command handler associated with this contextId and
	//   forward the message appropriately.
	ICommandHandler* ch = _mapper->getHandler(contextId);
	if(ch != NULL) {
		TPTP_LOG_DEBUG_MSG("Forwarding to command handler");
		ch->incomingCommand(node, command);
	}
	else {
		TPTP_LOG_DEBUG_MSG("Could not find command handler");
	}
	
}

/* Init */
void ConnectionImpl::init()
{

	_contextMapper=new ContextMapper();
	_cmdHandler = new ConnectionImpl::ConnectionHandler(_contextMapper);
	
	HANDLE th;
	TID tid;

	reader = new SocketReaderThread(this);
	tptpStartThread(ConnectionImpl::SocketReaderThread::startThread, (void *) &reader, &tid, &th);
	CLOSE_THREAD_HANDLE(th);

}


long ConnectionImpl::getNextContextID()
{
	return this->_contextId++;
}

void ConnectionImpl::addContext(long id, ICommandHandler* handler)
{
	this->_contextMapper->addContext(id, handler);

}

int ConnectionImpl::getConnectionId()
{
	return _connectionId;
}



void ConnectionImpl::disconnect() 
{
	int rc = -1; 

	if(!_isComplete) 
	{				
		try 
		{
			// Send DISCONNECT for data connections
			std::map<int, DataConnection*>::iterator  mapIterator = _dataConnectionMap.begin();
			for (mapIterator = _dataConnectionMap.begin(); mapIterator != _dataConnectionMap.end(); mapIterator++)
			{
				int dataConnectionID = -1;
				dataConnectionID = (int) mapIterator->first;				
				if (dataConnectionID != -1)
				{
					// Close the data connection and stop the TCP data server
					rc = this->closeDataConnection(dataConnectionID);
					if (rc < 0)
					{
						// Report Error and continue (TODO - need to handle this error in a better way
						TPTP_LOG_ERROR_MSG1("Error: Unable to destroy the data connection with ID - %d", dataConnectionID);
					}
				}				
			}
			
			if (rc == 0)
			{
				// TODO - remove only the connections that were closed succesfully
				_dataConnectionMap.clear();
			}

			// Send DISCONNECT command for control channel
			rc = sendControlCommand(sock, DISCONNECT);
			if (rc < 0)
			{
				// Report Error
				TPTP_LOG_ERROR_MSG1("Error: The DISCONNECT command failed on connection with id - %d", this->_connectionId);
			}

			_isComplete=true; // Close the Socketreader thread

			rc = closeThisSocket(sock);
			if (rc < 0)
			{
				// Report Error
				TPTP_LOG_ERROR_MSG1("Error: Unable to close the socket with id - %d", sock);
			}							
		}
		catch(exception e) 
		{

		}
	}
}

INode* ConnectionImpl::getNode() {
	return _node;
}

ContextMapper* ConnectionImpl::getContextMapper() {
	return _contextMapper;
}

bool ConnectionImpl::isActive() {
	if(_isInitialized) {
		return !_isComplete;
	}
	return false;
}


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



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

	rc = initForSocketCalls() ;
	if (rc == 0)
	{
		hpLocal = getHostInfoIPv4M() ; // TODO replace with portable impl

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

	return result;
}


/**
 *********************************************************
 *                                                       
 * @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 ConnectionImpl::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);

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

    }

	if(i == 0) {
		tptp_free(result);
		result = NULL;
	}

	*outResult = result;
	*outNumResults = i;

}





unsigned long ConnectionImpl::getIPAddressIPv4(char* hostname) {
	struct hostent *hpHost = NULL;
	unsigned long   result = 0 ;
	int    rc = 0 ;

	rc = initForSocketCalls() ;
	if (rc == 0)
	{
		hpHost = getTargetHostInfoIPv4M(hostname) ;

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

	return result;
}


int ConnectionImpl::connect(INode* node, int port)
{
	int rc = 0;

	_node = node;
	_port = port;
	/* go make connection */
	init();

	initForSocketCalls();

	rc = connectToTCPServer(_node->getName(), port, &sock) ;

	if (rc != 0)
	{
		TPTP_LOG_DEBUG_MSG1("Error: unable to connect to the server. Error Code - %d", rc);
	}
	else
	{
		/* CONNECT command */
		rc = sendControlCommand(sock, CONNECT);
	}

	return rc;

}


int ConnectionImpl::connect(class INode* node, ConnectionInfo* connInfo)
{
	int rc = 0;

	//TODO: This method should just call the connect() above passing in the connInfo port.
	//But why is host name here pulled from connInfo?  What would happen if the "node"
	//parameter had a different  host name, from the one used to make the connection?
	_node = node;
	_port = connInfo->getPortNum();
	/* go make connection */
	init();
	initForSocketCalls();
	rc = connectToTCPServer(connInfo->getHostName(), _port, &sock) ;

	if (rc != 0)
	{
		TPTP_LOG_DEBUG_MSG1("Error: unable to connect to the server. Error Code - %d", rc);
	}
	else
	{
		/* CONNECT command */
		rc = sendControlCommand(sock, CONNECT);
	}

	return rc;
}


void ConnectionImpl::sendMessage(ControlMessage* message, ICommandHandler* handler)
{
	int rc = 0 ;

	char buffer[1024] ;
	int  bufferLength = 0 ;
	int  bytesSent = 0 ;

	int commandCount = message->getCommandCount();

	/*
	 * Need to store the context for rerouting on return, lock is acquired so context
	 * static variable can be incremented safely, synchronized surrounds loop instead
	 * of inside loop to minimize acquisition of locks within loop.
	 *
	 * The part of the loop that did not deal with incrementing and storing the context
	 * has been extracted out into another loop as to minimize time of lock.
	 */
	//TODO: Add locking code here
		for(int i=0; i < commandCount; i++) 
		{			
			long cmdContextId = message->getCommand(i)->getContext();
			TPTP_LOG_DEBUG_MSG1("Setting the context: %ld", cmdContextId);
			this->_contextMapper->addContext(cmdContextId, handler);
		}

	//TODO
	TPTP_LOG_DEBUG_MSG1("Setting the context: %ld", _contextId);

	/* build the command string */
	message->writeToBuffer((unsigned char*)buffer, 0);

	/* sent a message to the agent controller (server) */
	bufferLength = 12; //Message Header length
	for (unsigned int cmdCount=0;cmdCount<message->getCommandCount();cmdCount++)
	{
		bufferLength = bufferLength + message->_entries[cmdCount]->getSize();
	}
	bytesSent = writeToSocket(sock, buffer, bufferLength);
	switch (bytesSent)
	{
		case 0: TPTP_LOG_DEBUG_MSG("Socket is closed gracefully..."); break ;
		case -1: TPTP_LOG_ERROR_MSG("Error: unable to send data to the socket."); break ;
		default:
			TPTP_LOG_DEBUG_MSG1("Sent out(%d bytes). ", bytesSent) ;
			printThisEnvelope((tptp_basic_msg_header_ptr_t) buffer) ;
			rc = bytesSent ;
	}
}


void ConnectionImpl::sendMessage(TPTPCommand* command, ICommandHandler* handler)
{
	char *buffer = NULL;
	int  bufferLength = 0;
	int  bytesSent = 0;
	unsigned int flags = 0;

	/*
	 * Need to store the context for rerouting on return, lock is acquired so context
	 * static variable can be incremented safely, synchronized surrounds loop instead
	 * of inside loop to minimize acquisition of locks within loop.
	 *
	 * The part of the loop that did not deal with incrementing and storing the context
	 * has been extracted out into another loop as to minimize time of lock.
	 */
	//TODO: Add locking code here
	long cmdContextId = command->getContextID();
	TPTP_LOG_DEBUG_MSG1("Setting the context: %ld", cmdContextId);
	this->_contextMapper->addContext(cmdContextId, handler);

	char* message = NULL;
	#ifdef MVS
		char * zosCmdStr = NULL;
		zosCmdStr = command->buildCommandString();
		native2unicode(&message, zosCmdStr, strlen(zosCmdStr));
	#else
		message = command->buildCommandString();
	#endif
	
	addBasicMsgHeader(message, strlen(message)+1, &buffer, &bufferLength, flags) ;

	/* send a message to the agent controller (server) */
	bytesSent = writeToSocket(sock, buffer, bufferLength);
	switch (bytesSent)
	{
		case 0: TPTP_LOG_DEBUG_MSG("Socket is closed gracefully..."); break ;
		case -1: TPTP_LOG_ERROR_MSG("Error: unable to send data to the socket."); break ;
		default:
			TPTP_LOG_DEBUG_MSG1("Sent out(%d bytes). ", bytesSent) ;
			printThisEnvelope((tptp_basic_msg_header_ptr_t) buffer) ;
	}

	if (buffer) tptp_free(buffer);

}


/**
 *********************************************************
 *                                                       
 * @brief                                                
 *    send in the CONNECT_DATA command
 *
 * @return
 *    0 if success
 *    nonzero if error
 *
 *********************************************************/
int ConnectionImpl::sendCONNECT_DATACommand(SOCKET datasock, int direction, int commandflags) 
{
	int rc = 0 ;

	char *buffer = NULL;
	int  bufferLength = 0;
	int  bytesSent = 0;
	char *cmd = NULL;
	int  cmdLength = 0;

	/* Connect Data command contains: dataPathType */
	cmdLength = sizeof(int);
	cmd = (char*)tptp_malloc(cmdLength);

	int dataPathType = direction;
	char* pCurr = cmd ;
	pCurr = (char*)writeUINTToBuffer((unsigned char*)pCurr, dataPathType);

	/* Add the flags and message header to the command string */
	addBasicMsgHeader(cmd, cmdLength, &buffer, &bufferLength, (unsigned int)commandflags) ;

	/* sent a message to the agent controller (server) */
	bytesSent = writeToSocket(datasock, buffer, bufferLength);
	switch (bytesSent)
	{
		case 0: TPTP_LOG_DEBUG_MSG("Socket is closed gracefully...") ; break ;
		case -1: TPTP_LOG_DEBUG_MSG("Error: unable to send data to the socket. "); break ;
		default:
			TPTP_LOG_DEBUG_MSG1("Sent out(%d bytes). ", bytesSent) ;
			printThisEnvelope((tptp_basic_msg_header_ptr_t) buffer) ;
			rc = bytesSent ;
	}

	if (cmd) tptp_free(cmd);
	if (buffer) tptp_free(buffer);

	return ( rc ) ;
}


/**
 *********************************************************
 *                                                       
 * @brief                                                
 *    send command
 *
 * @return
 *    0 if success
 *    nonzero if error
 *
 *********************************************************/
int ConnectionImpl::sendControlCommand(SOCKET sockconn, int commandflags) 
{
	int rc = 0;

	char *buffer = NULL;
	int  bufferLength = 0;
	int  bytesSent = 0;
	
	/* build the CONNECT command string - contains no data, just a flag */
	addBasicMsgHeader(NULL, 0, &buffer, &bufferLength, (unsigned int)commandflags) ;

	/* sent a message to the agent controller (server) */
	bytesSent = writeToSocket(sockconn, buffer, bufferLength);
	switch (bytesSent)
	{
		case 0: TPTP_LOG_DEBUG_MSG("Socket is closed gracefully...") ; break ;
		case -1: TPTP_LOG_DEBUG_MSG("Error: unable to send data to the socket. "); break ;
		default:
			TPTP_LOG_DEBUG_MSG1("Sent out(%d bytes). ", bytesSent) ;
			printThisEnvelope((tptp_basic_msg_header_ptr_t) buffer) ;
			rc = 0 ;
	}

	if (buffer) tptp_free(buffer);
	return ( rc ) ;
}


int ConnectionImpl::createDataConnection(int direction, int flags)
{
	int rc = 0;
	char buffer[1024] ;
	int  bufferLength = 1024 ;
	int  bytesRead = 0 ;
	SOCKET datasock = 0;
	int dataConnectionId;

	initForSocketCalls();
	rc = connectToTCPServer(this->_node->getName(), _port, &datasock) ;

	if (rc != 0)
	{
		TPTP_LOG_ERROR_MSG("Error: unable to connect to the server.");
	}
	else
	{
		/* CONNECT_DATA command */
		rc = sendCONNECT_DATACommand(datasock, direction, flags);
		if (rc > 0)
		{
			/* pause a little to receive the reply */
			Sleep(1000) ;
		
			do
			{
				rc = readFromSocket(datasock, 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("Received(%d bytes). ", bytesRead) ;
				dataConnectionId = ::getConnectionId(pMsg) ;
				TPTP_LOG_DEBUG_MSG1("The Data Channel ConnectionID: %d", dataConnectionId);
				
				TCPDataServer* tcpDataServer = new TCPDataServer();
				DataConnection* dataConnectionInfo = (DataConnection*)malloc(sizeof(struct DataConnection));
				dataConnectionInfo->dataSocket = datasock;
				dataConnectionInfo->dataServer = tcpDataServer;				

				std::map<int, DataConnection*>::iterator  mapIterator = _dataConnectionMap.find(dataConnectionId);
				if ((mapIterator == _dataConnectionMap.end()) && (datasock != 0))
				{
					// not found, add the given pair to the table
					_dataConnectionMap.insert(std::map<int, DataConnection*>::value_type(dataConnectionId, dataConnectionInfo));
					dataConnectionInfo->dataServer->startServer(datasock);
				}				
			}			
		}	
	}

	return dataConnectionId;
}


/*
 * Destroy Data Connection
 */
int ConnectionImpl::destroyDataConnection(int dataConnID)
{
	int rc = 0;

	rc = closeDataConnection(dataConnID);

	if (rc == 0)
	{
		//Remove the data connection from the collection
		_dataConnectionMap.erase(dataConnID);
	}

	return rc;
}

int ConnectionImpl::closeDataConnection(int dataConnID)
{
	int rc = -1;
	DataConnection* dataConnectionInfo = (DataConnection*)_dataConnectionMap[dataConnID];
	if (dataConnectionInfo != NULL)
	{
		// Send DISCONNECT command for data channel
		rc = sendControlCommand(dataConnectionInfo->dataSocket, DISCONNECT); 

		// Stop the TCP Data Server
		dataConnectionInfo->dataServer->shutdownServer();

		rc = closeThisSocket(dataConnectionInfo->dataSocket);
		if (rc < 0)
		{
			// Report Error
			TPTP_LOG_ERROR_MSG1("Error: Unable to close the socket with id - %d", dataConnectionInfo->dataSocket);
		}		
	}

	return rc;
}


int ConnectionImpl::addDataListener(int dataConnectionId, IDataProcessor* dataProcessor)
{
	DataConnection* dataConnectionInfo = (DataConnection*)_dataConnectionMap[dataConnectionId];

	if (dataConnectionInfo != NULL)
	{
		//dataConnectionInfo->dataServer->startServer(dataProcessor, dataConnectionInfo->dataSocket);
		dataConnectionInfo->dataServer->addDataProcessor(dataProcessor);
		//dataServer->startServer(dataProcessor, (SOCKET)_dataConnectionMap[dataConnectionId]);
	}
	
	return 0;
}


int ConnectionImpl::removeDataListener(int dataConnectionId, IDataProcessor* dataProcessor)
{
	DataConnection* dataConnectionInfo = (DataConnection*)_dataConnectionMap[dataConnectionId];

	if (dataConnectionInfo != NULL)
	{
		dataConnectionInfo->dataServer->removeDataProcessor(dataProcessor);
	}
	
	return 0;
}



int ConnectionImpl::sendData(int dataConnectionID, char buffer[], int bufferLength)
{
	DataConnection* dataConnInfo = (DataConnection*)_dataConnectionMap[dataConnectionID];
	int bytesSent = writeToSocket(dataConnInfo->dataSocket, buffer, bufferLength);
	switch (bytesSent)
	{
		case 0: TPTP_LOG_DEBUG_MSG("Socket is closed gracefully...") ; break ;
		//case SOCKET_ERROR: TPTP_LOG_DEBUG_MSG("Error: unable to send data to the socket. "); break ;
		case -1: TPTP_LOG_DEBUG_MSG("Error: unable to send data to the socket. "); break ;
		default:
			TPTP_LOG_DEBUG_MSG1("Sent out(%d bytes). ", bytesSent) ;
	}

	return bytesSent;
}


int ConnectionImpl::processControlMessage(char buffer[], int length) 
{
	int rc = 0;
	unsigned int intt = 0;
	if(_cmdHandler != NULL)
	{
		ControlMessage* msg=new ControlMessage();

		try 
		{
			TPTP_LOG_DEBUG_MSG1("Processing the Response Message: %s", buffer);
			//current=msg->readFromBuffer((unsigned char*)buffer, offset);
			buffer = (char*)msg->readFromBuffer((unsigned char*)buffer, 0);
		}		
		catch(exception e) {
		    /* There was some other exception reading the message so we can't handle the message.
		       return -1 to indicate an error
		    */
			rc = -1;
		    return rc; 
		}
		
		/* Get the transport commands and handle them */
		if (msg->getMessageType() == Constants::NEW_AC_MESSAGE)
		{
			long flags = msg->getFlags();

			if ((flags & CONNECTION_COMPLETE) == CONNECTION_COMPLETE)
			{
				intt = 0;
				TPTP_LOG_DEBUG_MSG("Received a CONNECTION_COMPLETE response.");

				#if (defined(__linux__) && defined(__s390__)) || defined(_SOLARIS) || defined(_AIX) || defined(MVS)
					msg->readTPTPUINTFromBuffer((unsigned char*)buffer, (unsigned int*)&intt);
					_connectionId = intt;
				#else
					msg->readTPTPUINTFromBuffer((unsigned char*)buffer, (unsigned int*)&_connectionId);
				#endif
				
				TPTP_LOG_DEBUG_MSG1("The ConnectionId returned by AC is: %d", _connectionId);
				return rc;
			} else if ((flags & CONNECTION_RECONNECT_REQUEST))
			{
				int            ret = -1;
				int            sourceID;
				int            contextID;
				char*          interfaceName;
				char*          cmdName;
				tptp_list_t*   paramList;
				int port;

				//we are being directed to reconnect using a specified port.
				ret = parseCommand(buffer, &sourceID, &contextID, &interfaceName, &cmdName, &paramList );

				getIntegerParam("port", paramList, &port);

				TPTP_LOG_DEBUG_MSG1("Received a CONNECTION_RECONNECT_REQUEST response. %d ", port);

				//close the socket we were using for the initial connect.
				closeThisSocket(sock);

				//then try to reconnect.
				return connect(_node, port);
			}
		}

		/* Valid pass on each command */
		int count=msg->getCommandCount();
		for(int i=0; i<count; i++) 
		{
			TPTP_LOG_DEBUG_MSG1("Calling the Command Handler: %s", _cmdHandler);
			_cmdHandler->incomingCommand(_node, msg->getCommand(i));
		}

		if (msg) delete(msg);
		return rc;
	}
	return -1;
}






ConnectionImpl::SocketReaderThread::SocketReaderThread()
{
	
}



ConnectionImpl::SocketReaderThread::~SocketReaderThread()
{
	
}


void ConnectionImpl::SocketReaderThread::run()
{
		TPTP_LOG_DEBUG_MSG("Starting the Socket Reader Thread ...");
		unsigned int offset=0;
		int bytesRead = -1;
		int rc = -1;
		unsigned char* buffer;
		unsigned int   bufferLength = TPTP_DEFAULT_BUFFER_LENGTH;

		buffer = (unsigned char*)tptp_malloc( TPTP_DEFAULT_BUFFER_LENGTH+1 );
		if ( buffer == NULL )
		{
			TPTP_LOG_SEVERE_MSG1("Unable to allocate memory of size %d", TPTP_DEFAULT_BUFFER_LENGTH+1);
			return;
		}

		while(!_connImpl->_isComplete) 
		{
			bytesRead = 0 ;

			/* Another message might come in while we're processing
			 *    so we read until the pipe is empty                 */
			while ( (rc = readFromSocket(_connImpl->sock, (char*)buffer+offset, bufferLength-offset, &bytesRead)) > 0)
			{
				unsigned int neededMsgLen;

				TPTP_LOG_DEBUG_MSG1("The Number of Response Bytes read: %d", bytesRead);
				rc = processRecdMsgs((char*)buffer, bufferLength, bytesRead+offset, &offset, &neededMsgLen);

				if ( rc == TPTP_CLIENT_CONTROL_CH_PARTIAL_MESSAGE )
				{
					if ( neededMsgLen > bufferLength )
					{
						unsigned char* temp = buffer;
						
						TPTP_LOG_DEBUG_MSG1("Overflow -- reallocating message buffer (%d bytes).", neededMsgLen+1 );

						buffer = (unsigned char*)tptp_realloc( buffer, neededMsgLen+1 );
						if ( buffer == NULL )
						{
							/* We can't allocate more memory, so we're going to lose this message
							 *    but we may be able to salvage the situation by emptying our
							 *    buffer.  The error scan in the process will find the start of the
							 *    next message.
							 */
							TPTP_LOG_ERROR_MSG("Unable to re-allocate message buffer.  The current message will be lost");
							buffer = temp;
							offset = 0;
						}
						else
						{
							/* 'offset' was set by processRecdMsgs */
							bufferLength = neededMsgLen;
						}
					}
				}
				else
				{
					offset = 0;
				}
			}			
		} /* end of outer while */

		/* The server is now stopping */
}

static BOOL validRACRequest(unsigned char *buffer, int length) 
{
	//Just checking if we have a RAC request.
	if(length < 4) 
	{
		return FALSE;
	}

	/* Compare against the majic number 0x82656780 */
	if (buffer[0]!=0x82 || buffer[1]!=0x65 || buffer[2]!=0x67 || buffer[3]!=0x80) 
	{
		return FALSE;
	}

	return TRUE;
}

/**
 *********************************************************
 *
 * @brief
 *    process received messages by examining the buffer.
 *    This will process one message/command at a time
 *    until the end of the received buffer.
 *
 *    If there is incomplete command, it will move to the beginning
 *    of the buffer.
 *
 * @return
 *    0 - process all messages
 *    negative - Error.
 *    positive - number of bytes left to be processed
 *               These bytes have been copied out in front of the buffer
 *
 *********************************************************/
 int ConnectionImpl::SocketReaderThread::processRecdMsgs( char*            pBuffer, 
                     unsigned int     bufferLength, 
					 unsigned int     bytesReadIn, 
					 unsigned int*    offset, 
					 unsigned int*    messageLenOut)
{

	unsigned char     *pMsg = NULL ;
	unsigned char* pBufferStart = (unsigned char*)pBuffer;
	unsigned int magicNumber = 0;
	unsigned int flags = 0 ;
	unsigned int payLoadLength = 0 ;
	unsigned int msgLength = 0 ;

	unsigned int bytesRead = (unsigned int)bytesReadIn;

	if ( ( offset == NULL ) || (messageLenOut == NULL) )
	{
		// TODO return proper error
		return -1;
	}

	while ( bytesRead > 0 )
	{		

		/* If we have enough left that constitutes a header continue processing */
		if ((bytesRead + *offset) < sizeof(tptp_basic_msg_header_t))
		{
			/* too small to determine the payload length. Go read some more */
			*offset += bytesRead ;			
			*messageLenOut = sizeof(tptp_basic_msg_header_t);
			return TPTP_CLIENT_CONTROL_CH_PARTIAL_MESSAGE;
		}

		// read in the magic number 
		pMsg = readUINTFromBuffer(pBufferStart, &magicNumber);

		// read in the flags 
		pMsg = readUINTFromBuffer(pMsg, &flags);

		// read in the payload length 
		pMsg = readUINTFromBuffer(pMsg, &payLoadLength) ;

		msgLength = payLoadLength + sizeof(tptp_basic_msg_header_t) ;

		//Check if the message that we have is a RAC message.
		//The only time this should be the case is when the CCTL has sent an ACK of our connect.
		//this should be ignored
		if (validRACRequest((unsigned char *)pBuffer, bytesRead) && (*offset == 0)) {
			TPTP_LOG_DEBUG_MSG1( "Valid RAC request of %d bytes, ignoring ACK request", bytesRead );
			return 0;
		}

		TPTP_LOG_DEBUG_MSG1( "Validated message of %d bytes", msgLength );

		/* Are we missing part of the message because of overflow in the buffer */
		if ( msgLength > bytesRead)
		{
			/* Return this and the serveRequests thread will read more for us */
			*offset = bytesRead;
			*messageLenOut = msgLength;  
			return TPTP_CLIENT_CONTROL_CH_PARTIAL_MESSAGE;
		}


		// we have at least one complete message, consume it
		_connImpl->processControlMessage(pBuffer, bytesRead);


		if ( bytesRead < msgLength )
		{
			/* This should never happen, but that's no reason to crash if it does */
			TPTP_LOG_SEVERE_MSG( "Internal error: message length is greater than bytes read in processRecdMsgs!" );
		}
		else
		{
			bytesRead = bytesRead - msgLength;
			memmove(pBuffer, pBuffer + msgLength, bytesRead);
		}		
	}

	*offset = 0;
	*messageLenOut = 0;

	return 0;
}


THREAD_USER_FUNC_RET_TYPE ConnectionImpl::SocketReaderThread::startThread(void* pRequestBlock)
{
	SocketReaderThread** socketReader;

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

	return 0;
}



