/*******************************************************************************
 * Copyright (c) 2005, 2007 Intel Corporation.
 * 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 API and Implementation
 *    IBM - Large portions of processing code moved from older RAC (by Andy Kaylor)
 *
 * $Id: ACTLServer.c,v 1.18 2007/12/07 04:04:03 jkubasta Exp $
 *
 *******************************************************************************/ 

#include "tptp/TPTPTypes.h"
#include "tptp/TPTPSupportTypes.h"
#include "tptp/TransportSupport.h"
#include "tptp/TPTPBaseTL.h"
#include "tptp/BaseTLLog.h"
#include "tptp/compatibility/CmdConv.h"
#include "AgentCTL.h"
#include "RACmdHandlers.h"
#include "RACAgentSupport.h"
#include "Connect2AC.h"

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

tptp_int32 handleShutdownRequest( tl_state_data_t* stateData );

BOOL processMessage(tl_state_data_t* stateData, ra_message_t *message) 
{
	ra_command_list_node_t* currentCommandNode;
	ra_command_t*           command;
	BOOL                    rc = FALSE;

	currentCommandNode = message->commands.head;

	/* If there are no command we abort..should never happen */
	if ( !currentCommandNode ) 
	{
		return TRUE;
	}

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

		/* Is this a new agent definition?  If so there should be only one command in this message to register the agent*/
		switch ( command->tag )
		{
			case RA_AGENT_SCOPING_INFORMATION:
				rc = handleAgentScopingCommand( stateData, command );
				break;

			case RA_AGENT_INACTIVE:
				rc = handleAgentInactive( stateData, command );
				break;

			case RA_GET_PROPERTY_LIST:
				rc = handleGetPropertyList( stateData, command );
				break;

			case RA_AGENT_REQUEST_MONITOR:
			case RA_AGENT_REQUEST_MONITOR_PORT: /* Bug 77768 */
				rc = handleAgentRequestMonitor(stateData, command );
				break;

			default:
				{
					agent_t *agent;
					process_t *process;

					/* All commands contain the same context information at the
					   same offsets in the command structure so we can locate
					   this agent and process sending the message
					*/
					process = ra_processFind(ra_getActiveProcessList(), command->info.custom_command.processId);

					if(!process) 
					{
						TPTP_LOG_WARNING_MSG1( stateData, "Could not find process %d.  The model may be missing information", (int) command->info.agent_active.processId);
						return FALSE;
					}

					agent=ra_agentFindByProcess(process, &command->info.custom_command.agent, TRUE);

					/* when agent becomes inactive, it is removed from the active list */
					if(agent == NULL) 
					{
						agent=ra_agentFindByProcess(process, &command->info.custom_command.agent, FALSE);
					}

					/* In the default case, the message may contain multiple command each of which needs to be processed */
					rc = forwardCommandsToHandlers( stateData, agent, message );

				}
				break;
		}
		currentCommandNode = currentCommandNode->next;
	}

	return rc;
}


BOOL forwardCommandsToHandlers( tl_state_data_t* stateData, agent_t *agent, ra_message_t *message )
{
	ra_command_list_node_t *current;

	/* Process each command in the message in order */
	current = message->commands.head;
	while ( current != NULL ) 
	{
		ra_command_t *command=current->command;
		TPTP_LOG_DEBUG_MSG1( stateData, "Processing command: tag=%d", (int) command->tag);
		switch(command->tag) 
		{
			case RA_CUSTOM_COMMAND:
				forwardCommandToAttachedClient(stateData, command, agent);
				break;
			case RA_BINARY_CUSTOM_COMMAND:
				forwardCommandToAttachedClient(stateData, command, agent);
				break;
			case RA_SET_NAME_VALUE_PAIR:
				forwardCommandToAttachedClient(stateData, command, agent);
				break;
			case RA_ERROR_STRING:
				processExternalErrorMessage(stateData, command, agent);
				break;
			default:
				TPTP_LOG_WARNING_MSG1( stateData, "Received unexpected commmand from agent: tag=%d", (int) command->tag);
				break;
		}
		current=current->next;
	}

	return TRUE;
}

void forwardCommandToAttachedClient( tl_state_data_t* stateData, ra_command_t *command, agent_t *agent )
{
	/* Forward to the attached client */
	if(agent->attached && agent->client) 
	{
		ra_message_t *forwardMessage;
		char*         buffer;

		forwardMessage = ra_createMessage(RA_CONTROL_MESSAGE, 0);
		ra_addCommandToMessage(forwardMessage, command);

		/* 187443 end */
		TPTP_LOG_DEBUG_MSG1( stateData, "Forwarding command to client: tag=%d", (int) command->tag);

		/* Convert this command into a format the client will recognize */
		if ( 0 != convertCommandToXML( stateData, agent->agentID, agent->client->clientID, command, &buffer ) )
		{
			/* TODO: Send an error back to the agent */
			return;
		}

		/* Send this command to the client */
		if ( 0 != forwardXmlCommand( stateData, buffer ) )
		{
			/* TODO: Send an error back to the agent */
			tptp_free( buffer );
			return;
		}

		tptp_free( buffer );

		ra_destroyMessage(forwardMessage, FALSE);
	}
	else
	{
		TPTP_LOG_WARNING_MSG1( stateData, "Agent with no attached client attempted to send custom command: tag=%d", (int) command->tag);
	}
}


/** PROCESS_EXTERNAL_ERROR_MESSAGE  *******************************************************
  *
  */
void processExternalErrorMessage( tl_state_data_t* stateData, 
                                  ra_command_t*    command,
                                  agent_t*         agent) 
{
	/* Log the message to the local log file */
#ifdef _HPUX
	TPTP_LOG_MSG4( stateData, (enum _severity)command->info.error_string.severity, "EXTERNAL: %s:%d  MESSAGE_ID=%s  MESSAGE= %s"
					, command->info.error_string.agent.data
					, command->info.error_string.processId
					, command->info.error_string.messageId.data
					, command->info.error_string.message.data);

#else
	TPTP_LOG_MSG4( stateData, command->info.error_string.severity, "EXTERNAL: %s:%d  MESSAGE_ID=%s  MESSAGE= %s"
					, command->info.error_string.agent.data
					, (int) command->info.error_string.processId
					, command->info.error_string.messageId.data
					, command->info.error_string.message.data);
#endif
	   
	/* forward the message appropriately */
	forwardCommandToAttachedClient( stateData, command, agent );
}

/**
 *********************************************************
 *
 * @brief
 *    process received messages by examining the buffer.
 *
 *    If there is an incomplete message block, it will move to the beginning
 *    of the buffer.
 *
 * @return
 *    0 - process all messages
 *    non-zero - Error.
 *
 *    If a partial message is found at the end of the buffer, the return code 
 *        will be ACTL_PRM_PARTIAL_MESSAGE and the offet parameter will give 
 *        the index of the end of the partial message and messageLenOut will
 *        give the expected length of the message
 *
 *********************************************************/

int processRecdMsgs( tl_state_data_t* stateData, 
                     char*            pBuffer, 
					 unsigned int     bytesReadIn, 
					 unsigned int*    offset, 
					 unsigned int*    messageLenOut)
{
	unsigned int bytesRead = (unsigned int)bytesReadIn;

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

	while ( bytesRead > 0 )
	{
		tptp_uint32         messageLength;
		ra_message_t*       message;
		actl_state_data_t*	actlData = (actl_state_data_t*)stateData->implData;

		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 ACTL_PRM_PARTIAL_MESSAGE;
			}
		}

		messageLength = getMessageLengthfromValidBuffer( pBuffer );

		TPTP_LOG_DEBUG_MSG1( 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 ACTL_PRM_PARTIAL_MESSAGE;
		}

		message = ra_readMessageFromBuffer( pBuffer, (unsigned long)bytesRead );

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

		/* Most messages use an agent pointer, so we'll protect the list while we're processing */
		ra_mutexEnter( &actlData->processList->lock );
		actlData->agentInUse++;
		ra_mutexExit( &actlData->processList->lock );

		processMessage( stateData, message );

		/* Signal the purge thread that it's safe to run */
		actlData->agentInUse--;
		if ( !actlData->agentInUse )
		{
			tptp_postSemaphore( &actlData->processScrubSem );
		}

		/* 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( 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
 *    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) 
{
	int      rc = 0;
	int      bytesRead = 0;
	HANDLE*  pPipe         = NULL;
	BOOL     isSuccessful  = FALSE;

	unsigned char* buffer;
	unsigned int   bufferLength = TPTP_DEFAULT_BUFFER_LENGTH;
	unsigned int   offset = 0;

	tl_state_data_t*    stateData = (tl_state_data_t*)args;
	actl_state_data_t*  actlData  = (actl_state_data_t*)stateData->implData;

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

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

	/* accept and process one connection at a time */
	while (stateData->processingThreadState == TPTP_TL_THREAD_RUNNING)
	{
		/* This call blocks until a writer opens the pipe */
		isSuccessful = waitForConnection(pPipe, stateData);

		if ( isSuccessful && (stateData->processingThreadState == TPTP_TL_THREAD_RUNNING) )
		{
			bytesRead = 0;

			/* Another message might come in while we're processing
			 *    so we read until the pipe is empty                 */
			while ( 0 < readFromNamedPipe(*pPipe, buffer, offset, bufferLength, &bytesRead) )
			{
				unsigned int neededMsgLen;

				TPTP_LOG_DEBUG_MSG1(stateData, "Received(%d bytes).", bytesRead+offset);

				/* Check for a special case */
				if(strncmp((char*)buffer, "SHUTDOWN", 8) == 0) 
				{
					TPTP_LOG_WARNING_MSG(stateData, "Agent Controller shutdown requested through ACTL" );
					handleShutdownRequest( stateData );
					offset = 0;
					continue; 
					/* AK -- I chose to continue instead of return to allow for non-default handling of the shutdown message */
				}

				rc = processRecdMsgs(stateData, buffer, bytesRead+offset, &offset, &neededMsgLen);
				if ( rc == ACTL_PRM_PARTIAL_MESSAGE )
				{
					if ( neededMsgLen > bufferLength )
					{
						unsigned char* temp = buffer;
						
						TPTP_LOG_DEBUG_MSG1(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(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;
				}
			}

			disconnectFromNamedPipe(*pPipe);

			/* Go back to the beginning of our buffer */
			offset = 0;
		}
	}

	TPTP_LOG_DEBUG_MSG(stateData, "Named pipe server has ended.");

	cleanPipeUp(pPipe);

	return 0;
}


/**
 *********************************************************
 *
 * @brief
 *    accept new request from this named pipe
 *
 * @return
 *    TRUE - Success
 *    FALSE - Error.
 *********************************************************/

int waitForConnection(HANDLE *pPipeHandle, tl_state_data_t* stateData)
{
	actl_state_data_t* actlData = (actl_state_data_t*)stateData->implData;
	int rc = -1;
	int retryCount = 0;

	TPTP_LOG_DEBUG_MSG( stateData, "Ready to accept next named pipe request." );

	/* Wait for a connection. */
	/* We will only try and open the file up to 64 times.  
	 * If we cannot access our master pipe we will exit the server
	 */
	while ( (rc != 0)  && (retryCount < MAX_RETRIES) && 
	        (stateData->processingThreadState == TPTP_TL_THREAD_RUNNING) )
	{
		rc = connectToNamedPipe(pPipeHandle, actlData->fullPipeName);

		if (rc != 0)
			retryCount++;
	}

	if (retryCount >= MAX_RETRIES)
	{
		TPTP_LOG_ERROR_MSG1(stateData, "Error: we have tried to connect this named pipe %d times.", retryCount) ;
	}

	return ( rc == 0 ) ;
}

/** 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 majic number 0x82656780 */
	if (buffer[0]!=0x82 || buffer[1]!=0x65 || buffer[2]!=0x67 || buffer[3]!=0x80) 
	{
		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;
}


/** HANDLE_SHUTDOWN_REQUEST *****************************************************
  * When "SHUTDOWN" gets written to the named pipe, we need to notify the AC
  * that a shutdown has been requested (probably by another instance of the AC).
  */
tptp_int32 handleShutdownRequest( tl_state_data_t* stateData )
{
	/* The hard-coded source and destination both map to AGENT_MANAGER */
	char shutdownCmd[] = "<Cmd src=\"1\" dest=\"1\" ctxt=\"-1\"><shutdown iid=\"org.eclipse.tptp.agentManager\" /></Cmd>";

	return forwardXmlCommand( stateData, shutdownCmd );
}
