/*******************************************************************************
 * Copyright (c) 2005, 2008 Intel Corporation and others.
 * 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:
 *    Andy Kaylor, Intel - Initial Implementation
 *    Kevin O'Leary, Intel - Extended Implementation
 *    IBM - Large portions of processing code moved from older RAC (by Andy Kaylor)
 *
 * $Id: CCTLServer.c,v 1.31 2008/06/16 20:58:03 jkubasta Exp $
 *
 *******************************************************************************/ 

#include "tptp/TPTPTypes.h"
#include "tptp/TPTPSupportTypes.h"
#include "tptp/TransportSupport.h"
#include "tptp/TPTPBaseTL.h"
#include "tptp/BaseTLLog.h"
#include "ClientCTL.h"
#include "RACmdHandlers.h"
#include "RACClientSupport.h"

static BOOL validateRecvBuffer(unsigned char *buffer, int length);
static BOOL validACRequest(unsigned char *buffer, int length);
static tptp_uint32 getMessageLengthfromValidBuffer(unsigned char *buffer);
static tptp_uint32 errorRecoveryScan(unsigned char *buffer, int length);

THREAD_USER_FUNC_RET_TYPE processJavaClientRequest(LPVOID args);
static int readJavaConnection(tl_state_data_t* stateData, jobject clientHandlerImplObj, unsigned char *buffer, int length, unsigned int *bytesRead);
static void closeJavaConnection(client_connection_block_t* ccb);
extern jmethodID getmthdid(JNIEnv *env, jclass clsid, char *mthd, char *sig);


int acPortRequestReply(client_connection_block_t* ccb)
{
	char          portRequestFormat[] = "<Cmd src=\"%d\" dest=\"%d\" ctxt=\"%d\"><portRequest iid=\"org.eclipse.tptp.client\"><port>%lu</port></portRequest></Cmd>";
	char portRequest[1024];
	cctl_state_data_t*  cctlData;
	tptp_int32 rc;
	tptp_int32 bufferLength;
	char *buffer = NULL;

	cctlData = (cctl_state_data_t*)ccb->stateData->implData;

	sprintf(portRequest, portRequestFormat, 1, 0, 0, cctlData->acProtocolPort);
	addBasicMsgHeader(portRequest, strlen(portRequest) +1, &buffer, &bufferLength, CONNECTION_RECONNECT_REQUEST);

	rc = writeToSocket(ccb->sock, buffer, bufferLength);

	if (buffer) {
		tptp_free(buffer);
	}

	return 0;
}

/**
 *********************************************************
 *
 * @brief
 *    process a given complete message 
 *
 *********************************************************/

BOOL processMessage(client_connection_block_t* ccb, ra_message_t *message) 
{
	ra_command_list_node_t* currentCommandNode;
	ra_command_t*           command;
	ra_message_t *replyMessage; /* reply */
	ra_command_t *replyCommand; /* reply */
	BOOL                    rc = FALSE;
	cctl_state_data_t*  cctlData;

	cctlData = (cctl_state_data_t*)ccb->stateData->implData;

	currentCommandNode = message->commands.head;

	/* If there are no command we abort..should never happen */
	if ( !currentCommandNode ) 
	{
		TPTP_LOG_DEBUG_MSG(ccb->stateData, "processMessage: There is no command in the message");
		return TRUE;
	}

	while (currentCommandNode!=NULL) 
	{
		command = currentCommandNode->command;

		if(!ccb->authenticated && ccb->secured) 
		{
			TPTP_LOG_DEBUG_MSG(ccb->stateData, "processMessage: Client not yet authenticated");
			switch ( command->tag )
			{
				case RA_AUTHENTICATE:
					TPTP_LOG_DEBUG_MSG(ccb->stateData, "processMessage: Authenticating client");
					rc = handleAuthenticate( ccb, command );
					break;
				default:
					TPTP_LOG_DEBUG_MSG(ccb->stateData, "processMessage: Requesting client to authenticate");
					replyMessage = ra_createMessage(RA_CONTROL_MESSAGE, 0);
					replyCommand = ra_addCommandToMessage(replyMessage, NULL);
					replyCommand->tag = RA_AUTHENTICATION_FAILED;
					replyCommand->info.authenticate_failed.ticket = 1;
					ra_forwardMessage(replyMessage, ccb);
					ra_destroyMessage(replyMessage, FALSE);
					break;
			}
		}
		else 
		{
			switch ( command->tag )
			{
				case RA_LAUNCH_PROCESS:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_LAUNCH_PROCESS" );
					rc = handleLaunchProcess( ccb, command );
					break;
				case RA_AUTHENTICATE:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_AUTHENTICATE" );
					rc = handleAuthenticate( ccb, command );
					break;
				case RA_KILL_PROCESS:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_KILL_PROCESS" );
					rc = handleKillProcess( ccb, command );
					break;
				case RA_QUERY_PROCESS_LIST:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_QUERY_PROCESS_LIST" );
					rc = handleQueryProcessList( ccb, command );
					break;
				case RA_QUERY_AGENT_LIST:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_QUERY_AGENT_LIST" );
					rc = handleQueryAgentList( ccb, command );
					break;
				case RA_QUERY_AGENT_DETAILS:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_QUERY_AGENT_DETAILS" );
					rc = handleQueryAgentDetails( ccb, command );
					break;
				case RA_AGENT_QUERY_STATE:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_AGENT_QUERY_STATE" );
					rc = handleQueryAgentState( ccb, command );
					break;
				case RA_REGISTER_AGENT_NOTIFICATION:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_REGISTER_AGENT_NOTIFICATION" );
					rc = handleRegisterAgentNotification( ccb, command );
					break;
				case RA_ATTACH_TO_AGENT:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_ATTACH_TO_AGENT" );
					rc = handleAttachToAgent( ccb, command );
					break;
				case RA_DETACH_FROM_AGENT:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_DETACH_FROM_AGENT" );
					rc = handleDetachFromAgent( ccb, command );
					break;
				case RA_START_MONITORING_AGENT_REMOTE:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_START_MONITORING_AGENT_REMOTE" );
					rc = handleStartMonitoring( ccb, command );
					break;
				case RA_STOP_MONITORING_AGENT:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_STOP_MONITORING_AGENT" );
					rc = handleStopMonitoring( ccb, command );
					break;
				case RA_MANAGE_FILE:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_MANAGE_FILE" );
					rc = handleManageFile( ccb, command );
					break;
				case RA_GET_PROPERTY_LIST:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_GET_PROPERTY_LIST" );
					rc = handleGetPropertyList( ccb, command );
					break;
				case RA_SET_NAME_VALUE_PAIR:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_SET_NAME_VALUE_PAIR" );
					rc = forwardCommandToAgent( ccb, command );
					break;
				case RA_CUSTOM_COMMAND:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_CUSTOM_COMMAND" );
					rc = forwardCommandToAgent( ccb, command );
					break;
				case RA_BINARY_CUSTOM_COMMAND:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_BINARY_CUSTOM_COMMAND" );
					rc = forwardCommandToAgent( ccb, command );
					break;
				case RA_CONTROLLER_REQUEST_MONITOR:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_CONTROLLER_REQUEST_MONITOR" );
					rc = handlePeerRequestMonitor( ccb, command );
					break;
				case RA_CONTROLLER_REQUEST_MONITOR_PORT:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_CONTROLLER_REQUEST_MONITOR_PORT" );
					rc = handlePeerRequestMonitor( ccb, command );
					break;
				case RA_PEER_UNREACHABLE:
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Processing RA_PEER_UNREACHABLE" );
					rc = handlePeerUnreachable( ccb, command );
					break;
				case RA_AUTHENTICATION_SUCCESSFUL:
					/* This is a response we get when we do a peer monitoring connection */
					/* We don't need to do anything with it */
					TPTP_LOG_DEBUG_MSG( ccb->stateData, "Received RA_AUTHENTICATION_SUCCESSFUL" );
					rc = FALSE;
					break;
				case RA_AUTHENTICATION_FAILED:
					/* This means we requested peer monitoring from an AC that has security enabled */
					/* We're not prepared to handle this case */
					TPTP_LOG_WARNING_MSG( ccb->stateData, "Received RA_AUTHENTICATION_FAILED" );
					rc = FALSE;
					break;
				case RA_SERVER_SECURITY_REQUIREMENTS:
					/* This means we requested peer monitoring from an AC that has security enabled */
					/* We're not prepared to handle this case */
					TPTP_LOG_WARNING_MSG( ccb->stateData, "Received RA_SERVER_SECURITY_REQUIREMENTS" );
					rc = FALSE;
					break;
				default:
					TPTP_LOG_WARNING_MSG( ccb->stateData, "Client unknown or unexpected command" );
					rc = FALSE;
					break;
			}
		}
		currentCommandNode = currentCommandNode->next;
	}

	return rc;
}

/**
 *********************************************************
 *
 * @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 processRecdMsgs(client_connection_block_t* ccb,
                    char*                      pBuffer, 
					unsigned int               bytesReadIn, 
					unsigned int*              offset, 
					unsigned int*              messageLenOut)
{
	unsigned int bytesRead = (unsigned int)bytesReadIn;
	unsigned int bytesToByProcessed = (unsigned int)bytesReadIn;

	if ( ( offset == NULL ) || (messageLenOut == NULL) )
	{
		return CCTL_PRM_ERROR_BAD_ARG;
	}

	while ( bytesRead > 0 )
	{
		tptp_uint32         messageLength;
		ra_message_t*       message;
		cctl_state_data_t*  cctlData = (cctl_state_data_t*)ccb->stateData->implData;

		// when we get a request from the AC as indicated by magic number
		// we know that the client is requesting the port
		// number that it should do a connect on.

		if (validACRequest(pBuffer, bytesRead))
		{
			acPortRequestReply(ccb);
			return 0;
		}

		if ( !validateRecvBuffer( pBuffer, bytesRead ) )
		{
			bytesRead = errorRecoveryScan( pBuffer, bytesRead );

			/* If we have enough left that constitutes a header continue processing */
			if( bytesRead >= 24) 
			{
				continue;
			}
			else 
			{
				/* We need to read some more data.  Keep what we have so far */
				*offset = bytesRead;
				*messageLenOut = 24;  /* At least 24, we need that to know how long it is */
				return CCTL_PRM_PARTIAL_MESSAGE;
			}
		}

		messageLength = getMessageLengthfromValidBuffer( pBuffer );

		TPTP_LOG_DEBUG_MSG1( ccb->stateData, "Validated message of %d bytes", messageLength );

		/* Are we missing part of the message because of overflow in the buffer */
		if ( messageLength > bytesRead)
		{
			/* Return this and the serveRequests thread will read more for us */
			*offset = bytesRead;
			*messageLenOut = messageLength;  /* At least 24, we need that to know how long it is */
			return CCTL_PRM_PARTIAL_MESSAGE;
		}

		message = ra_readMessageFromBuffer( pBuffer, (unsigned long)bytesRead );
		ra_mutexEnter( &cctlData->processList->lock );

		TPTP_LOG_DEBUG_MSG1( ccb->stateData, "Processing message of length %d bytes.", message->length );

		processMessage( ccb, message );

		ra_mutexExit( &cctlData->processList->lock );

		/* For whatever reason, the message length is wrong before this call */
		ra_determineMessageLength(message);

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

	*offset = 0;
	*messageLenOut = 0;

	return 0;
}


/**
 *********************************************************
 *
 * @brief
 *    receive and process incoming request message
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/

THREAD_USER_FUNC_RET_TYPE processClientRequest(LPVOID args) 
{
	int    rc = 1;

	SOCKET clientSock ;

	unsigned int   bytesRead;
	unsigned char* buffer;
	unsigned int   bufferLength = TPTP_DEFAULT_BUFFER_LENGTH;
	int            offset = 0;
	removeConnectionEntry_ptr_t removeConnection = NULL;

	/* set up environmental info for this incoming message */
	client_connection_block_t* ccb = (client_connection_block_t*)args;
	clientSock = ccb->sock ;

	buffer = (unsigned char*)tptp_malloc( TPTP_DEFAULT_BUFFER_LENGTH+1 );
	if ( buffer == NULL )
	{
		return TPTP_SYS_NO_MEM;
	}

	/* initial status before the thread is running */
	ccb->processingThreadState = TPTP_TL_THREAD_RUNNING;

	/* accept and process one connection at a time */
	while ((ccb->processingThreadState == TPTP_TL_THREAD_RUNNING) && (rc > 0))
	{
		bytesRead = 0;

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

			rc = processRecdMsgs(ccb, buffer, bytesRead+offset, &offset, &neededMsgLen);

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

					buffer = 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(ccb->stateData, "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;
			}
		}
	}

	/* Remove this connection from our table */
	/* kevin
	baseTL_removeControlConnectionEntry( ccb->stateData, ccb->clientID );
	*/

	/* Tell the AC that this connection has ended */
	/* kevin
	removeConnection = ccb->stateData->ac.removeConnection;
	if ( removeConnection != NULL )
	{
		removeConnection( ccb->stateData->ac.cmo, ccb->clientID );
	}
	*/

	/* Free our memory (TODO:?) */
//	freeClientConnectionBlock( ccb );

	return ( 0 ) ;
}

/**
 *********************************************************
 *
 * @brief
 *    main running thread that accepts connection,
 *    set up the environment and process it in the current thread
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/

THREAD_USER_FUNC_RET_TYPE serveRequests(LPVOID args) 
{
	SOCKET              clientSock;
	int                 rc = 0;
	TID                 threadId;
	HANDLE              threadHandle;
	tl_state_data_t*    stateData = (tl_state_data_t*)args;
	cctl_state_data_t*  cctlData  = (cctl_state_data_t*)stateData->implData;

	/* initial status before the thread is running */
	stateData->processingThreadState = TPTP_TL_THREAD_RUNNING;

	/* accept and process one connection at a time */
	while (stateData->processingThreadState == TPTP_TL_THREAD_RUNNING)
	{
        // clientSock = acceptSocketConnection(cctlData->sock);
		clientSock = acceptSocketConnections(cctlData->sockets, cctlData->numSockets, &(cctlData->sockAccept));

		if (isSocketValid(clientSock) == 0)
		{
			TPTP_LOG_ERROR_MSG(stateData, "Accept() received an invalid socket request." );
		}
		else
		{
			client_connection_block_t* clientConnectionBlock;
			tptp_uint32                connectionID;
			ra_message_t*              message;
			ra_command_t*              command;
			char *ipAddr;
			tptp_int32 rc;
			struct sockaddr_storage *sockAddr;
			
			sockAddr = getPeerAddrOfSocket(clientSock);

			rc = getPeerIPString(clientSock, &ipAddr);
			if (rc != 0)
			{
				TPTP_LOG_ERROR_MSG(stateData, "Connection refused could not getPeerIPString");
				continue;
			}
			
			if (!tptp_checkHostIPv6(cctlData->network_list, sockAddr)) {

				TPTP_LOG_ERROR_MSG2(stateData, "Connection refused on socket %d on host %s", clientSock, ipAddr);
				//printf("Connection refused on socket %d on host %s", clientSock, ipAddr);
				tptp_free(ipAddr);
				tptp_free(sockAddr);
				closeThisSocket(clientSock);
				continue;
			}

			tptp_free(ipAddr);
			tptp_free(sockAddr);

			setHandleInherited((HANDLE) clientSock) ;

			/* set up the data block for each request */
			clientConnectionBlock = createClientConnectionBlock(clientSock, stateData);
			clientConnectionBlock->connectionType = CCTL_NATIVE_SOCKET;

			/*
			*	Tell the client to authenticate with the secured server instead
			*
			*/
			if(cctlData->securityEnabled) {					// send redirect to secure server for RAC clients 
				TPTP_LOG_INFO_MSG1(stateData, "Security is enabled. Notify client to authenticate and drop this connection and reconnect to port %d", cctlData->securedPort);
				message = ra_createMessage(RA_CONTROL_MESSAGE, 0);
				command = ra_addCommandToMessage(message, NULL);
				command->tag = RA_SERVER_SECURITY_REQUIREMENTS;
				command->info.serverSecurityRequirements.securePort = cctlData->securedPort;
				ra_forwardMessage(message, clientConnectionBlock);
				ra_destroyMessage(message, FALSE);
				
				acPortRequestReply(clientConnectionBlock);	// send redirect to AC clients
				
				freeClientConnectionBlock(clientConnectionBlock);
			}
			/*
			*	This is not a secured connection. Send sucessful message so that client can proceed
			*
			*/
			else {
				/* Ask the AC for a new connection */
				rc = stateData->ac.addConnection( stateData->ac.cmo, stateData->transportID, &connectionID );
				if ( rc != 0 )
				{
					TPTP_LOG_ERROR_MSG1( stateData, "Error adding Agent Controller connection for socket: %u", (unsigned int)clientSock );
					return rc;
				}

				/* Add this connection to a list maintained by the base TL implementation */
				baseTL_addControlConnectionEntry( stateData, connectionID, (void*)clientConnectionBlock );

				/* Save this ID for later use */
				clientConnectionBlock->clientID = connectionID;
				clientConnectionBlock->connectionType = CCTL_NATIVE_SOCKET;
				clientConnectionBlock->secured = FALSE;

				/* Bug 85075 */
				/* Notify the client that we are insecure and ready for work */
	//			ra_logServiceMessage(__FILE__, __LINE__, RA_WARNING, "Notify the client that we are ready now since security is not enabled");
				message = ra_createMessage(RA_CONTROL_MESSAGE, 0);
				command = ra_addCommandToMessage(message, NULL);
				command->tag = RA_AUTHENTICATION_SUCCESSFUL;
				ra_createRASTRING(&command->info.authenticate_successful.key, "none");
				ra_forwardMessage(message, clientConnectionBlock);
				ra_destroyMessage(message, FALSE);
				/* Bug 85075 */

				/* go create a new thread to process each incoming connection request */
				rc = tptpStartThread(processClientRequest, 
					(LPVOID)clientConnectionBlock, &threadId, &threadHandle) ;
				CLOSE_THREAD_HANDLE(threadHandle);
			}
		}
	}

	return ( rc ) ;
}

/**
 *********************************************************
 *
 * @brief
 *    utility function that creates and initializes a 
 *    block of information to be associated with a client 
 *    connection
 *
 * @return
 *    Pointer to the client data block 
 *    NULL if error
 *********************************************************/

client_connection_block_t* createClientConnectionBlock(SOCKET clientSock, tl_state_data_t* stateData)
{
	client_connection_block_t* clientConnectionBlock = NULL ;

	clientConnectionBlock = (client_connection_block_t*)tptp_malloc(sizeof(client_connection_block_t)) ;
	clientConnectionBlock->sock = clientSock;
	clientConnectionBlock->stateData = stateData;
	clientConnectionBlock->clientID = 0; // Unknown at this point
	clientConnectionBlock->processingThreadState = TPTP_TL_THREAD_NOT_STARTED;
	tptp_list_init( &clientConnectionBlock->agents );

	return clientConnectionBlock;
}

/**
 *********************************************************
 *
 * @brief
 *    free the request block for this socket connection
 *
 *********************************************************/

void  freeClientConnectionBlock(client_connection_block_t* ccb)
{
	tptp_list_clear( &ccb->agents );
	tptp_free( ccb );
}

/** VALIDATE_RECV_BUFFER  *****************************************************
  * This is a simple test to ensure a message starts with the majic number
  * and that it at least contains enough information that we can determine
  * the message length.
  */
static BOOL validateRecvBuffer(unsigned char *buffer, int length) 
{
	/* Do we have a full header? */
	if(length<24) 
	{
		return FALSE;
	}

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

	return TRUE;
}

static BOOL validACRequest(unsigned char *buffer, int length) 
{
	/* Do we have a full header? */
	/* In my testing this must be 12... not greater... but the implementation
	 * could change */
	if(length < 12) 
	{
		return FALSE;
	}

	/* Compare against the magic number of the AC which is 0x54b674de */
	if (buffer[0]!=0x54 || buffer[1]!=0xb6 || buffer[2]!=0x74 || buffer[3]!=0xde) 
	{
		return FALSE;
	}

	return TRUE;
}


/** GET_MESSAGE_LENGTH_FROM_VALID_BUFFER  *************************************
  * This simply extracts the message length from a message buffer that has
  * previously been validated successfuly.
  */
static tptp_uint32 getMessageLengthfromValidBuffer(unsigned char *buffer) 
{
	return ((buffer[16]<<24)
					  |(tptp_uint32)(buffer[17]<<16)
					  |(tptp_uint32)(buffer[18]<<8)
					  | buffer[19]);
}

/** ERROR_RECOVERY_SCAN  *****************************************************
  * When we get a bad message flow we go into recovery mode, searching for the
  * next occurance of the magic number in the stream.  The buffer is then
  * compressed and number of valid remaining bytes is returned.
  */
static tptp_uint32 errorRecoveryScan(unsigned char *buffer, int length) 
{
	int offset;

	/* If there aren't enough bytes to check the magic number return zero */
	if ( length < 4 ) 
	{
		return length;
	}

	/* Search for the next occurance of the magic number */
	for ( offset = 0; offset < length - 3; offset++ ) 
	{
		if ( buffer[offset] == 0x82 && buffer[offset+1] == 0x65 && buffer[offset+2] == 0x67 && buffer[offset+3] == 0x80) 
		{
			memmove( buffer, buffer+offset, length-offset );
			return length-offset;
		}
	}

	/* If we didn't find the magic number we need to save the last 3 bytes and return */
	memmove( buffer, buffer+offset, length - offset );
	return length - offset;
}

tptp_int32 determineCmdContext(client_connection_block_t* ccb, ra_command_t *cmd) 
{
	return cmd->tag;
}


/**
 *********************************************************
 *
 * @brief
 *    receive and process incoming request message from Java
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/

THREAD_USER_FUNC_RET_TYPE processJavaClientRequest(LPVOID args) 
{
	int    rc = 1;

	/***** SOCKET clientSock ; *****/

	unsigned int   bytesRead;
	unsigned char* buffer;
	unsigned int   bufferLength = TPTP_DEFAULT_BUFFER_LENGTH;
	int            offset = 0;
	removeConnectionEntry_ptr_t removeConnection = NULL;
	jobject				clientHandlerImplObj;

	/* set up environmental info for this incoming message */
	client_connection_block_t* ccb = (client_connection_block_t*)args;
	clientHandlerImplObj = ccb->javaObj;
	/***** clientSock = ccb->sock ; *****/

	buffer = (unsigned char*)tptp_malloc( TPTP_DEFAULT_BUFFER_LENGTH+1 );
	if ( buffer == NULL )
	{
		return TPTP_SYS_NO_MEM;
	}

	/* initial status before the thread is running */
	ccb->processingThreadState = TPTP_TL_THREAD_RUNNING;

	if(ccb->secured) {
		ra_message_t *message;
		ra_command_t *command;

		ccb->authenticated = FALSE;
		TPTP_LOG_INFO_MSG(ccb->stateData, "Asking client to authenticate");

		message = ra_createMessage(RA_CONTROL_MESSAGE, 0);
		command = ra_addCommandToMessage(message, NULL);
		command->tag = RA_AUTHENTICATION_FAILED;
		command->info.authenticate_failed.ticket = 0;
		ra_forwardMessage(message, ccb);
		ra_destroyMessage(message, FALSE);
	}

	/* accept and process one connection at a time */
	while ((ccb->processingThreadState == TPTP_TL_THREAD_RUNNING) && (rc > 0))
	{
		bytesRead = 0;

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

			TPTP_LOG_DEBUG_MSG1(ccb->stateData, "Socket processJavaClientRequest: Read %d bytes.", bytesRead) ;

			rc = processRecdMsgs(ccb, buffer, bytesRead+offset, &offset, &neededMsgLen);

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

					buffer = 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(ccb->stateData, "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;
			}
		}
	}

	/* Remove this connection from our table */
	/* kevin
	baseTL_removeControlConnectionEntry( ccb->stateData, ccb->clientID );
	*/

	/* Tell the AC that this connection has ended */
	/* kevin
	removeConnection = ccb->stateData->ac.removeConnection;
	if ( removeConnection != NULL )
	{
		removeConnection( ccb->stateData->ac.cmo, ccb->clientID );
	}
	*/

	/* Free our memory (TODO:?) */
//	freeClientConnectionBlock( ccb );

	closeJavaConnection(ccb);

	return ( 0 ) ;
}

static int readJavaConnection(tl_state_data_t* stateData, jobject clientHandlerImplObj, unsigned char *buffer, int length, unsigned int *bytesRead) 
{
		/* Read from java socket by calling the java object read method */
		JNIEnv *env;
		jmethodID readMethod;
		jclass cls;
		jint result;

		if ( 0 == ATTACH_THREAD(env) ) {

			/* Find the class object */
			cls=ENV(env)->GetObjectClass(ENVPARM(env) clientHandlerImplObj);

			readMethod = getmthdid(env, cls, "read", "([BII)I");
			if(readMethod) {
				/* create a new byte array */
				jbyteArray jbarr;
				jbarr = ENV(env)->NewByteArray(ENVPARM(env) length-0);  /* replacing offset with 0 */
                

				/* Read the data from the java InputStream */
				result = ENV(env)->CallIntMethod(ENVPARM(env) clientHandlerImplObj, readMethod, jbarr, 0, length-0);

				if (result <= 0) {
					*bytesRead = 0;
				}
				else {
					ENV(env)->GetByteArrayRegion(ENVPARM(env) jbarr, 0, result, (signed char *)buffer+0);
					*bytesRead = result;
				}
			}
			else {
				ENV(env)->ExceptionClear(ENVPARM1(env));
				TPTP_LOG_SEVERE_MSG(stateData, "Could not get ID for method read");
				*bytesRead = 0;
				result = -1;
			}

			DETACH_THREAD();
		}
		else {
			TPTP_LOG_SEVERE_MSG(stateData, "Could not attach thread to JVM");
			*bytesRead = 0;
			result = -1;
		}
		return (int)result;
}


static void closeJavaConnection(client_connection_block_t* ccb)
{
		/* Call close method of java object */
		jmethodID closeMethod;
		jclass cls;
		JNIEnv *env;


		if ( 0 == ATTACH_THREAD(env) ) {
			/* Find the class object */
			cls=ENV(env)->GetObjectClass(ENVPARM(env) ccb->javaObj);

			closeMethod = getmthdid(env, cls, "closeConnection", "()V");
			if(closeMethod) {

				/* Close the connection */

				ENV(env)->CallVoidMethod(ENVPARM(env) ccb->javaObj, closeMethod, NULL);
			}
			else {
				ENV(env)->ExceptionClear(ENVPARM1(env));
				TPTP_LOG_SEVERE_MSG(ccb->stateData, "Could not get ID for method closeConnection");
			}

			/* Delete the global reference because we no longer need it */

			ENV(env)->DeleteGlobalRef(ENVPARM(env) ccb->javaObj);

			DETACH_THREAD();
		}
		else {
			TPTP_LOG_SEVERE_MSG(ccb->stateData, "Could not attach thread to JVM");
		}
}
